diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPSender.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPSender.java index 0aff21d2..742b8bbb 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPSender.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPSender.java @@ -47,61 +47,65 @@ public class SIPSender { } public void transmitRequest(String ip, Message message, SipSubscribe.Event errorEvent, SipSubscribe.Event okEvent) throws SipException, ParseException { - ViaHeader viaHeader = (ViaHeader)message.getHeader(ViaHeader.NAME); - String transport = "UDP"; - if (viaHeader == null) { - logger.warn("[消息头缺失]： ViaHeader， 使用默认的UDP方式处理数据"); - }else { - transport = viaHeader.getTransport(); - } - if (message.getHeader(UserAgentHeader.NAME) == null) { - try { - message.addHeader(SipUtils.createUserAgentHeader(sipLayer.getSipFactory(), gitUtil)); - } catch (ParseException e) { - logger.error("添加UserAgentHeader失败", e); + try { + ViaHeader viaHeader = (ViaHeader)message.getHeader(ViaHeader.NAME); + String transport = "UDP"; + if (viaHeader == null) { + logger.warn("[消息头缺失]： ViaHeader， 使用默认的UDP方式处理数据"); + }else { + transport = viaHeader.getTransport(); } - } - - CallIdHeader callIdHeader = (CallIdHeader) message.getHeader(CallIdHeader.NAME); - // 添加错误订阅 - if (errorEvent != null) { - sipSubscribe.addErrorSubscribe(callIdHeader.getCallId(), (eventResult -> { - errorEvent.response(eventResult); - sipSubscribe.removeErrorSubscribe(eventResult.callId); - sipSubscribe.removeOkSubscribe(eventResult.callId); - })); - } - // 添加订阅 - if (okEvent != null) { - sipSubscribe.addOkSubscribe(callIdHeader.getCallId(), eventResult -> { - okEvent.response(eventResult); - sipSubscribe.removeOkSubscribe(eventResult.callId); - sipSubscribe.removeErrorSubscribe(eventResult.callId); - }); - } - if ("TCP".equals(transport)) { - SipProviderImpl tcpSipProvider = sipLayer.getTcpSipProvider(ip); - if (tcpSipProvider == null) { - logger.error("[发送信息失败] 未找到tcp://{}的监听信息", ip); - return; - } - if (message instanceof Request) { - tcpSipProvider.sendRequest((Request)message); - }else if (message instanceof Response) { - tcpSipProvider.sendResponse((Response)message); + if (message.getHeader(UserAgentHeader.NAME) == null) { + try { + message.addHeader(SipUtils.createUserAgentHeader(sipLayer.getSipFactory(), gitUtil)); + } catch (ParseException e) { + logger.error("添加UserAgentHeader失败", e); + } } - } else if ("UDP".equals(transport)) { - SipProviderImpl sipProvider = sipLayer.getUdpSipProvider(ip); - if (sipProvider == null) { - logger.error("[发送信息失败] 未找到udp://{}的监听信息", ip); - return; + CallIdHeader callIdHeader = (CallIdHeader) message.getHeader(CallIdHeader.NAME); + // 添加错误订阅 + if (errorEvent != null) { + sipSubscribe.addErrorSubscribe(callIdHeader.getCallId(), (eventResult -> { + errorEvent.response(eventResult); + sipSubscribe.removeErrorSubscribe(eventResult.callId); + sipSubscribe.removeOkSubscribe(eventResult.callId); + })); } - if (message instanceof Request) { - sipProvider.sendRequest((Request)message); - }else if (message instanceof Response) { - sipProvider.sendResponse((Response)message); + // 添加订阅 + if (okEvent != null) { + sipSubscribe.addOkSubscribe(callIdHeader.getCallId(), eventResult -> { + okEvent.response(eventResult); + sipSubscribe.removeOkSubscribe(eventResult.callId); + sipSubscribe.removeErrorSubscribe(eventResult.callId); + }); } + if ("TCP".equals(transport)) { + SipProviderImpl tcpSipProvider = sipLayer.getTcpSipProvider(ip); + if (tcpSipProvider == null) { + logger.error("[发送信息失败] 未找到tcp://{}的监听信息", ip); + return; + } + if (message instanceof Request) { + tcpSipProvider.sendRequest((Request)message); + }else if (message instanceof Response) { + tcpSipProvider.sendResponse((Response)message); + } + + } else if ("UDP".equals(transport)) { + SipProviderImpl sipProvider = sipLayer.getUdpSipProvider(ip); + if (sipProvider == null) { + logger.error("[发送信息失败] 未找到udp://{}的监听信息", ip); + return; + } + if (message instanceof Request) { + sipProvider.sendRequest((Request)message); + }else if (message instanceof Response) { + sipProvider.sendResponse((Response)message); + } + } + } finally { + logger.info("[SEND]:SUCCESS:{}", message); } } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommander.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommander.java index 40571fe2..e9b20392 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommander.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommander.java @@ -183,7 +183,7 @@ public interface ISIPCommander { * @param channelId 预览通道 * @param recordCmdStr 录像命令：Record / StopRecord */ - void recordCmd(Device device, String channelId, String recordCmdStr, SipSubscribe.Event errorEvent) throws InvalidArgumentException, SipException, ParseException; + void recordCmd(Device device, String channelId, String recordCmdStr, SipSubscribe.Event errorEvent, SipSubscribe.Event okEvent) throws InvalidArgumentException, SipException, ParseException; /** * 远程启动控制命令 @@ -197,7 +197,7 @@ public interface ISIPCommander { * * @param device 视频设备 */ - void guardCmd(Device device, String guardCmdStr, SipSubscribe.Event errorEvent) throws InvalidArgumentException, SipException, ParseException; + void guardCmd(Device device, String guardCmdStr, SipSubscribe.Event errorEvent, SipSubscribe.Event okEvent) throws InvalidArgumentException, SipException, ParseException; /** * 报警复位命令 @@ -206,7 +206,7 @@ public interface ISIPCommander { * @param alarmMethod 报警方式（可选） * @param alarmType 报警类型（可选） */ - void alarmCmd(Device device, String alarmMethod, String alarmType, SipSubscribe.Event errorEvent) throws InvalidArgumentException, SipException, ParseException; + void alarmCmd(Device device, String alarmMethod, String alarmType, SipSubscribe.Event errorEvent, SipSubscribe.Event okEvent) throws InvalidArgumentException, SipException, ParseException; /** * 强制关键帧命令,设备收到此命令应立刻发送一个IDR帧 diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java index 47c798fb..69376416 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java @@ -664,7 +664,7 @@ public class SIPCommander implements ISIPCommander { * @param recordCmdStr 录像命令：Record / StopRecord */ @Override - public void recordCmd(Device device, String channelId, String recordCmdStr, SipSubscribe.Event errorEvent) throws InvalidArgumentException, SipException, ParseException { + public void recordCmd(Device device, String channelId, String recordCmdStr, SipSubscribe.Event errorEvent, SipSubscribe.Event okEvent) throws InvalidArgumentException, SipException, ParseException { StringBuffer cmdXml = new StringBuffer(200); String charset = device.getCharset(); cmdXml.append("\r\n"); @@ -682,7 +682,7 @@ public class SIPCommander implements ISIPCommander { Request request = headerProvider.createMessageRequest(device, cmdXml.toString(), null, SipUtils.getNewFromTag(), null,sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()),device.getTransport())); - sipSender.transmitRequest(sipLayer.getLocalIp(device.getLocalIp()), request, errorEvent); + sipSender.transmitRequest(sipLayer.getLocalIp(device.getLocalIp()), request, errorEvent,okEvent); } /** @@ -716,7 +716,7 @@ public class SIPCommander implements ISIPCommander { * @param guardCmdStr "SetGuard"/"ResetGuard" */ @Override - public void guardCmd(Device device, String guardCmdStr, SipSubscribe.Event errorEvent) throws InvalidArgumentException, SipException, ParseException { + public void guardCmd(Device device, String guardCmdStr, SipSubscribe.Event errorEvent, SipSubscribe.Event okEvent) throws InvalidArgumentException, SipException, ParseException { StringBuffer cmdXml = new StringBuffer(200); String charset = device.getCharset(); @@ -729,7 +729,7 @@ public class SIPCommander implements ISIPCommander { cmdXml.append("\r\n"); Request request = headerProvider.createMessageRequest(device, cmdXml.toString(), null, SipUtils.getNewFromTag(), null,sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()),device.getTransport())); - sipSender.transmitRequest(sipLayer.getLocalIp(device.getLocalIp()), request, errorEvent); + sipSender.transmitRequest(sipLayer.getLocalIp(device.getLocalIp()), request, errorEvent,okEvent); } /** @@ -738,7 +738,7 @@ public class SIPCommander implements ISIPCommander { * @param device 视频设备 */ @Override - public void alarmCmd(Device device, String alarmMethod, String alarmType, SipSubscribe.Event errorEvent) throws InvalidArgumentException, SipException, ParseException { + public void alarmCmd(Device device, String alarmMethod, String alarmType, SipSubscribe.Event errorEvent, SipSubscribe.Event okEvent) throws InvalidArgumentException, SipException, ParseException { StringBuffer cmdXml = new StringBuffer(200); String charset = device.getCharset(); @@ -765,7 +765,7 @@ public class SIPCommander implements ISIPCommander { Request request = headerProvider.createMessageRequest(device, cmdXml.toString(), null, SipUtils.getNewFromTag(), null,sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()),device.getTransport())); - sipSender.transmitRequest(sipLayer.getLocalIp(device.getLocalIp()), request, errorEvent); + sipSender.transmitRequest(sipLayer.getLocalIp(device.getLocalIp()), request, errorEvent,okEvent); } /** diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/control/cmd/DeviceControlQueryMessageHandler.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/control/cmd/DeviceControlQueryMessageHandler.java index 46e8e928..fa54a0fd 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/control/cmd/DeviceControlQueryMessageHandler.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/control/cmd/DeviceControlQueryMessageHandler.java @@ -131,16 +131,16 @@ public class DeviceControlQueryMessageHandler extends SIPRequestProcessorParent handleRecordCmd(deviceForPlatform,channelId,rootElement,request,DeviceControlType.RECORD); break; case I_FRAME: - handleIFameCmd(deviceForPlatform,channelId); + handleIFameCmd(deviceForPlatform,request,channelId); break; case TELE_BOOT: - handleTeleBootCmd(deviceForPlatform); + handleTeleBootCmd(deviceForPlatform,request); break; case DRAG_ZOOM_IN: - handleDragZoom(deviceForPlatform,channelId,rootElement,DeviceControlType.DRAG_ZOOM_IN); + handleDragZoom(deviceForPlatform,channelId,rootElement,request,DeviceControlType.DRAG_ZOOM_IN); break; case DRAG_ZOOM_OUT: - handleDragZoom(deviceForPlatform,channelId,rootElement,DeviceControlType.DRAG_ZOOM_OUT); + handleDragZoom(deviceForPlatform,channelId,rootElement,request,DeviceControlType.DRAG_ZOOM_OUT); break; case HOME_POSITION: handleHomePositionCmd(deviceForPlatform,channelId,rootElement,request,DeviceControlType.HOME_POSITION); @@ -167,7 +167,6 @@ public class DeviceControlQueryMessageHandler extends SIPRequestProcessorParent } catch (InvalidArgumentException | SipException | ParseException e) { logger.error("[命令发送失败] 云台/前端: {}", e.getMessage()); } - } /** @@ -175,9 +174,10 @@ public class DeviceControlQueryMessageHandler extends SIPRequestProcessorParent * @param device 设备 * @param channelId 通道id */ - private void handleIFameCmd(Device device,String channelId){ + private void handleIFameCmd(Device device,SIPRequest request,String channelId){ try { cmder.iFrameCmd(device,channelId); + responseAck(request, Response.OK); } catch (InvalidArgumentException | SipException | ParseException e) { logger.error("[命令发送失败] 强制关键帧: {}", e.getMessage()); } @@ -187,9 +187,10 @@ public class DeviceControlQueryMessageHandler extends SIPRequestProcessorParent * 处理重启命令 * @param device 设备信息 */ - private void handleTeleBootCmd(Device device){ + private void handleTeleBootCmd(Device device,SIPRequest request){ try { cmder.teleBootCmd(device); + responseAck(request, Response.OK); } catch (InvalidArgumentException | SipException | ParseException e) { logger.error("[命令发送失败] 重启: {}", e.getMessage()); } @@ -203,7 +204,7 @@ public class DeviceControlQueryMessageHandler extends SIPRequestProcessorParent * @param rootElement 根节点 * @param type 消息类型 */ - private void handleDragZoom(Device device,String channelId,Element rootElement,DeviceControlType type){ + private void handleDragZoom(Device device,String channelId,Element rootElement,SIPRequest request,DeviceControlType type){ String cmdString = getText(rootElement,type.getVal()); StringBuffer cmdXml = new StringBuffer(200); cmdXml.append("<" + type.getVal() + ">\r\n"); @@ -211,6 +212,7 @@ public class DeviceControlQueryMessageHandler extends SIPRequestProcessorParent cmdXml.append("\r\n"); try { cmder.dragZoomCmd(device,channelId,cmdXml.toString()); + responseAck(request, Response.OK); } catch (InvalidArgumentException | SipException | ParseException e) { logger.error("[命令发送失败] 拉框控制: {}", e.getMessage()); } @@ -255,7 +257,8 @@ public class DeviceControlQueryMessageHandler extends SIPRequestProcessorParent } try { cmder.alarmCmd(device, alarmMethod,alarmType, - errorResult -> onError(request,errorResult)); + errorResult -> onError(request,errorResult), + okResult -> onOk(request,okResult)); } catch (InvalidArgumentException | SipException | ParseException e) { logger.error("[命令发送失败] 告警消息: {}", e.getMessage()); } @@ -274,7 +277,8 @@ public class DeviceControlQueryMessageHandler extends SIPRequestProcessorParent String cmdString = getText(rootElement,type.getVal()); try { cmder.recordCmd(device, channelId,cmdString, - errorResult -> onError(request,errorResult)); + errorResult -> onError(request,errorResult), + okResult -> onOk(request,okResult)); } catch (InvalidArgumentException | SipException | ParseException e) { logger.error("[命令发送失败] 录像控制: {}", e.getMessage()); } @@ -292,7 +296,8 @@ public class DeviceControlQueryMessageHandler extends SIPRequestProcessorParent String cmdString = getText(rootElement,type.getVal()); try { cmder.guardCmd(device, cmdString, - errorResult -> onError(request,errorResult)); + errorResult -> onError(request,errorResult), + okResult -> onOk(request,okResult)); } catch (InvalidArgumentException | SipException | ParseException e) { logger.error("[命令发送失败] 布防/撤防命令: {}", e.getMessage()); } diff --git a/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/device/DeviceControl.java b/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/device/DeviceControl.java index 2a44eb54..3e1b423a 100644 --- a/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/device/DeviceControl.java +++ b/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/device/DeviceControl.java @@ -110,7 +110,7 @@ public class DeviceControl { msg.setKey(key); msg.setData(String.format("开始/停止录像操作失败，错误码： %s, %s", event.statusCode, event.msg)); resultHolder.invokeAllResult(msg); - }); + },null); } catch (InvalidArgumentException | SipException | ParseException e) { logger.error("[命令发送失败] 开始/停止录像: {}", e.getMessage()); throw new ControllerException(ErrorCode.ERROR100.getCode(), "命令发送失败: " + e.getMessage()); @@ -143,7 +143,7 @@ public class DeviceControl { msg.setKey(key); msg.setData(String.format("布防/撤防操作失败，错误码： %s, %s", event.statusCode, event.msg)); resultHolder.invokeResult(msg); - }); + },null); } catch (InvalidArgumentException | SipException | ParseException e) { logger.error("[命令发送失败] 布防/撤防操作: {}", e.getMessage()); throw new ControllerException(ErrorCode.ERROR100.getCode(), "命令发送: " + e.getMessage()); @@ -192,7 +192,7 @@ public class DeviceControl { msg.setKey(key); msg.setData(String.format("报警复位操作失败，错误码： %s, %s", event.statusCode, event.msg)); resultHolder.invokeResult(msg); - }); + },null); } catch (InvalidArgumentException | SipException | ParseException e) { logger.error("[命令发送失败] 报警复位: {}", e.getMessage()); throw new ControllerException(ErrorCode.ERROR100.getCode(), "命令发送失败: " + e.getMessage());