修复WVP作为下级平台接受deviceControl指令的问题-修复指令响应

gaofw189
1 parent 40ece192
Showing 5 changed files with 81 additions and 72 deletions
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPSender.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommander.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/control/cmd/DeviceControlQueryMessageHandler.java
src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/device/DeviceControl.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();
+            }
+            if (message.getHeader(UserAgentHeader.NAME) == null) {
+                try {
+                    message.addHeader(SipUtils.createUserAgentHeader(sipLayer.getSipFactory(), gitUtil));
+                } catch (ParseException e) {
+                    logger.error("添加UserAgentHeader失败", e);
+                }
             }
-        }
-        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;
+            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) {
-                tcpSipProvider.sendRequest((Request)message);
-            }else if (message instanceof Response) {
-                tcpSipProvider.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);
+            } 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);
         }
     }
@@ -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帧
@@ -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("<?xml version=\"1.0\" encoding=\"" + charset + "\"?>\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("</Control>\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);
     }
     /**
@@ -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("</" + type.getVal() + ">\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());
         }
@@ -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());