From 59bda74e88ec95890cb13d54aefef38e27411c01 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期三, 17 八月 2022 17:06:33 +0800
Subject: [PATCH] 处理获取消息体内容为空时造成的空指针异常

---
 src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java |  193 ++++++++++++++++++++++++++++++-----------------
 1 files changed, 123 insertions(+), 70 deletions(-)

diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
index b164492..0c571e9 100644
--- a/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
@@ -1,18 +1,43 @@
 package com.genersoft.iot.vmp.service.impl;
 
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.util.*;
+
+import javax.sip.ResponseEvent;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.stereotype.Service;
+import org.springframework.web.context.request.async.DeferredResult;
+
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import com.genersoft.iot.vmp.common.StreamInfo;
 import com.genersoft.iot.vmp.conf.DynamicTask;
 import com.genersoft.iot.vmp.conf.UserSetting;
-import com.genersoft.iot.vmp.gb28181.bean.*;
+import com.genersoft.iot.vmp.gb28181.bean.Device;
+import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
+import com.genersoft.iot.vmp.gb28181.bean.InviteStreamCallback;
+import com.genersoft.iot.vmp.gb28181.bean.InviteStreamInfo;
+import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
+import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
+import com.genersoft.iot.vmp.gb28181.bean.SsrcTransaction;
 import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
 import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
 import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
 import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
 import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
 import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
+import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory;
+import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange;
+import com.genersoft.iot.vmp.media.zlm.dto.HookType;
 import com.genersoft.iot.vmp.utils.DateUtil;
 import com.genersoft.iot.vmp.media.zlm.AssistRESTfulUtils;
 import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
@@ -27,23 +52,11 @@
 import com.genersoft.iot.vmp.service.bean.SSRCInfo;
 import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
 import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
+import com.genersoft.iot.vmp.utils.DateUtil;
 import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
 import com.genersoft.iot.vmp.vmanager.gb28181.play.bean.PlayResult;
-import gov.nist.javax.sip.stack.SIPDialog;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.http.HttpStatus;
-import org.springframework.http.ResponseEntity;
-import org.springframework.stereotype.Service;
-import org.springframework.util.ResourceUtils;
-import org.springframework.web.context.request.async.DeferredResult;
 
-import javax.sip.ResponseEvent;
-import java.io.FileNotFoundException;
-import java.math.BigDecimal;
-import java.math.RoundingMode;
-import java.util.*;
+import gov.nist.javax.sip.stack.SIPDialog;
 
 @SuppressWarnings(value = {"rawtypes", "unchecked"})
 @Service
@@ -91,6 +104,10 @@
     private ZLMHttpHookSubscribe subscribe;
 
 
+    @Qualifier("taskExecutor")
+    @Autowired
+    private ThreadPoolTaskExecutor taskExecutor;
+
 
 
     @Override
@@ -122,22 +139,9 @@
 
         result.onCompletion(()->{
             // 鐐规挱缁撴潫鏃惰皟鐢ㄦ埅鍥炬帴鍙�
-            // TODO 搴旇鍦ㄤ笂娴佹椂璋冪敤鏇村ソ锛岀粨鏉熶篃鍙兘鏄敊璇粨鏉�
-            try {
-                String classPath = ResourceUtils.getURL("classpath:").getPath();
-                // 鍏煎鎵撳寘涓簀ar鐨刢lass璺緞
-                if(classPath.contains("jar")) {
-                    classPath = classPath.substring(0, classPath.lastIndexOf("."));
-                    classPath = classPath.substring(0, classPath.lastIndexOf("/") + 1);
-                }
-                if (classPath.startsWith("file:")) {
-                    classPath = classPath.substring(classPath.indexOf(":") + 1);
-                }
-                String path = classPath + "static/static/snap/";
-                // 鍏煎Windows绯荤粺璺緞锛堝幓闄ゅ墠闈㈢殑鈥�/鈥濓級
-                if(System.getProperty("os.name").contains("indows")) {
-                    path = path.substring(1);
-                }
+            taskExecutor.execute(()->{
+                // TODO 搴旇鍦ㄤ笂娴佹椂璋冪敤鏇村ソ锛岀粨鏉熶篃鍙兘鏄敊璇粨鏉�
+                String path =  "snap";
                 String fileName =  deviceId + "_" + channelId + ".jpg";
                 ResponseEntity responseEntity =  (ResponseEntity)result.getResult();
                 if (responseEntity != null && responseEntity.getStatusCode() == HttpStatus.OK) {
@@ -151,9 +155,7 @@
                         zlmresTfulUtils.getSnap(mediaInfo, streamUrl, 15, 1, path, fileName);
                     }
                 }
-            } catch (FileNotFoundException e) {
-                e.printStackTrace();
-            }
+            });
         });
         if (streamInfo != null) {
             String streamId = streamInfo.getStream();
@@ -169,22 +171,30 @@
             MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId);
 
             JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(mediaInfo, streamId);
-            if (rtpInfo != null && rtpInfo.getBoolean("exist")) {
+            if(rtpInfo.getInteger("code") == 0){
+                if (rtpInfo.getBoolean("exist")) {
 
-                WVPResult wvpResult = new WVPResult();
-                wvpResult.setCode(0);
-                wvpResult.setMsg("success");
-                wvpResult.setData(streamInfo);
-                msg.setData(wvpResult);
+                    WVPResult wvpResult = new WVPResult();
+                    wvpResult.setCode(0);
+                    wvpResult.setMsg("success");
+                    wvpResult.setData(streamInfo);
+                    msg.setData(wvpResult);
 
-                resultHolder.invokeAllResult(msg);
-                if (hookEvent != null) {
-                    hookEvent.response(mediaServerItem, JSONObject.parseObject(JSON.toJSONString(streamInfo)));
+                    resultHolder.invokeAllResult(msg);
+                    if (hookEvent != null) {
+                        hookEvent.response(mediaServerItem, JSONObject.parseObject(JSON.toJSONString(streamInfo)));
+                    }
+                }else {
+                    redisCatchStorage.stopPlay(streamInfo);
+                    storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
+                    streamInfo = null;
                 }
             }else {
+                //zlm杩炴帴澶辫触
                 redisCatchStorage.stopPlay(streamInfo);
                 storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
                 streamInfo = null;
+
             }
 
         }
@@ -239,33 +249,41 @@
         if (ssrcInfo == null) {
             ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, device.isSsrcCheck(), false);
         }
-
+        logger.info("[鐐规挱寮�濮媇 deviceId: {}, channelId: {}, SSRC: {}", device.getDeviceId(), channelId, ssrcInfo.getSsrc() );
         // 瓒呮椂澶勭悊
         String timeOutTaskKey = UUID.randomUUID().toString();
         SSRCInfo finalSsrcInfo = ssrcInfo;
         dynamicTask.startDelay( timeOutTaskKey,()->{
-            logger.warn(String.format("璁惧鐐规挱瓒呮椂锛宒eviceId锛�%s 锛宑hannelId锛�%s", device.getDeviceId(), channelId));
 
             SIPDialog dialog = streamSession.getDialogByStream(device.getDeviceId(), channelId, finalSsrcInfo.getStream());
             if (dialog != null) {
+                logger.info("[鐐规挱瓒呮椂] 鏀舵祦瓒呮椂 deviceId: {}, channelId: {}", device.getDeviceId(), channelId);
                 timeoutCallback.run(1, "鏀舵祦瓒呮椂");
                 // 鐐规挱瓒呮椂鍥炲BYE 鍚屾椂閲婃斁ssrc浠ュ強姝ゆ鐐规挱鐨勮祫婧�
                 cmder.streamByeCmd(device.getDeviceId(), channelId, finalSsrcInfo.getStream(), null);
             }else {
+                logger.info("[鐐规挱瓒呮椂] 娑堟伅鏈搷搴� deviceId: {}, channelId: {}", device.getDeviceId(), channelId);
                 timeoutCallback.run(0, "鐐规挱瓒呮椂");
                 mediaServerService.releaseSsrc(mediaServerItem.getId(), finalSsrcInfo.getSsrc());
                 mediaServerService.closeRTPServer(device.getDeviceId(), channelId, finalSsrcInfo.getStream());
                 streamSession.remove(device.getDeviceId(), channelId, finalSsrcInfo.getStream());
             }
-        }, userSetting.getPlayTimeout()*1000);
+        }, userSetting.getPlayTimeout());
         final String ssrc = ssrcInfo.getSsrc();
         final String stream = ssrcInfo.getStream();
+        //绔彛鑾峰彇澶辫触鐨剆srcInfo 娌℃湁蹇呰鍙戦�佺偣鎾寚浠�
+        if(ssrcInfo.getPort() <= 0){
+            logger.info("[鐐规挱绔彛鍒嗛厤寮傚父]锛宒eviceId={},channelId={},ssrcInfo={}", device.getDeviceId(), channelId, ssrcInfo);
+            return;
+        }
         cmder.playStreamCmd(mediaServerItem, ssrcInfo, device, channelId, (MediaServerItem mediaServerItemInuse, JSONObject response) -> {
             logger.info("鏀跺埌璁㈤槄娑堟伅锛� " + response.toJSONString());
             dynamicTask.stop(timeOutTaskKey);
             // hook鍝嶅簲
             onPublishHandlerForPlay(mediaServerItemInuse, response, device.getDeviceId(), channelId, uuid);
             hookEvent.response(mediaServerItemInuse, response);
+            logger.info("[鐐规挱鎴愬姛] deviceId: {}, channelId: {}", device.getDeviceId(), channelId);
+
         }, (event) -> {
             ResponseEvent responseEvent = (ResponseEvent)event.event;
             String contentString = new String(responseEvent.getResponse().getRawContent());
@@ -279,8 +297,10 @@
                 if (ssrc.equals(ssrcInResponse)) {
                     return;
                 }
-                logger.info("[SIP 娑堟伅] 鏀跺埌invite 200, 鍙戠幇涓嬬骇鑷畾涔変簡ssrc 寮�鍚慨姝�");
+                logger.info("[鐐规挱娑堟伅] 鏀跺埌invite 200, 鍙戠幇涓嬬骇鑷畾涔変簡ssrc: {}", ssrcInResponse );
                 if (!mediaServerItem.isRtpEnable() || device.isSsrcCheck()) {
+                    logger.info("[SIP 娑堟伅] SSRC淇 {}->{}", ssrc, ssrcInResponse);
+
                     if (!mediaServerItem.getSsrcConfig().checkSsrc(ssrcInResponse)) {
                         // ssrc 涓嶅彲鐢�
                         // 閲婃斁ssrc
@@ -295,16 +315,10 @@
                     // 鍗曠鍙fā寮弒treamId涔熸湁鍙樺寲锛岄渶瑕侀噸鏂拌缃洃鍚�
                     if (!mediaServerItem.isRtpEnable()) {
                         // 娣诲姞璁㈤槄
-                        JSONObject subscribeKey = new JSONObject();
-                        subscribeKey.put("app", "rtp");
-                        subscribeKey.put("stream", stream);
-                        subscribeKey.put("regist", true);
-                        subscribeKey.put("schema", "rtmp");
-                        subscribeKey.put("mediaServerId", mediaServerItem.getId());
-                        subscribe.removeSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed,subscribeKey);
-                        subscribeKey.put("stream", String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase());
-                        subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,
-                                (MediaServerItem mediaServerItemInUse, JSONObject response)->{
+                        HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", stream, true, "rtmp", mediaServerItem.getId());
+                        subscribe.removeSubscribe(hookSubscribe);
+                        hookSubscribe.getContent().put("stream", String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase());
+                        subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject response)->{
                                     logger.info("[ZLM HOOK] ssrc淇鍚庢敹鍒拌闃呮秷鎭細 " + response.toJSONString());
                                     dynamicTask.stop(timeOutTaskKey);
                                     // hook鍝嶅簲
@@ -315,7 +329,7 @@
                     // 鍏抽棴rtp server
                     mediaServerService.closeRTPServer(device.getDeviceId(), channelId, finalSsrcInfo.getStream());
                     // 閲嶆柊寮�鍚痵src server
-                    mediaServerService.openRTPServer(mediaServerItem, finalSsrcInfo.getStream(), ssrcInResponse, device.isSsrcCheck(), false);
+                    mediaServerService.openRTPServer(mediaServerItem, finalSsrcInfo.getStream(), ssrcInResponse, device.isSsrcCheck(), false, finalSsrcInfo.getPort());
 
                 }
             }
@@ -433,7 +447,7 @@
             cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream(), null);
             // 鍥炲涔嬪墠鎵�鏈夌殑鐐规挱璇锋眰
             playBackCallback.call(playBackResult);
-        }, userSetting.getPlayTimeout()*1000);
+        }, userSetting.getPlayTimeout());
 
         cmder.playbackStreamCmd(mediaServerItem, ssrcInfo, device, channelId, startTime, endTime, infoCallBack,
                 (InviteStreamInfo inviteStreamInfo) -> {
@@ -522,7 +536,7 @@
             cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream(), null);
             // 鍥炲涔嬪墠鎵�鏈夌殑鐐规挱璇锋眰
             hookCallBack.call(downloadResult);
-        }, userSetting.getPlayTimeout()*1000);
+        }, userSetting.getPlayTimeout());
         cmder.downloadStreamCmd(mediaServerItem, ssrcInfo, device, channelId, startTime, endTime, downloadSpeed, infoCallBack,
                 inviteStreamInfo -> {
                     logger.info("鏀跺埌璁㈤槄娑堟伅锛� " + inviteStreamInfo.getResponse().toJSONString());
@@ -530,14 +544,6 @@
                     StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId);
                     streamInfo.setStartTime(startTime);
                     streamInfo.setEndTime(endTime);
-                    if (streamInfo == null) {
-                        logger.warn("褰曞儚涓嬭浇API璋冪敤澶辫触锛�");
-                        wvpResult.setCode(-1);
-                        wvpResult.setMsg("褰曞儚涓嬭浇API璋冪敤澶辫触");
-                        downloadResult.setCode(-1);
-                        hookCallBack.call(downloadResult);
-                        return ;
-                    }
                     redisCatchStorage.startDownload(streamInfo, inviteStreamInfo.getCallId());
                     wvpResult.setCode(0);
                     wvpResult.setMsg("success");
@@ -619,7 +625,7 @@
     public StreamInfo onPublishHandler(MediaServerItem mediaServerItem, JSONObject resonse, String deviceId, String channelId) {
         String streamId = resonse.getString("stream");
         JSONArray tracks = resonse.getJSONArray("tracks");
-        StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(mediaServerItem,"rtp", streamId, tracks);
+        StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(mediaServerItem,"rtp", streamId, tracks, null);
         streamInfo.setDeviceID(deviceId);
         streamInfo.setChannelId(channelId);
         return streamInfo;
@@ -651,6 +657,53 @@
 
     @Override
     public void zlmServerOnline(String mediaServerId) {
-        // 浼间箮娌″暐闇�瑕佸仛鐨�
+        // TODO 鏌ユ壘涔嬪墠鐨勭偣鎾紝娴佸鏋滀笉瀛樺湪鍒欑粰涓嬬骇鍙戦�乥ye
+//        MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
+//        zlmresTfulUtils.getMediaList(mediaServerItem, (mediaList ->{
+//            Integer code = mediaList.getInteger("code");
+//            if (code == 0) {
+//                JSONArray data = mediaList.getJSONArray("data");
+//                if (data == null || data.size() == 0) {
+//                    zlmServerOffline(mediaServerId);
+//                }else {
+//                    Map<String, JSONObject> mediaListMap = new HashMap<>();
+//                    for (int i = 0; i < data.size(); i++) {
+//                        JSONObject json = data.getJSONObject(i);
+//                        String app = json.getString("app");
+//                        if ("rtp".equals(app)) {
+//                            String stream = json.getString("stream");
+//                            if (mediaListMap.get(stream) != null) {
+//                                continue;
+//                            }
+//                            mediaListMap.put(stream, json);
+//                            // 澶勭悊姝e湪瑙傜湅鐨勫浗鏍囪澶�
+//                            List<SsrcTransaction> ssrcTransactions = streamSession.getSsrcTransactionForAll(null, null, null, stream);
+//                            if (ssrcTransactions.size() > 0) {
+//                                for (SsrcTransaction ssrcTransaction : ssrcTransactions) {
+//                                    if(ssrcTransaction.getMediaServerId().equals(mediaServerId)) {
+//                                        cmder.streamByeCmd(ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId(),
+//                                                ssrcTransaction.getStream(), null);
+//                                    }
+//                                }
+//                            }
+//                        }
+//                    }
+//                    if (mediaListMap.size() > 0 ) {
+//                        // 澶勭悊姝e湪鍚戜笂鎺ㄦ祦鐨勪笂绾у钩鍙�
+//                        List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServer(null);
+//                        if (sendRtpItems.size() > 0) {
+//                            for (SendRtpItem sendRtpItem : sendRtpItems) {
+//                                if (sendRtpItem.getMediaServerId().equals(mediaServerId)) {
+//                                    if (mediaListMap.get(sendRtpItem.getStreamId()) == null) {
+//                                        ParentPlatform platform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
+//                                        sipCommanderFroPlatform.streamByeCmd(platform, sendRtpItem.getCallId());
+//                                    }
+//                                }
+//                            }
+//                        }
+//                    }
+//                }
+//            }
+//        }));
     }
 }

--
Gitblit v1.8.0