guzijian / wvp-GB28181-pro · Commits

Commit 4c66f6b29eda2d459aed86f7a138438191de7e47

Authored by 648540858
1 parent 4f2191dc

优化消息处理中存在可能异常的处理流程

Inline Side-by-side
Showing 12 changed files with 411 additions and 303 deletions
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java
  View file @4c66f6b
... ... @@ -132,7 +132,11 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
132 132 if (requesterId == null || channelId == null) {
133 133 logger.info("无法从FromHeader的Address中获取到平台id，返回400");
134 134 // 参数不全， 发400，请求错误
135   - responseAck(serverTransaction, Response.BAD_REQUEST);
  135 + try {
  136 + responseAck(serverTransaction, Response.BAD_REQUEST);
  137 + } catch (SipException | InvalidArgumentException | ParseException e) {
  138 + logger.error("[命令发送失败] invite BAD_REQUEST: {}", e.getMessage());
  139 + }
136 140 return;
137 141 }
138 142  
... ... @@ -141,6 +145,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
141 145 ParentPlatform platform = storager.queryParentPlatByServerGBId(requesterId);
142 146 if (platform == null) {
143 147 inviteFromDeviceHandle(serverTransaction, requesterId);
  148 +
144 149 } else {
145 150 // 查询平台下是否有该通道
146 151 DeviceChannel channel = storager.queryChannelInParentPlatform(requesterId, channelId);
... ... @@ -158,7 +163,11 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
158 163 // return;
159 164 // }
160 165 // 通道存在，发100，TRYING
161   - responseAck(serverTransaction, Response.TRYING);
  166 + try {
  167 + responseAck(serverTransaction, Response.TRYING);
  168 + } catch (SipException | InvalidArgumentException | ParseException e) {
  169 + logger.error("[命令发送失败] invite TRYING: {}", e.getMessage());
  170 + }
162 171 } else if (channel == null && gbStream != null) {
163 172  
164 173 String mediaServerId = gbStream.getMediaServerId();
... ... @@ -166,13 +175,21 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
166 175 if (mediaServerItem == null) {
167 176 if ("proxy".equals(gbStream.getStreamType())) {
168 177 logger.info("[ app={}, stream={} ]找不到zlm {}，返回410", gbStream.getApp(), gbStream.getStream(), mediaServerId);
169   - responseAck(serverTransaction, Response.GONE);
  178 + try {
  179 + responseAck(serverTransaction, Response.GONE);
  180 + } catch (SipException | InvalidArgumentException | ParseException e) {
  181 + logger.error("[命令发送失败] invite GONE: {}", e.getMessage());
  182 + }
170 183 return;
171 184 } else {
172 185 streamPushItem = streamPushService.getPush(gbStream.getApp(), gbStream.getStream());
173 186 if (streamPushItem == null || streamPushItem.getServerId().equals(userSetting.getServerId())) {
174 187 logger.info("[ app={}, stream={} ]找不到zlm {}，返回410", gbStream.getApp(), gbStream.getStream(), mediaServerId);
175   - responseAck(serverTransaction, Response.GONE);
  188 + try {
  189 + responseAck(serverTransaction, Response.GONE);
  190 + } catch (SipException | InvalidArgumentException | ParseException e) {
  191 + logger.error("[命令发送失败] invite GONE: {}", e.getMessage());
  192 + }
176 193 return;
177 194 }
178 195 }
... ... @@ -181,25 +198,47 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
181 198 streamPushItem = streamPushService.getPush(gbStream.getApp(), gbStream.getStream());
182 199 if (streamPushItem == null) {
183 200 logger.info("[ app={}, stream={} ]找不到zlm {}，返回410", gbStream.getApp(), gbStream.getStream(), mediaServerId);
184   - responseAck(serverTransaction, Response.GONE);
  201 + try {
  202 + responseAck(serverTransaction, Response.GONE);
  203 + } catch (SipException | InvalidArgumentException | ParseException e) {
  204 + logger.error("[命令发送失败] invite GONE: {}", e.getMessage());
  205 + }
185 206 return;
186 207 }
187 208 }else if("proxy".equals(gbStream.getStreamType())){
188 209 proxyByAppAndStream = streamProxyService.getStreamProxyByAppAndStream(gbStream.getApp(), gbStream.getStream());
189 210 if (proxyByAppAndStream == null) {
190 211 logger.info("[ app={}, stream={} ]找不到zlm {}，返回410", gbStream.getApp(), gbStream.getStream(), mediaServerId);
191   - responseAck(serverTransaction, Response.GONE);
  212 + try {
  213 + responseAck(serverTransaction, Response.GONE);
  214 + } catch (SipException | InvalidArgumentException | ParseException e) {
  215 + logger.error("[命令发送失败] invite GONE: {}", e.getMessage());
  216 + }
192 217 return;
193 218 }
194 219 }
195 220 }
196   - responseAck(serverTransaction, Response.CALL_IS_BEING_FORWARDED); // 通道存在，发181，呼叫转接中
  221 + try {
  222 + responseAck(serverTransaction, Response.CALL_IS_BEING_FORWARDED);
  223 + } catch (SipException | InvalidArgumentException | ParseException e) {
  224 + logger.error("[命令发送失败] invite CALL_IS_BEING_FORWARDED: {}", e.getMessage());
  225 + }
197 226 } else if (catalog != null) {
198   - responseAck(serverTransaction, Response.BAD_REQUEST, "catalog channel can not play"); // 目录不支持点播
  227 + try {
  228 + // 目录不支持点播
  229 + responseAck(serverTransaction, Response.BAD_REQUEST, "catalog channel can not play");
  230 + } catch (SipException | InvalidArgumentException | ParseException e) {
  231 + logger.error("[命令发送失败] invite 目录不支持点播: {}", e.getMessage());
  232 + }
199 233 return;
200 234 } else {
201 235 logger.info("通道不存在，返回404");
202   - responseAck(serverTransaction, Response.NOT_FOUND); // 通道不存在，发404，资源不存在
  236 + try {
  237 + // 通道不存在，发404，资源不存在
  238 + responseAck(serverTransaction, Response.NOT_FOUND);
  239 + } catch (SipException | InvalidArgumentException | ParseException e) {
  240 + logger.error("[命令发送失败] invite 通道不存在: {}", e.getMessage());
  241 + }
203 242 return;
204 243 }
205 244 // 解析sdp消息, 使用jainsip 自带的sdp解析方式
... ... @@ -270,7 +309,12 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
270 309 if (port == -1) {
271 310 logger.info("不支持的媒体格式，返回415");
272 311 // 回复不支持的格式
273   - responseAck(serverTransaction, Response.UNSUPPORTED_MEDIA_TYPE); // 不支持的格式，发415
  312 + try {
  313 + // 不支持的格式，发415
  314 + responseAck(serverTransaction, Response.UNSUPPORTED_MEDIA_TYPE);
  315 + } catch (SipException | InvalidArgumentException | ParseException e) {
  316 + logger.error("[命令发送失败] invite 不支持的格式: {}", e.getMessage());
  317 + }
274 318 return;
275 319 }
276 320 String username = sdp.getOrigin().getUsername();
... ... @@ -283,13 +327,21 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
283 327 device = storager.queryVideoDeviceByPlatformIdAndChannelId(requesterId, channelId);
284 328 if (device == null) {
285 329 logger.warn("点播平台{}的通道{}时未找到设备信息", requesterId, channel);
286   - responseAck(serverTransaction, Response.SERVER_INTERNAL_ERROR);
  330 + try {
  331 + responseAck(serverTransaction, Response.SERVER_INTERNAL_ERROR);
  332 + } catch (SipException | InvalidArgumentException | ParseException e) {
  333 + logger.error("[命令发送失败] invite 未找到设备信息: {}", e.getMessage());
  334 + }
287 335 return;
288 336 }
289 337 mediaServerItem = playService.getNewMediaServerItem(device);
290 338 if (mediaServerItem == null) {
291 339 logger.warn("未找到可用的zlm");
292   - responseAck(serverTransaction, Response.BUSY_HERE);
  340 + try {
  341 + responseAck(serverTransaction, Response.BUSY_HERE);
  342 + } catch (SipException | InvalidArgumentException | ParseException e) {
  343 + logger.error("[命令发送失败] invite BUSY_HERE: {}", e.getMessage());
  344 + }
293 345 return;
294 346 }
295 347 SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
... ... @@ -301,7 +353,11 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
301 353 }
302 354 if (sendRtpItem == null) {
303 355 logger.warn("服务器端口资源不足");
304   - responseAck(serverTransaction, Response.BUSY_HERE);
  356 + try {
  357 + responseAck(serverTransaction, Response.BUSY_HERE);
  358 + } catch (SipException | InvalidArgumentException | ParseException e) {
  359 + logger.error("[命令发送失败] invite 服务器端口资源不足: {}", e.getMessage());
  360 + }
305 361 return;
306 362 }
307 363 sendRtpItem.setCallId(callIdHeader.getCallId());
... ... @@ -474,13 +530,8 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
474 530 }
475 531 }
476 532 }
477   -
478   - } catch (SipException | InvalidArgumentException | ParseException e) {
479   - e.printStackTrace();
480   - logger.warn("sdp解析错误");
481   - e.printStackTrace();
482 533 } catch (SdpParseException e) {
483   - e.printStackTrace();
  534 + logger.error("sdp解析错误", e);
484 535 } catch (SdpException e) {
485 536 e.printStackTrace();
486 537 }
... ... @@ -492,7 +543,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
492 543 private void pushProxyStream(RequestEvent evt, ServerTransaction serverTransaction, GbStream gbStream, ParentPlatform platform,
493 544 CallIdHeader callIdHeader, MediaServerItem mediaServerItem,
494 545 int port, Boolean tcpActive, boolean mediaTransmissionTCP,
495   - String channelId, String addressStr, String ssrc, String requesterId) throws InvalidArgumentException, ParseException, SipException {
  546 + String channelId, String addressStr, String ssrc, String requesterId) {
496 547 Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, gbStream.getApp(), gbStream.getStream());
497 548 if (streamReady) {
498 549 // 自平台内容
... ... @@ -502,7 +553,11 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
502 553  
503 554 if (sendRtpItem == null) {
504 555 logger.warn("服务器端口资源不足");
505   - responseAck(serverTransaction, Response.BUSY_HERE);
  556 + try {
  557 + responseAck(serverTransaction, Response.BUSY_HERE);
  558 + } catch (SipException | InvalidArgumentException | ParseException e) {
  559 + logger.error("[命令发送失败] invite 服务器端口资源不足: {}", e.getMessage());
  560 + }
506 561 return;
507 562 }
508 563 if (tcpActive != null) {
... ... @@ -527,7 +582,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
527 582 private void pushStream(RequestEvent evt, ServerTransaction serverTransaction, GbStream gbStream, StreamPushItem streamPushItem, ParentPlatform platform,
528 583 CallIdHeader callIdHeader, MediaServerItem mediaServerItem,
529 584 int port, Boolean tcpActive, boolean mediaTransmissionTCP,
530   - String channelId, String addressStr, String ssrc, String requesterId) throws InvalidArgumentException, ParseException, SipException {
  585 + String channelId, String addressStr, String ssrc, String requesterId) {
531 586 // 推流
532 587 if (streamPushItem.isSelf()) {
533 588 Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, gbStream.getApp(), gbStream.getStream());
... ... @@ -539,7 +594,11 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
539 594  
540 595 if (sendRtpItem == null) {
541 596 logger.warn("服务器端口资源不足");
542   - responseAck(serverTransaction, Response.BUSY_HERE);
  597 + try {
  598 + responseAck(serverTransaction, Response.BUSY_HERE);
  599 + } catch (SipException | InvalidArgumentException | ParseException e) {
  600 + logger.error("[命令发送失败] invite 服务器端口资源不足: {}", e.getMessage());
  601 + }
543 602 return;
544 603 }
545 604 if (tcpActive != null) {
... ... @@ -577,15 +636,23 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
577 636 private void notifyStreamOnline(RequestEvent evt, ServerTransaction serverTransaction, GbStream gbStream, StreamPushItem streamPushItem, ParentPlatform platform,
578 637 CallIdHeader callIdHeader, MediaServerItem mediaServerItem,
579 638 int port, Boolean tcpActive, boolean mediaTransmissionTCP,
580   - String channelId, String addressStr, String ssrc, String requesterId) throws InvalidArgumentException, ParseException, SipException {
  639 + String channelId, String addressStr, String ssrc, String requesterId) {
581 640 if ("proxy".equals(gbStream.getStreamType())) {
582 641 // TODO 控制启用以使设备上线
583 642 logger.info("[ app={}, stream={} ]通道未推流，启用流后开始推流", gbStream.getApp(), gbStream.getStream());
584   - responseAck(serverTransaction, Response.BAD_REQUEST, "channel [" + gbStream.getGbId() + "] offline");
  643 + try {
  644 + responseAck(serverTransaction, Response.BAD_REQUEST, "channel [" + gbStream.getGbId() + "] offline");
  645 + } catch (SipException | InvalidArgumentException | ParseException e) {
  646 + logger.error("[命令发送失败] invite 通道未推流: {}", e.getMessage());
  647 + }
585 648 } else if ("push".equals(gbStream.getStreamType())) {
586 649 if (!platform.isStartOfflinePush()) {
587 650 // 平台设置中关闭了拉起离线的推流则直接回复
588   - responseAck(serverTransaction, Response.TEMPORARILY_UNAVAILABLE, "channel stream not pushing");
  651 + try {
  652 + responseAck(serverTransaction, Response.TEMPORARILY_UNAVAILABLE, "channel stream not pushing");
  653 + } catch (SipException | InvalidArgumentException | ParseException e) {
  654 + logger.error("[命令发送失败] invite 通道未推流: {}", e.getMessage());
  655 + }
589 656 return;
590 657 }
591 658 // 发送redis消息以使设备上线
... ... @@ -713,38 +780,28 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
713 780 }
714 781 redisCatchStorage.updateSendRTPSever(sendRtpItem);
715 782 }, (wvpResult) -> {
716   - try {
717   - // 错误
718   - if (wvpResult.getCode() == RedisGbPlayMsgListener.ERROR_CODE_OFFLINE) {
719   - // 离线
720   - // 查询是否在本机上线了
721   - StreamPushItem currentStreamPushItem = streamPushService.getPush(streamPushItem.getApp(), streamPushItem.getStream());
722   - if (currentStreamPushItem.isPushIng()) {
723   - // 在线状态
724   - pushStream(evt, serverTransaction, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
725   - mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
726 783  
727   - } else {
728   - // 不在线 拉起
729   - notifyStreamOnline(evt, serverTransaction, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
730   - mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
731   - }
  784 + // 错误
  785 + if (wvpResult.getCode() == RedisGbPlayMsgListener.ERROR_CODE_OFFLINE) {
  786 + // 离线
  787 + // 查询是否在本机上线了
  788 + StreamPushItem currentStreamPushItem = streamPushService.getPush(streamPushItem.getApp(), streamPushItem.getStream());
  789 + if (currentStreamPushItem.isPushIng()) {
  790 + // 在线状态
  791 + pushStream(evt, serverTransaction, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
  792 + mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
  793 +
  794 + } else {
  795 + // 不在线 拉起
  796 + notifyStreamOnline(evt, serverTransaction, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
  797 + mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
732 798 }
733   - } catch (InvalidArgumentException | ParseException | SipException e) {
734   - logger.error("[命令发送失败] 国标级联 点播回复: {}", e.getMessage());
735 799 }
736   -
737   -
738 800 try {
739 801 responseAck(serverTransaction, Response.BUSY_HERE);
740   - } catch (SipException e) {
741   - e.printStackTrace();
742   - } catch (InvalidArgumentException e) {
743   - e.printStackTrace();
744   - } catch (ParseException e) {
745   - e.printStackTrace();
  802 + } catch (InvalidArgumentException | ParseException | SipException e) {
  803 + logger.error("[命令发送失败] 国标级联 点播回复 BUSY_HERE: {}", e.getMessage());
746 804 }
747   - return;
748 805 });
749 806 }
750 807  
... ... @@ -782,14 +839,17 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
782 839 return null;
783 840 }
784 841  
785   - public void inviteFromDeviceHandle(ServerTransaction serverTransaction, String requesterId) throws InvalidArgumentException, ParseException, SipException, SdpException {
  842 + public void inviteFromDeviceHandle(ServerTransaction serverTransaction, String requesterId) {
786 843  
787 844 // 非上级平台请求，查询是否设备请求（通常为接收语音广播的设备）
788 845 Device device = redisCatchStorage.getDevice(requesterId);
789 846 if (device != null) {
790 847 logger.info("收到设备" + requesterId + "的语音广播Invite请求");
791   - responseAck(serverTransaction, Response.TRYING);
792   -
  848 + try {
  849 + responseAck(serverTransaction, Response.TRYING);
  850 + } catch (SipException | InvalidArgumentException | ParseException e) {
  851 + logger.error("[命令发送失败] invite BAD_REQUEST: {}", e.getMessage());
  852 + }
793 853 String contentString = new String(serverTransaction.getRequest().getRawContent());
794 854 // jainSip不支持y=字段， 移除移除以解析。
795 855 String substring = contentString;
... ... @@ -803,51 +863,65 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
803 863 if (ssrcIndex > 0) {
804 864 substring = contentString.substring(0, ssrcIndex);
805 865 }
806   - SessionDescription sdp = SdpFactory.getInstance().createSessionDescription(substring);
807   -
808   - // 获取支持的格式
809   - Vector mediaDescriptions = sdp.getMediaDescriptions(true);
810   - // 查看是否支持PS 负载96
811   - int port = -1;
812   - //boolean recvonly = false;
813   - boolean mediaTransmissionTCP = false;
814   - Boolean tcpActive = null;
815   - for (int i = 0; i < mediaDescriptions.size(); i++) {
816   - MediaDescription mediaDescription = (MediaDescription) mediaDescriptions.get(i);
817   - Media media = mediaDescription.getMedia();
818   -
819   - Vector mediaFormats = media.getMediaFormats(false);
820   - if (mediaFormats.contains("8")) {
821   - port = media.getMediaPort();
822   - String protocol = media.getProtocol();
823   - // 区分TCP发流还是udp， 当前默认udp
824   - if ("TCP/RTP/AVP".equals(protocol)) {
825   - String setup = mediaDescription.getAttribute("setup");
826   - if (setup != null) {
827   - mediaTransmissionTCP = true;
828   - if ("active".equals(setup)) {
829   - tcpActive = true;
830   - } else if ("passive".equals(setup)) {
831   - tcpActive = false;
  866 + SessionDescription sdp = null;
  867 + try {
  868 + sdp = SdpFactory.getInstance().createSessionDescription(substring);
  869 + // 获取支持的格式
  870 + Vector mediaDescriptions = sdp.getMediaDescriptions(true);
  871 + // 查看是否支持PS 负载96
  872 + int port = -1;
  873 + //boolean recvonly = false;
  874 + boolean mediaTransmissionTCP = false;
  875 + Boolean tcpActive = null;
  876 + for (int i = 0; i < mediaDescriptions.size(); i++) {
  877 + MediaDescription mediaDescription = (MediaDescription) mediaDescriptions.get(i);
  878 + Media media = mediaDescription.getMedia();
  879 +
  880 + Vector mediaFormats = media.getMediaFormats(false);
  881 + if (mediaFormats.contains("8")) {
  882 + port = media.getMediaPort();
  883 + String protocol = media.getProtocol();
  884 + // 区分TCP发流还是udp， 当前默认udp
  885 + if ("TCP/RTP/AVP".equals(protocol)) {
  886 + String setup = mediaDescription.getAttribute("setup");
  887 + if (setup != null) {
  888 + mediaTransmissionTCP = true;
  889 + if ("active".equals(setup)) {
  890 + tcpActive = true;
  891 + } else if ("passive".equals(setup)) {
  892 + tcpActive = false;
  893 + }
832 894 }
833 895 }
  896 + break;
834 897 }
835   - break;
836 898 }
  899 + if (port == -1) {
  900 + logger.info("不支持的媒体格式，返回415");
  901 + // 回复不支持的格式
  902 + try {
  903 + responseAck(serverTransaction, Response.UNSUPPORTED_MEDIA_TYPE); // 不支持的格式，发415
  904 + } catch (SipException | InvalidArgumentException | ParseException e) {
  905 + logger.error("[命令发送失败] invite 不支持的媒体格式，返回415， {}", e.getMessage());
  906 + }
  907 + return;
  908 + }
  909 + String username = sdp.getOrigin().getUsername();
  910 + String addressStr = sdp.getOrigin().getAddress();
  911 + logger.info("设备{}请求语音流，地址：{}:{}，ssrc：{}", username, addressStr, port, ssrc);
  912 + } catch (SdpException e) {
  913 + logger.error("[SDP解析异常]", e);
837 914 }
838   - if (port == -1) {
839   - logger.info("不支持的媒体格式，返回415");
840   - // 回复不支持的格式
841   - responseAck(serverTransaction, Response.UNSUPPORTED_MEDIA_TYPE); // 不支持的格式，发415
842   - return;
843   - }
844   - String username = sdp.getOrigin().getUsername();
845   - String addressStr = sdp.getOrigin().getAddress();
846   - logger.info("设备{}请求语音流，地址：{}:{}，ssrc：{}", username, addressStr, port, ssrc);
  915 +
  916 +
847 917  
848 918 } else {
849 919 logger.warn("来自无效设备/平台的请求");
850   - responseAck(serverTransaction, Response.BAD_REQUEST);
  920 + try {
  921 + responseAck(serverTransaction, Response.BAD_REQUEST);; // 不支持的格式，发415
  922 + } catch (SipException | InvalidArgumentException | ParseException e) {
  923 + logger.error("[命令发送失败] invite 来自无效设备/平台的请求， {}", e.getMessage());
  924 + }
851 925 }
852 926 }
853 927 }
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestProcessor.java
  View file @4c66f6b
... ... @@ -93,46 +93,44 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
93 93  
94 94 @Override
95 95 public void process(RequestEvent evt) {
  96 + ServerTransaction serverTransaction = getServerTransaction(evt);
96 97 try {
97   - taskQueue.offer(new HandlerCatchData(evt, null, null));
98   - ServerTransaction serverTransaction = getServerTransaction(evt);
99 98 responseAck(serverTransaction, Response.OK);
100   - if (!taskQueueHandlerRun) {
101   - taskQueueHandlerRun = true;
102   - taskExecutor.execute(()-> {
103   - while (!taskQueue.isEmpty()) {
104   - try {
105   - HandlerCatchData take = taskQueue.poll();
106   - Element rootElement = getRootElement(take.getEvt());
107   - if (rootElement == null) {
108   - logger.error("处理NOTIFY消息时未获取到消息体,{}", take.getEvt().getRequest());
109   - continue;
110   - }
111   - String cmd = XmlUtil.getText(rootElement, "CmdType");
112   -
113   - if (CmdType.CATALOG.equals(cmd)) {
114   - logger.info("接收到Catalog通知");
115   - processNotifyCatalogList(take.getEvt());
116   - } else if (CmdType.ALARM.equals(cmd)) {
117   - logger.info("接收到Alarm通知");
118   - processNotifyAlarm(take.getEvt());
119   - } else if (CmdType.MOBILE_POSITION.equals(cmd)) {
120   - logger.info("接收到MobilePosition通知");
121   - processNotifyMobilePosition(take.getEvt());
122   - } else {
123   - logger.info("接收到消息：" + cmd);
124   - }
125   - } catch (DocumentException e) {
126   - logger.error("处理NOTIFY消息时错误", e);
  99 + }catch (SipException | InvalidArgumentException | ParseException e) {
  100 + e.printStackTrace();
  101 + }
  102 + taskQueue.offer(new HandlerCatchData(evt, null, null));
  103 + if (!taskQueueHandlerRun) {
  104 + taskQueueHandlerRun = true;
  105 + taskExecutor.execute(()-> {
  106 + while (!taskQueue.isEmpty()) {
  107 + try {
  108 + HandlerCatchData take = taskQueue.poll();
  109 + Element rootElement = getRootElement(take.getEvt());
  110 + if (rootElement == null) {
  111 + logger.error("处理NOTIFY消息时未获取到消息体,{}", take.getEvt().getRequest());
  112 + continue;
  113 + }
  114 + String cmd = XmlUtil.getText(rootElement, "CmdType");
  115 +
  116 + if (CmdType.CATALOG.equals(cmd)) {
  117 + logger.info("接收到Catalog通知");
  118 + processNotifyCatalogList(take.getEvt());
  119 + } else if (CmdType.ALARM.equals(cmd)) {
  120 + logger.info("接收到Alarm通知");
  121 + processNotifyAlarm(take.getEvt());
  122 + } else if (CmdType.MOBILE_POSITION.equals(cmd)) {
  123 + logger.info("接收到MobilePosition通知");
  124 + processNotifyMobilePosition(take.getEvt());
  125 + } else {
  126 + logger.info("接收到消息：" + cmd);
127 127 }
  128 + } catch (DocumentException e) {
  129 + logger.error("处理NOTIFY消息时错误", e);
128 130 }
129   - taskQueueHandlerRun = false;
130   - });
131   - }
132   - } catch (SipException | InvalidArgumentException | ParseException e) {
133   - e.printStackTrace();
134   - } finally {
135   - taskQueueHandlerRun = false;
  131 + }
  132 + taskQueueHandlerRun = false;
  133 + });
136 134 }
137 135 }
138 136  
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/control/cmd/DeviceControlQueryMessageHandler.java
  View file @4c66f6b
... ... @@ -112,10 +112,10 @@ public class DeviceControlQueryMessageHandler extends SIPRequestProcessorParent
112 112 if (deviceForPlatform == null) {
113 113 try {
114 114 responseAck(serverTransaction, Response.NOT_FOUND);
115   - return;
116 115 } catch (SipException | InvalidArgumentException | ParseException e) {
117 116 logger.error("[命令发送失败] 错误信息: {}", e.getMessage());
118 117 }
  118 + return;
119 119 }
120 120 try {
121 121 cmder.fronEndCmd(deviceForPlatform, channelId, cmdString, eventResult -> {
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/KeepaliveNotifyMessageHandler.java
  View file @4c66f6b
... ... @@ -52,35 +52,36 @@ public class KeepaliveNotifyMessageHandler extends SIPRequestProcessorParent imp
52 52 // 未注册的设备不做处理
53 53 return;
54 54 }
  55 + // 回复200 OK
55 56 try {
56   - // 判断RPort是否改变，改变则说明路由nat信息变化，修改设备信息
57   - // 获取到通信地址等信息
58   - ViaHeader viaHeader = (ViaHeader) evt.getRequest().getHeader(ViaHeader.NAME);
59   - String received = viaHeader.getReceived();
60   - int rPort = viaHeader.getRPort();
61   - // 解析本地地址替代
62   - if (ObjectUtils.isEmpty(received) || rPort == -1) {
63   - received = viaHeader.getHost();
64   - rPort = viaHeader.getPort();
65   - }
66   - if (device.getPort() != rPort) {
67   - device.setPort(rPort);
68   - device.setHostAddress(received.concat(":").concat(String.valueOf(rPort)));
69   - }
70   - device.setKeepaliveTime(DateUtil.getNow());
71   - // 回复200 OK
72 57 responseAck(getServerTransaction(evt), Response.OK);
73   - if (device.getOnline() == 1) {
74   - deviceService.updateDevice(device);
75   - }else {
76   - // 对于已经离线的设备判断他的注册是否已经过期
77   - if (!deviceService.expire(device)){
78   - deviceService.online(device);
79   - }
80   - }
81 58 } catch (SipException | InvalidArgumentException | ParseException e) {
82 59 logger.error("[命令发送失败] 国标级联 心跳回复: {}", e.getMessage());
83 60 }
  61 + // 判断RPort是否改变，改变则说明路由nat信息变化，修改设备信息
  62 + // 获取到通信地址等信息
  63 + ViaHeader viaHeader = (ViaHeader) evt.getRequest().getHeader(ViaHeader.NAME);
  64 + String received = viaHeader.getReceived();
  65 + int rPort = viaHeader.getRPort();
  66 + // 解析本地地址替代
  67 + if (ObjectUtils.isEmpty(received) || rPort == -1) {
  68 + received = viaHeader.getHost();
  69 + rPort = viaHeader.getPort();
  70 + }
  71 + if (device.getPort() != rPort) {
  72 + device.setPort(rPort);
  73 + device.setHostAddress(received.concat(":").concat(String.valueOf(rPort)));
  74 + }
  75 + device.setKeepaliveTime(DateUtil.getNow());
  76 +
  77 + if (device.getOnline() == 1) {
  78 + deviceService.updateDevice(device);
  79 + }else {
  80 + // 对于已经离线的设备判断他的注册是否已经过期
  81 + if (!deviceService.expire(device)){
  82 + deviceService.online(device);
  83 + }
  84 + }
84 85 }
85 86  
86 87 @Override
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/MobilePositionNotifyMessageHandler.java
  View file @4c66f6b
... ... @@ -81,8 +81,12 @@ public class MobilePositionNotifyMessageHandler extends SIPRequestProcessorParen
81 81 try {
82 82 Element rootElementAfterCharset = getRootElement(sipMsgInfo.getEvt(), sipMsgInfo.getDevice().getCharset());
83 83 if (rootElementAfterCharset == null) {
84   - logger.warn("[ 移动设备位置数据通知 ] content cannot be null, {}", sipMsgInfo.getEvt().getRequest());
85   - responseAck(getServerTransaction(sipMsgInfo.getEvt()), Response.BAD_REQUEST);
  84 + try {
  85 + logger.warn("[ 移动设备位置数据通知 ] content cannot be null, {}", sipMsgInfo.getEvt().getRequest());
  86 + responseAck(getServerTransaction(sipMsgInfo.getEvt()), Response.BAD_REQUEST);
  87 + } catch (SipException | InvalidArgumentException | ParseException e) {
  88 + logger.error("[命令发送失败] 移动设备位置数据通知 内容为空: {}", e.getMessage());
  89 + }
86 90 continue;
87 91 }
88 92 MobilePosition mobilePosition = new MobilePosition();
... ... @@ -133,7 +137,11 @@ public class MobilePositionNotifyMessageHandler extends SIPRequestProcessorParen
133 137 }
134 138 storager.updateChannelPosition(deviceChannel);
135 139 //回复 200 OK
136   - responseAck(getServerTransaction(sipMsgInfo.getEvt()), Response.OK);
  140 + try {
  141 + responseAck(getServerTransaction(sipMsgInfo.getEvt()), Response.OK);
  142 + } catch (SipException | InvalidArgumentException | ParseException e) {
  143 + logger.error("[命令发送失败] 移动设备位置数据回复200: {}", e.getMessage());
  144 + }
137 145  
138 146 // 发送redis消息。 通知位置信息的变化
139 147 JSONObject jsonObject = new JSONObject();
... ... @@ -147,7 +155,7 @@ public class MobilePositionNotifyMessageHandler extends SIPRequestProcessorParen
147 155 jsonObject.put("speed", mobilePosition.getSpeed());
148 156 redisCatchStorage.sendMobilePositionMsg(jsonObject);
149 157  
150   - } catch (DocumentException | SipException | InvalidArgumentException | ParseException e) {
  158 + } catch (DocumentException e) {
151 159 e.printStackTrace();
152 160 }
153 161  
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/query/cmd/CatalogQueryMessageHandler.java
  View file @4c66f6b
... ... @@ -67,33 +67,37 @@ public class CatalogQueryMessageHandler extends SIPRequestProcessorParent implem
67 67 try {
68 68 // 回复200 OK
69 69 responseAck(getServerTransaction(evt), Response.OK);
70   - Element snElement = rootElement.element("SN");
71   - String sn = snElement.getText();
72   - // 准备回复通道信息
73   - List<DeviceChannel> deviceChannelInPlatforms = storager.queryChannelWithCatalog(parentPlatform.getServerGBId());
74   - // 查询关联的直播通道
75   - List<DeviceChannel> gbStreams = storager.queryGbStreamListInPlatform(parentPlatform.getServerGBId());
76   - // 回复目录信息
77   - List<DeviceChannel> catalogs = storager.queryCatalogInPlatform(parentPlatform.getServerGBId());
78   -
79   - List<DeviceChannel> allChannels = new ArrayList<>();
80   -
81   - // 回复平台
  70 + } catch (SipException | InvalidArgumentException | ParseException e) {
  71 + logger.error("[命令发送失败] 国标级联 目录查询回复200OK: {}", e.getMessage());
  72 + }
  73 + Element snElement = rootElement.element("SN");
  74 + String sn = snElement.getText();
  75 + // 准备回复通道信息
  76 + List<DeviceChannel> deviceChannelInPlatforms = storager.queryChannelWithCatalog(parentPlatform.getServerGBId());
  77 + // 查询关联的直播通道
  78 + List<DeviceChannel> gbStreams = storager.queryGbStreamListInPlatform(parentPlatform.getServerGBId());
  79 + // 回复目录信息
  80 + List<DeviceChannel> catalogs = storager.queryCatalogInPlatform(parentPlatform.getServerGBId());
  81 +
  82 + List<DeviceChannel> allChannels = new ArrayList<>();
  83 +
  84 + // 回复平台
82 85 // DeviceChannel deviceChannel = getChannelForPlatform(parentPlatform);
83 86 // allChannels.add(deviceChannel);
84 87  
85   - // 回复目录
86   - if (catalogs.size() > 0) {
87   - allChannels.addAll(catalogs);
88   - }
89   - // 回复级联的通道
90   - if (deviceChannelInPlatforms.size() > 0) {
91   - allChannels.addAll(deviceChannelInPlatforms);
92   - }
93   - // 回复直播的通道
94   - if (gbStreams.size() > 0) {
95   - allChannels.addAll(gbStreams);
96   - }
  88 + // 回复目录
  89 + if (catalogs.size() > 0) {
  90 + allChannels.addAll(catalogs);
  91 + }
  92 + // 回复级联的通道
  93 + if (deviceChannelInPlatforms.size() > 0) {
  94 + allChannels.addAll(deviceChannelInPlatforms);
  95 + }
  96 + // 回复直播的通道
  97 + if (gbStreams.size() > 0) {
  98 + allChannels.addAll(gbStreams);
  99 + }
  100 + try {
97 101 if (allChannels.size() > 0) {
98 102 cmderFroPlatform.catalogQuery(allChannels, parentPlatform, sn, fromHeader.getTag());
99 103 }else {
... ... @@ -101,9 +105,11 @@ public class CatalogQueryMessageHandler extends SIPRequestProcessorParent implem
101 105 cmderFroPlatform.catalogQuery(null, parentPlatform, sn, fromHeader.getTag(), 0);
102 106 }
103 107 } catch (SipException | InvalidArgumentException | ParseException e) {
104   - logger.error("[命令发送失败] 国标级联 目录查询: {}", e.getMessage());
  108 + logger.error("[命令发送失败] 国标级联 目录查询回复: {}", e.getMessage());
105 109 }
106 110  
  111 +
  112 +
107 113 }
108 114  
109 115 private DeviceChannel getChannelForPlatform(ParentPlatform platform) {
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/ConfigDownloadResponseMessageHandler.java
  View file @4c66f6b
... ... @@ -53,19 +53,20 @@ public class ConfigDownloadResponseMessageHandler extends SIPRequestProcessorPar
53 53 try {
54 54 // 回复200 OK
55 55 responseAck(getServerTransaction(evt), Response.OK);
56   - // 此处是对本平台发出DeviceControl指令的应答
57   - JSONObject json = new JSONObject();
58   - XmlUtil.node2Json(element, json);
59   - if (logger.isDebugEnabled()) {
60   - logger.debug(json.toJSONString());
61   - }
62   - RequestMessage msg = new RequestMessage();
63   - msg.setKey(key);
64   - msg.setData(json);
65   - deferredResultHolder.invokeAllResult(msg);
66 56 } catch (SipException | InvalidArgumentException | ParseException e) {
67   - logger.error("[命令发送失败] 国标级联 设备配置查询: {}", e.getMessage());
  57 + logger.error("[命令发送失败] 设备配置查询: {}", e.getMessage());
68 58 }
  59 + // 此处是对本平台发出DeviceControl指令的应答
  60 + JSONObject json = new JSONObject();
  61 + XmlUtil.node2Json(element, json);
  62 + if (logger.isDebugEnabled()) {
  63 + logger.debug(json.toJSONString());
  64 + }
  65 + RequestMessage msg = new RequestMessage();
  66 + msg.setKey(key);
  67 + msg.setData(json);
  68 + deferredResultHolder.invokeAllResult(msg);
  69 +
69 70  
70 71 }
71 72  
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/DeviceControlResponseMessageHandler.java
  View file @4c66f6b
... ... @@ -47,20 +47,21 @@ public class DeviceControlResponseMessageHandler extends SIPRequestProcessorPare
47 47 // 此处是对本平台发出DeviceControl指令的应答
48 48 try {
49 49 responseAck(getServerTransaction(evt), Response.OK);
50   - JSONObject json = new JSONObject();
51   - String channelId = getText(element, "DeviceID");
52   - XmlUtil.node2Json(element, json);
53   - if (logger.isDebugEnabled()) {
54   - logger.debug(json.toJSONString());
55   - }
56   - RequestMessage msg = new RequestMessage();
57   - String key = DeferredResultHolder.CALLBACK_CMD_DEVICECONTROL + device.getDeviceId() + channelId;
58   - msg.setKey(key);
59   - msg.setData(json);
60   - deferredResultHolder.invokeAllResult(msg);
61 50 } catch (SipException | InvalidArgumentException | ParseException e) {
62 51 logger.error("[命令发送失败] 国标级联 设备控制: {}", e.getMessage());
63 52 }
  53 + JSONObject json = new JSONObject();
  54 + String channelId = getText(element, "DeviceID");
  55 + XmlUtil.node2Json(element, json);
  56 + if (logger.isDebugEnabled()) {
  57 + logger.debug(json.toJSONString());
  58 + }
  59 + RequestMessage msg = new RequestMessage();
  60 + String key = DeferredResultHolder.CALLBACK_CMD_DEVICECONTROL + device.getDeviceId() + channelId;
  61 + msg.setKey(key);
  62 + msg.setData(json);
  63 + deferredResultHolder.invokeAllResult(msg);
  64 +
64 65 }
65 66  
66 67 @Override
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/DeviceInfoResponseMessageHandler.java
  View file @4c66f6b
... ... @@ -78,9 +78,14 @@ public class DeviceInfoResponseMessageHandler extends SIPRequestProcessorParent
78 78 ServerTransaction serverTransaction = getServerTransaction(evt);
79 79 try {
80 80 rootElement = getRootElement(evt, device.getCharset());
81   - if (rootElement == null) {
  81 +
  82 + if (rootElement == null) {
82 83 logger.warn("[ 接收到DeviceInfo应答消息 ] content cannot be null, {}", evt.getRequest());
83   - responseAck(serverTransaction, Response.BAD_REQUEST);
  84 + try {
  85 + responseAck(serverTransaction, Response.BAD_REQUEST);
  86 + } catch (SipException | InvalidArgumentException | ParseException e) {
  87 + logger.error("[命令发送失败] DeviceInfo应答消息 BAD_REQUEST: {}", e.getMessage());
  88 + }
84 89 return;
85 90 }
86 91 Element deviceIdElement = rootElement.element("DeviceID");
... ... @@ -100,17 +105,16 @@ public class DeviceInfoResponseMessageHandler extends SIPRequestProcessorParent
100 105 msg.setKey(key);
101 106 msg.setData(device);
102 107 deferredResultHolder.invokeAllResult(msg);
  108 + } catch (DocumentException e) {
  109 + throw new RuntimeException(e);
  110 + }
  111 + try {
103 112 // 回复200 OK
104 113 responseAck(serverTransaction, Response.OK);
105   - } catch (DocumentException e) {
106   - e.printStackTrace();
107   - } catch (InvalidArgumentException e) {
108   - e.printStackTrace();
109   - } catch (ParseException e) {
110   - e.printStackTrace();
111   - } catch (SipException e) {
112   - e.printStackTrace();
  114 + } catch (SipException | InvalidArgumentException | ParseException e) {
  115 + logger.error("[命令发送失败] DeviceInfo应答消息 200: {}", e.getMessage());
113 116 }
  117 +
114 118 }
115 119  
116 120 @Override
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/MobilePositionResponseMessageHandler.java
  View file @4c66f6b
... ... @@ -71,7 +71,11 @@ public class MobilePositionResponseMessageHandler extends SIPRequestProcessorPar
71 71 rootElement = getRootElement(evt, device.getCharset());
72 72 if (rootElement == null) {
73 73 logger.warn("[ 移动设备位置数据查询回复 ] content cannot be null, {}", evt.getRequest());
74   - responseAck(serverTransaction, Response.BAD_REQUEST);
  74 + try {
  75 + responseAck(serverTransaction, Response.BAD_REQUEST);
  76 + } catch (SipException | InvalidArgumentException | ParseException e) {
  77 + logger.error("[命令发送失败] 移动设备位置数据查询 BAD_REQUEST: {}", e.getMessage());
  78 + }
75 79 return;
76 80 }
77 81 MobilePosition mobilePosition = new MobilePosition();
... ... @@ -133,8 +137,13 @@ public class MobilePositionResponseMessageHandler extends SIPRequestProcessorPar
133 137 jsonObject.put("speed", mobilePosition.getSpeed());
134 138 redisCatchStorage.sendMobilePositionMsg(jsonObject);
135 139 //回复 200 OK
136   - responseAck(serverTransaction, Response.OK);
137   - } catch (DocumentException | SipException | InvalidArgumentException | ParseException e) {
  140 + try {
  141 + responseAck(serverTransaction, Response.OK);
  142 + } catch (SipException | InvalidArgumentException | ParseException e) {
  143 + logger.error("[命令发送失败] 移动设备位置数据查询 200: {}", e.getMessage());
  144 + }
  145 +
  146 + } catch (DocumentException e) {
138 147 e.printStackTrace();
139 148 }
140 149 }
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/PresetQueryResponseMessageHandler.java
  View file @4c66f6b
... ... @@ -58,7 +58,11 @@ public class PresetQueryResponseMessageHandler extends SIPRequestProcessorParent
58 58  
59 59 if (rootElement == null) {
60 60 logger.warn("[ 设备预置位查询应答 ] content cannot be null, {}", evt.getRequest());
61   - responseAck(serverTransaction, Response.BAD_REQUEST);
  61 + try {
  62 + responseAck(serverTransaction, Response.BAD_REQUEST);
  63 + } catch (InvalidArgumentException | ParseException | SipException e) {
  64 + logger.error("[命令发送失败] 设备预置位查询应答处理: {}", e.getMessage());
  65 + }
62 66 return;
63 67 }
64 68 Element presetListNumElement = rootElement.element("PresetList");
... ... @@ -67,7 +71,11 @@ public class PresetQueryResponseMessageHandler extends SIPRequestProcessorParent
67 71 String deviceId = getText(rootElement, "DeviceID");
68 72 String key = DeferredResultHolder.CALLBACK_CMD_PRESETQUERY + deviceId;
69 73 if (snElement == null || presetListNumElement == null) {
70   - responseAck(serverTransaction, Response.BAD_REQUEST, "xml error");
  74 + try {
  75 + responseAck(serverTransaction, Response.BAD_REQUEST, "xml error");
  76 + } catch (InvalidArgumentException | ParseException | SipException e) {
  77 + logger.error("[命令发送失败] 设备预置位查询应答处理: {}", e.getMessage());
  78 + }
71 79 return;
72 80 }
73 81 int sumNum = Integer.parseInt(presetListNumElement.attributeValue("Num"));
... ... @@ -94,11 +102,13 @@ public class PresetQueryResponseMessageHandler extends SIPRequestProcessorParent
94 102 requestMessage.setKey(key);
95 103 requestMessage.setData(presetQuerySipReqList);
96 104 deferredResultHolder.invokeAllResult(requestMessage);
97   - responseAck(serverTransaction, Response.OK);
  105 + try {
  106 + responseAck(serverTransaction, Response.OK);
  107 + } catch (InvalidArgumentException | ParseException | SipException e) {
  108 + logger.error("[命令发送失败] 设备预置位查询应答处理: {}", e.getMessage());
  109 + }
98 110 } catch (DocumentException e) {
99 111 logger.error("[解析xml]失败: ", e);
100   - } catch (InvalidArgumentException | ParseException | SipException e) {
101   - logger.error("[命令发送失败] 设备预置位查询应答处理: {}", e.getMessage());
102 112 }
103 113 }
104 114  
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/RecordInfoResponseMessageHandler.java
  View file @4c66f6b
... ... @@ -69,95 +69,91 @@ public class RecordInfoResponseMessageHandler extends SIPRequestProcessorParent
69 69  
70 70 @Override
71 71 public void handForDevice(RequestEvent evt, Device device, Element rootElement) {
72   -
73   - // 回复200 OK
74 72 try {
  73 + // 回复200 OK
75 74 responseAck(getServerTransaction(evt), Response.OK);
76   - taskQueue.offer(new HandlerCatchData(evt, device, rootElement));
77   - if (!taskQueueHandlerRun) {
78   - taskQueueHandlerRun = true;
79   - taskExecutor.execute(()->{
80   - while (!taskQueue.isEmpty()) {
81   - try {
82   - HandlerCatchData take = taskQueue.poll();
83   - Element rootElementForCharset = getRootElement(take.getEvt(), take.getDevice().getCharset());
84   - if (rootElement == null) {
85   - logger.warn("[ 国标录像 ] content cannot be null, {}", evt.getRequest());
86   - continue;
87   - }
88   - String sn = getText(rootElementForCharset, "SN");
89   - String channelId = getText(rootElementForCharset, "DeviceID");
90   - RecordInfo recordInfo = new RecordInfo();
91   - recordInfo.setChannelId(channelId);
92   - recordInfo.setDeviceId(take.getDevice().getDeviceId());
93   - recordInfo.setSn(sn);
94   - recordInfo.setName(getText(rootElementForCharset, "Name"));
95   - String sumNumStr = getText(rootElementForCharset, "SumNum");
96   - int sumNum = 0;
97   - if (!ObjectUtils.isEmpty(sumNumStr)) {
98   - sumNum = Integer.parseInt(sumNumStr);
99   - }
100   - recordInfo.setSumNum(sumNum);
101   - Element recordListElement = rootElementForCharset.element("RecordList");
102   - if (recordListElement == null || sumNum == 0) {
103   - logger.info("无录像数据");
104   - eventPublisher.recordEndEventPush(recordInfo);
105   - recordDataCatch.put(take.getDevice().getDeviceId(), sn, sumNum, new ArrayList<>());
106   - releaseRequest(take.getDevice().getDeviceId(), sn);
107   - } else {
108   - Iterator<Element> recordListIterator = recordListElement.elementIterator();
109   - if (recordListIterator != null) {
110   - List<RecordItem> recordList = new ArrayList<>();
111   - // 遍历DeviceList
112   - while (recordListIterator.hasNext()) {
113   - Element itemRecord = recordListIterator.next();
114   - Element recordElement = itemRecord.element("DeviceID");
115   - if (recordElement == null) {
116   - logger.info("记录为空，下一个...");
117   - continue;
118   - }
119   - RecordItem record = new RecordItem();
120   - record.setDeviceId(getText(itemRecord, "DeviceID"));
121   - record.setName(getText(itemRecord, "Name"));
122   - record.setFilePath(getText(itemRecord, "FilePath"));
123   - record.setFileSize(getText(itemRecord, "FileSize"));
124   - record.setAddress(getText(itemRecord, "Address"));
125   -
126   - String startTimeStr = getText(itemRecord, "StartTime");
127   - record.setStartTime(DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(startTimeStr));
128   -
129   - String endTimeStr = getText(itemRecord, "EndTime");
130   - record.setEndTime(DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(endTimeStr));
131   -
132   - record.setSecrecy(itemRecord.element("Secrecy") == null ? 0
133   - : Integer.parseInt(getText(itemRecord, "Secrecy")));
134   - record.setType(getText(itemRecord, "Type"));
135   - record.setRecorderId(getText(itemRecord, "RecorderID"));
136   - recordList.add(record);
  75 + }catch (SipException | InvalidArgumentException | ParseException e) {
  76 + logger.error("[命令发送失败] 国标级联 国标录像: {}", e.getMessage());
  77 + }
  78 + taskQueue.offer(new HandlerCatchData(evt, device, rootElement));
  79 + if (!taskQueueHandlerRun) {
  80 + taskQueueHandlerRun = true;
  81 + taskExecutor.execute(()->{
  82 + while (!taskQueue.isEmpty()) {
  83 + try {
  84 + HandlerCatchData take = taskQueue.poll();
  85 + Element rootElementForCharset = getRootElement(take.getEvt(), take.getDevice().getCharset());
  86 + if (rootElement == null) {
  87 + logger.warn("[ 国标录像 ] content cannot be null, {}", evt.getRequest());
  88 + continue;
  89 + }
  90 + String sn = getText(rootElementForCharset, "SN");
  91 + String channelId = getText(rootElementForCharset, "DeviceID");
  92 + RecordInfo recordInfo = new RecordInfo();
  93 + recordInfo.setChannelId(channelId);
  94 + recordInfo.setDeviceId(take.getDevice().getDeviceId());
  95 + recordInfo.setSn(sn);
  96 + recordInfo.setName(getText(rootElementForCharset, "Name"));
  97 + String sumNumStr = getText(rootElementForCharset, "SumNum");
  98 + int sumNum = 0;
  99 + if (!ObjectUtils.isEmpty(sumNumStr)) {
  100 + sumNum = Integer.parseInt(sumNumStr);
  101 + }
  102 + recordInfo.setSumNum(sumNum);
  103 + Element recordListElement = rootElementForCharset.element("RecordList");
  104 + if (recordListElement == null || sumNum == 0) {
  105 + logger.info("无录像数据");
  106 + eventPublisher.recordEndEventPush(recordInfo);
  107 + recordDataCatch.put(take.getDevice().getDeviceId(), sn, sumNum, new ArrayList<>());
  108 + releaseRequest(take.getDevice().getDeviceId(), sn);
  109 + } else {
  110 + Iterator<Element> recordListIterator = recordListElement.elementIterator();
  111 + if (recordListIterator != null) {
  112 + List<RecordItem> recordList = new ArrayList<>();
  113 + // 遍历DeviceList
  114 + while (recordListIterator.hasNext()) {
  115 + Element itemRecord = recordListIterator.next();
  116 + Element recordElement = itemRecord.element("DeviceID");
  117 + if (recordElement == null) {
  118 + logger.info("记录为空，下一个...");
  119 + continue;
137 120 }
138   - recordInfo.setRecordList(recordList);
139   - // 发送消息，如果是上级查询此录像，则会通过这里通知给上级
140   - eventPublisher.recordEndEventPush(recordInfo);
141   - int count = recordDataCatch.put(take.getDevice().getDeviceId(), sn, sumNum, recordList);
142   - logger.info("[国标录像]， {}->{}: {}/{}", take.getDevice().getDeviceId(), sn, count, sumNum);
  121 + RecordItem record = new RecordItem();
  122 + record.setDeviceId(getText(itemRecord, "DeviceID"));
  123 + record.setName(getText(itemRecord, "Name"));
  124 + record.setFilePath(getText(itemRecord, "FilePath"));
  125 + record.setFileSize(getText(itemRecord, "FileSize"));
  126 + record.setAddress(getText(itemRecord, "Address"));
  127 +
  128 + String startTimeStr = getText(itemRecord, "StartTime");
  129 + record.setStartTime(DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(startTimeStr));
  130 +
  131 + String endTimeStr = getText(itemRecord, "EndTime");
  132 + record.setEndTime(DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(endTimeStr));
  133 +
  134 + record.setSecrecy(itemRecord.element("Secrecy") == null ? 0
  135 + : Integer.parseInt(getText(itemRecord, "Secrecy")));
  136 + record.setType(getText(itemRecord, "Type"));
  137 + record.setRecorderId(getText(itemRecord, "RecorderID"));
  138 + recordList.add(record);
143 139 }
  140 + recordInfo.setRecordList(recordList);
  141 + // 发送消息，如果是上级查询此录像，则会通过这里通知给上级
  142 + eventPublisher.recordEndEventPush(recordInfo);
  143 + int count = recordDataCatch.put(take.getDevice().getDeviceId(), sn, sumNum, recordList);
  144 + logger.info("[国标录像]， {}->{}: {}/{}", take.getDevice().getDeviceId(), sn, count, sumNum);
  145 + }
144 146  
145   - if (recordDataCatch.isComplete(take.getDevice().getDeviceId(), sn)){
146   - releaseRequest(take.getDevice().getDeviceId(), sn);
147   - }
  147 + if (recordDataCatch.isComplete(take.getDevice().getDeviceId(), sn)){
  148 + releaseRequest(take.getDevice().getDeviceId(), sn);
148 149 }
149   - } catch (DocumentException e) {
150   - logger.error("xml解析异常： ", e);
151 150 }
  151 + } catch (DocumentException e) {
  152 + logger.error("xml解析异常： ", e);
152 153 }
153   - taskQueueHandlerRun = false;
154   - });
155   - }
156   -
157   - } catch (SipException | InvalidArgumentException | ParseException e) {
158   - logger.error("[命令发送失败] 国标级联 国标录像: {}", e.getMessage());
159   - } finally {
160   - taskQueueHandlerRun = false;
  154 + }
  155 + taskQueueHandlerRun = false;
  156 + });
161 157 }
162 158 }
163 159  
... ...