lawrencehj
2021-08-26 45969a00b74b5a240c0dc6e2c37f5381cd0cb65f
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/MessageRequestProcessor.java
@@ -4,17 +4,12 @@
import java.text.ParseException;
import java.util.*;
import javax.sip.*;
import javax.sip.address.SipURI;
import javax.sip.header.FromHeader;
import javax.sip.header.HeaderAddress;
import javax.sip.header.ToHeader;
import javax.sip.InvalidArgumentException;
import javax.sip.ListeningPoint;
import javax.sip.ObjectInUseException;
import javax.sip.RequestEvent;
import javax.sip.SipException;
import javax.sip.SipProvider;
import javax.sip.message.Request;
import javax.sip.message.Response;
@@ -35,12 +30,13 @@
import com.genersoft.iot.vmp.gb28181.utils.DateUtil;
import com.genersoft.iot.vmp.gb28181.utils.NumericUtil;
import com.genersoft.iot.vmp.gb28181.utils.XmlUtil;
import com.genersoft.iot.vmp.service.IDeviceAlarmService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
import com.genersoft.iot.vmp.utils.GpsUtil;
import com.genersoft.iot.vmp.utils.SpringBeanFactory;
import com.genersoft.iot.vmp.utils.redis.RedisUtil;
import com.genersoft.iot.vmp.vmanager.platform.bean.ChannelReduce;
import com.genersoft.iot.vmp.vmanager.gb28181.platform.bean.ChannelReduce;
import gov.nist.javax.sip.SipStackImpl;
import gov.nist.javax.sip.address.AddressImpl;
@@ -84,6 +80,8 @@
   private DeviceOffLineDetector offLineDetector;
   private IDeviceAlarmService deviceAlarmService;
   private final static String CACHE_RECORDINFO_KEY = "CACHE_RECORDINFO_";
   private static final String MESSAGE_KEEP_ALIVE = "Keepalive";
@@ -114,13 +112,13 @@
         String cmd = XmlUtil.getText(rootElement, "CmdType");
         if (MESSAGE_KEEP_ALIVE.equals(cmd)) {
            logger.info("接收到KeepAlive消息");
            logger.debug("接收到KeepAlive消息");
            processMessageKeepAlive(evt);
         } else if (MESSAGE_CONFIG_DOWNLOAD.equals(cmd)) {
            logger.info("接收到ConfigDownload消息");
            logger.debug("接收到ConfigDownload消息");
            processMessageConfigDownload(evt);
         } else if (MESSAGE_CATALOG.equals(cmd)) {
            logger.info("接收到Catalog消息");
            logger.debug("接收到Catalog消息");
            processMessageCatalogList(evt);
         } else if (MESSAGE_DEVICE_INFO.equals(cmd)) {
            // DeviceInfo消息处理
@@ -129,31 +127,31 @@
            // DeviceStatus消息处理
            processMessageDeviceStatus(evt);
         } else if (MESSAGE_DEVICE_CONTROL.equals(cmd)) {
            logger.info("接收到DeviceControl消息");
            logger.debug("接收到DeviceControl消息");
            processMessageDeviceControl(evt);
         } else if (MESSAGE_DEVICE_CONFIG.equals(cmd)) {
            logger.info("接收到DeviceConfig消息");
            processMessageDeviceConfig(evt);
         } else if (MESSAGE_ALARM.equals(cmd)) {
            logger.info("接收到Alarm消息");
            logger.debug("接收到Alarm消息");
            processMessageAlarm(evt);
         } else if (MESSAGE_RECORD_INFO.equals(cmd)) {
            logger.info("接收到RecordInfo消息");
            logger.debug("接收到RecordInfo消息");
            processMessageRecordInfo(evt);
         }else if (MESSAGE_MEDIA_STATUS.equals(cmd)) {
            logger.info("接收到MediaStatus消息");
            logger.debug("接收到MediaStatus消息");
            processMessageMediaStatus(evt);
         } else if (MESSAGE_MOBILE_POSITION.equals(cmd)) {
            logger.info("接收到MobilePosition消息");
            logger.debug("接收到MobilePosition消息");
            processMessageMobilePosition(evt);
         } else if (MESSAGE_PRESET_QUERY.equals(cmd)) {
            logger.info("接收到PresetQuery消息");
            logger.debug("接收到PresetQuery消息");
            processMessagePresetQuery(evt);
         } else if (MESSAGE_BROADCAST.equals(cmd)) {
            // Broadcast消息处理
            processMessageBroadcast(evt);
         } else {
            logger.info("接收到消息:" + cmd);
            logger.debug("接收到消息:" + cmd);
            responseAck(evt);
         }
      } catch (DocumentException | SipException |InvalidArgumentException | ParseException e) {
@@ -226,7 +224,7 @@
         String name = rootElement.getName();
         Element deviceIdElement = rootElement.element("DeviceID");
         String deviceId = deviceIdElement.getText();
         Device device = storager.queryVideoDevice(deviceId);
         if (name.equalsIgnoreCase("Query")) { // 区分是Response——查询响应,还是Query——查询请求
            logger.info("接收到DeviceStatus查询消息");
            FromHeader fromHeader = (FromHeader) evt.getRequest().getHeader(FromHeader.NAME);
@@ -238,7 +236,7 @@
               // 回复200 OK
               responseAck(evt);
               String sn = rootElement.element("SN").getText();
               ParentPlatform parentPlatform = storager.queryParentPlatById(platformId);
               ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(platformId);
               cmderFroPlatform.deviceStatusResponse(parentPlatform, sn, fromHeader.getTag());
            }
         } else {
@@ -259,7 +257,7 @@
               deferredResultHolder.invokeResult(msg);
               if (offLineDetector.isOnline(deviceId)) {
                  publisher.onlineEventPublish(deviceId, VideoManagerConstants.EVENT_ONLINE_KEEPLIVE);
                  publisher.onlineEventPublish(device, VideoManagerConstants.EVENT_ONLINE_MESSAGE);
               } else {
               }
            }
@@ -282,7 +280,7 @@
         //String result = XmlUtil.getText(rootElement, "Result");
         // 回复200 OK
         responseAck(evt);
         if (rootElement.getName().equals("Response")) {//} !XmlUtil.isEmpty(result)) {
         if (rootElement.getName().equals("Response")) {//} !StringUtils.isEmpty(result)) {
            // 此处是对本平台发出DeviceControl指令的应答
            JSONObject json = new JSONObject();
            XmlUtil.node2Json(rootElement, json);
@@ -299,11 +297,11 @@
            String platformId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(FromHeader.NAME)).getAddress().getURI()).getUser();
            String targetGBId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(ToHeader.NAME)).getAddress().getURI()).getUser();
            // 远程启动功能
            if (!XmlUtil.isEmpty(XmlUtil.getText(rootElement, "TeleBoot"))) {
            if (!StringUtils.isEmpty(XmlUtil.getText(rootElement, "TeleBoot"))) {
               if (deviceId.equals(targetGBId)) {
                  // 远程启动本平台:需要在重新启动程序后先对SipStack解绑
                  logger.info("执行远程启动本平台命令");
                  ParentPlatform parentPlatform = storager.queryParentPlatById(platformId);
                  ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(platformId);
                  cmderFroPlatform.unregister(parentPlatform, null, null);
   
                  Thread restartThread = new Thread(new Runnable() {
@@ -337,7 +335,7 @@
               }
            }
            // 云台/前端控制命令
            if (!XmlUtil.isEmpty(XmlUtil.getText(rootElement,"PTZCmd")) && !deviceId.equals(targetGBId)) {
            if (!StringUtils.isEmpty(XmlUtil.getText(rootElement,"PTZCmd")) && !deviceId.equals(targetGBId)) {
               String cmdString = XmlUtil.getText(rootElement,"PTZCmd");
               Device device = storager.queryVideoDeviceByPlatformIdAndChannelId(platformId, deviceId);
               cmder.fronEndCmd(device, deviceId, cmdString);
@@ -421,7 +419,7 @@
         String deviceId = XmlUtil.getText(rootElement, "DeviceID");
         // 回复200 OK
         responseAck(evt);
         if (rootElement.getName().equals("Response")) {//   !XmlUtil.isEmpty(result)) {
         if (rootElement.getName().equals("Response")) {//   !StringUtils.isEmpty(result)) {
            // 此处是对本平台发出DeviceControl指令的应答
            JSONObject json = new JSONObject();
            XmlUtil.node2Json(rootElement, json);
@@ -452,6 +450,7 @@
         String requestName = rootElement.getName();
         Element deviceIdElement = rootElement.element("DeviceID");
         String deviceId = deviceIdElement.getTextTrim().toString();
         Device device = storager.queryVideoDevice(deviceId);
         if (requestName.equals("Query")) {
            logger.info("接收到DeviceInfo查询消息");
            FromHeader fromHeader = (FromHeader) evt.getRequest().getHeader(FromHeader.NAME);
@@ -463,12 +462,11 @@
               // 回复200 OK
               responseAck(evt);
               String sn = rootElement.element("SN").getText();
               ParentPlatform parentPlatform = storager.queryParentPlatById(platformId);
               ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(platformId);
               cmderFroPlatform.deviceInfoResponse(parentPlatform, sn, fromHeader.getTag());
            }
         } else {
            logger.info("接收到DeviceInfo应答消息");
            Device device = storager.queryVideoDevice(deviceId);
            logger.debug("接收到DeviceInfo应答消息");
            if (device == null) {
               return;
            }
@@ -489,7 +487,7 @@
            // 回复200 OK
            responseAck(evt);
            if (offLineDetector.isOnline(deviceId)) {
               publisher.onlineEventPublish(deviceId, VideoManagerConstants.EVENT_ONLINE_KEEPLIVE);
               publisher.onlineEventPublish(device, VideoManagerConstants.EVENT_ONLINE_MESSAGE);
            }
         }
      } catch (DocumentException | SipException | InvalidArgumentException | ParseException e) {
@@ -517,7 +515,7 @@
         // if (deviceListElement == null) { // 存在DeviceList则为响应 catalog, 不存在DeviceList则为查询请求
         if (name.equalsIgnoreCase("Query")) { // 区分是Response——查询响应,还是Query——查询请求
            // TODO 后续将代码拆分
            ParentPlatform parentPlatform = storager.queryParentPlatById(platformId);
            ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(platformId);
            if (parentPlatform == null) {
               response404Ack(evt);
               return;
@@ -529,13 +527,44 @@
               String sn = snElement.getText();
               // 准备回复通道信息
               List<ChannelReduce> channelReduces = storager.queryChannelListInParentPlatform(parentPlatform.getServerGBId());
               // 查询关联的直播通道
               List<GbStream> gbStreams = storager.queryGbStreamListInPlatform(parentPlatform.getServerGBId());
               int size = channelReduces.size() + gbStreams.size();
               // 回复级联的通道
               if (channelReduces.size() > 0) {
                  for (ChannelReduce channelReduce : channelReduces) {
                     DeviceChannel deviceChannel = storager.queryChannel(channelReduce.getDeviceId(), channelReduce.getChannelId());
                     cmderFroPlatform.catalogQuery(deviceChannel, parentPlatform, sn, fromHeader.getTag(), channelReduces.size());
                     cmderFroPlatform.catalogQuery(deviceChannel, parentPlatform, sn, fromHeader.getTag(), size);
                  }
               }
               // 回复直播的通道
               if (gbStreams.size() > 0) {
                  for (GbStream gbStream : gbStreams) {
                     DeviceChannel deviceChannel = new DeviceChannel();
                     deviceChannel.setChannelId(gbStream.getGbId());
                     deviceChannel.setName(gbStream.getName());
                     deviceChannel.setLongitude(gbStream.getLongitude());
                     deviceChannel.setLatitude(gbStream.getLatitude());
                     deviceChannel.setDeviceId(parentPlatform.getDeviceGBId());
                     deviceChannel.setManufacture("wvp-pro");
                     deviceChannel.setStatus(gbStream.isStatus()?1:0);
//                     deviceChannel.setParentId(parentPlatform.getDeviceGBId());
                     deviceChannel.setRegisterWay(1);
                     deviceChannel.setCivilCode(cmder.getSipConfig().getDomain());
                     deviceChannel.setModel("live");
                     deviceChannel.setOwner("wvp-pro");
//                     deviceChannel.setAddress("test");
                     deviceChannel.setParental(0);
                     deviceChannel.setSecrecy("0");
                     deviceChannel.setSecrecy("0");
                     cmderFroPlatform.catalogQuery(deviceChannel, parentPlatform, sn, fromHeader.getTag(), size);
                  }
               }
               if (size == 0) {
                  // 回复无通道
                  cmderFroPlatform.catalogQuery(null, parentPlatform, sn, fromHeader.getTag(), size);
               }
            }
@@ -638,7 +667,7 @@
               // 回复200 OK
               responseAck(evt);
               if (offLineDetector.isOnline(deviceId)) {
                  publisher.onlineEventPublish(deviceId, VideoManagerConstants.EVENT_ONLINE_KEEPLIVE);
                  publisher.onlineEventPublish(device, VideoManagerConstants.EVENT_ONLINE_MESSAGE);
               }
            }
         }
@@ -687,7 +716,7 @@
               deviceAlarm.setLatitude(0.00);
            }
   
            if (!XmlUtil.isEmpty(deviceAlarm.getAlarmMethod())) {
            if (!StringUtils.isEmpty(deviceAlarm.getAlarmMethod())) {
               if ( deviceAlarm.getAlarmMethod().equals("4")) {
                  MobilePosition mobilePosition = new MobilePosition();
                  mobilePosition.setDeviceId(deviceAlarm.getDeviceId());
@@ -707,7 +736,9 @@
                  storager.insertMobilePosition(mobilePosition);
               }
            }
            // TODO: 需要实现存储报警信息、报警分类
            System.out.println("存储报警信息、报警分类");
            // 存储报警信息、报警分类
            deviceAlarmService.add(deviceAlarm);
   
            if (offLineDetector.isOnline(deviceId)) {
               publisher.deviceAlarmEventPublish(deviceAlarm);
@@ -725,7 +756,6 @@
            deferredResultHolder.invokeResult(msg);
         }
      } catch (DocumentException | SipException | InvalidArgumentException | ParseException e) {
         // } catch (DocumentException e) {
         e.printStackTrace();
      }
   }
@@ -739,15 +769,34 @@
      try {
         Element rootElement = getRootElement(evt);
         String deviceId = XmlUtil.getText(rootElement, "DeviceID");
         // 检查设备是否存在, 不存在则不回复
         if (storager.exists(deviceId)) {
         Device device = storager.queryVideoDevice(deviceId);
         // 检查设备是否存在并在线, 不在线则设置为在线
         if (device != null ) {
            // 回复200 OK
            responseAck(evt);
            if (offLineDetector.isOnline(deviceId)) {
               publisher.onlineEventPublish(deviceId, VideoManagerConstants.EVENT_ONLINE_KEEPLIVE);
            } else {
            }
            publisher.onlineEventPublish(device, VideoManagerConstants.EVENT_ONLINE_KEEPLIVE);
         }else{
            logger.warn("收到[ "+deviceId+" ]心跳信息, 但是设备不存在, 回复404");
            Response response = getMessageFactory().createResponse(Response.NOT_FOUND, evt.getRequest());
            ServerTransaction serverTransaction = getServerTransaction(evt);
            serverTransaction.sendResponse(response);
            if (serverTransaction.getDialog() != null) serverTransaction.getDialog().delete();
         }
//         if (device != null && device.getOnline() == 1) {
//
//            if (offLineDetector.isOnline(deviceId)) {
//               publisher.onlineEventPublish(device, VideoManagerConstants.EVENT_ONLINE_KEEPLIVE);
//            } else {
//            }
//         }else {
////            logger.warn("收到[ "+deviceId+" ]心跳信息, 但是设备" + (device == null? "不存在":"离线") + ", 回复401");
////            Response response = getMessageFactory().createResponse(Response.UNAUTHORIZED, evt.getRequest());
////            getServerTransaction(evt).sendResponse(response);
//            publisher.onlineEventPublish(device, VideoManagerConstants.EVENT_ONLINE_KEEPLIVE);
//
//         }
      } catch (ParseException | SipException | InvalidArgumentException | DocumentException e) {
         e.printStackTrace();
      }
@@ -891,7 +940,7 @@
            StreamInfo streamInfo = redisCatchStorage.queryPlaybackByDevice(deviceId, "*");
            if (streamInfo != null) {
               redisCatchStorage.stopPlayback(streamInfo);
               cmder.streamByeCmd(streamInfo.getStreamId());
               cmder.streamByeCmd(streamInfo.getDeviceID(), streamInfo.getChannelId());
            }
         }
      } catch (ParseException | SipException | InvalidArgumentException | DocumentException e) {
@@ -940,7 +989,9 @@
    */
   private void responseAck(RequestEvent evt) throws SipException, InvalidArgumentException, ParseException {
      Response response = getMessageFactory().createResponse(Response.OK, evt.getRequest());
      getServerTransaction(evt).sendResponse(response);
      ServerTransaction serverTransaction = getServerTransaction(evt);
      serverTransaction.sendResponse(response);
      if (serverTransaction.getDialog() != null) serverTransaction.getDialog().delete();
   }
   /***
@@ -952,7 +1003,9 @@
    */
   private void response404Ack(RequestEvent evt) throws SipException, InvalidArgumentException, ParseException {
      Response response = getMessageFactory().createResponse(Response.NOT_FOUND, evt.getRequest());
      getServerTransaction(evt).sendResponse(response);
      ServerTransaction serverTransaction = getServerTransaction(evt);
      serverTransaction.sendResponse(response);
      if (serverTransaction.getDialog() != null) serverTransaction.getDialog().delete();
   }
   private Element getRootElement(RequestEvent evt) throws DocumentException {
@@ -1002,4 +1055,8 @@
   public void setCmderFroPlatform(SIPCommanderFroPlatform cmderFroPlatform) {
      this.cmderFroPlatform = cmderFroPlatform;
   }
   public void setDeviceAlarmService(IDeviceAlarmService deviceAlarmService) {
      this.deviceAlarmService = deviceAlarmService;
   }
}