From b63a89a0a83dfddab0d714bb9aad90114ab9d514 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期五, 18 八月 2023 14:06:39 +0800
Subject: [PATCH] 修复发流参数错误

---
 src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java |  163 +++++++++++++++++++++++++++++++++---------------------
 1 files changed, 99 insertions(+), 64 deletions(-)

diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
index ca5bb13..acc9548 100644
--- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
+++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
@@ -3,10 +3,12 @@
 import com.alibaba.fastjson2.JSON;
 import com.alibaba.fastjson2.JSONObject;
 import com.genersoft.iot.vmp.common.StreamInfo;
+import com.genersoft.iot.vmp.common.VideoManagerConstants;
 import com.genersoft.iot.vmp.conf.UserSetting;
 import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;
 import com.genersoft.iot.vmp.gb28181.bean.*;
 import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
+import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
 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;
@@ -18,16 +20,15 @@
 import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
 import com.genersoft.iot.vmp.media.zlm.dto.hook.*;
 import com.genersoft.iot.vmp.service.*;
+import com.genersoft.iot.vmp.service.bean.MessageForPushChannel;
 import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
 import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
-import com.genersoft.iot.vmp.vmanager.bean.DeferredResultEx;
-import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
-import com.genersoft.iot.vmp.vmanager.bean.StreamContent;
-import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
+import com.genersoft.iot.vmp.vmanager.bean.*;
 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.data.redis.core.RedisTemplate;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.util.ObjectUtils;
 import org.springframework.web.bind.annotation.*;
@@ -104,6 +105,9 @@
     @Autowired
     private AssistRESTfulUtils assistRESTfulUtils;
 
+    @Autowired
+    private RedisTemplate<Object, Object> redisTemplate;
+
     @Qualifier("taskExecutor")
     @Autowired
     private ThreadPoolTaskExecutor taskExecutor;
@@ -116,7 +120,7 @@
     @PostMapping(value = "/on_server_keepalive", produces = "application/json;charset=UTF-8")
     public HookResult onServerKeepalive(@RequestBody OnServerKeepaliveHookParam param) {
 
-        logger.info("[ZLM HOOK] 鏀跺埌zlm蹇冭烦锛�" + param.getMediaServerId());
+//        logger.info("[ZLM HOOK] 鏀跺埌zlm蹇冭烦锛�" + param.getMediaServerId());
 
         taskExecutor.execute(() -> {
             List<ZlmHttpHookSubscribe.Event> subscribes = this.subscribe.getSubscribes(HookType.on_server_keepalive);
@@ -218,10 +222,7 @@
 
 
         HookResultForOnPublish result = HookResultForOnPublish.SUCCESS();
-        if (!"rtp".equals(param.getApp())) {
-            result.setEnable_audio(true);
-        }
-
+        result.setEnable_audio(true);
         taskExecutor.execute(() -> {
             ZlmHttpHookSubscribe.Event subscribe = this.subscribe.sendNotify(HookType.on_publish, json);
             if (subscribe != null) {
@@ -249,10 +250,10 @@
             // 濡傛灉鏄綍鍍忎笅杞藉氨璁剧疆瑙嗛闂撮殧鍗佺
             if (ssrcTransactionForAll.get(0).getType() == VideoStreamSessionManager.SessionType.download) {
                 result.setMp4_max_second(10);
-                result.setEnable_audio(true);
                 result.setEnable_mp4(true);
             }
         }
+
         if (mediaInfo.getRecordAssistPort() > 0 && userSetting.getRecordPath() == null) {
             logger.info("鎺ㄦ祦鏃跺彂鐜板皻鏈缃綍鍍忚矾寰勶紝浠巃ssist鏈嶅姟涓鍙�");
             JSONObject info = assistRESTfulUtils.getInfo(mediaInfo, null);
@@ -271,6 +272,17 @@
                 }
             }
         }
+        if (param.getApp().equalsIgnoreCase("rtp")) {
+            String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_RTP_INFO + userSetting.getServerId() + "_" + param.getStream();
+            OtherRtpSendInfo otherRtpSendInfo = (OtherRtpSendInfo)redisTemplate.opsForValue().get(receiveKey);
+
+            String receiveKeyForPS = VideoManagerConstants.WVP_OTHER_RECEIVE_PS_INFO + userSetting.getServerId() + "_" + param.getStream();
+            OtherPsSendInfo otherPsSendInfo = (OtherPsSendInfo)redisTemplate.opsForValue().get(receiveKeyForPS);
+            if (otherRtpSendInfo != null || otherPsSendInfo != null) {
+                result.setEnable_mp4(true);
+            }
+        }
+        logger.info("[ZLM HOOK]鎺ㄦ祦閴存潈 鍝嶅簲锛歿}->{}->>>>{}", param.getMediaServerId(), param, result);
         return result;
     }
 
@@ -281,7 +293,7 @@
     @ResponseBody
     @PostMapping(value = "/on_stream_changed", produces = "application/json;charset=UTF-8")
     public HookResult onStreamChanged(@RequestBody OnStreamChangedHookParam param) {
-
+        System.out.println(JSON.toJSONString(param));
         if (param.isRegist()) {
             logger.info("[ZLM HOOK] 娴佹敞鍐�, {}->{}->{}/{}", param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream());
         } else {
@@ -292,22 +304,22 @@
         JSONObject json = (JSONObject) JSON.toJSON(param);
         taskExecutor.execute(() -> {
             ZlmHttpHookSubscribe.Event subscribe = this.subscribe.sendNotify(HookType.on_stream_changed, json);
+            MediaServerItem mediaInfo = mediaServerService.getOne(param.getMediaServerId());
+            if (mediaInfo == null) {
+                logger.info("[ZLM HOOK] 娴佸彉鍖栨湭鎵惧埌ZLM, {}", param.getMediaServerId());
+                return;
+            }
             if (subscribe != null) {
-                MediaServerItem mediaInfo = mediaServerService.getOne(param.getMediaServerId());
-                if (mediaInfo != null) {
-                    subscribe.response(mediaInfo, json);
-                }
+                subscribe.response(mediaInfo, json);
             }
 
             List<OnStreamChangedHookParam.MediaTrack> tracks = param.getTracks();
             // TODO 閲嶆瀯姝ゅ閫昏緫
-
             if (param.isRegist()) {
                 // 澶勭悊娴佹敞鍐岀殑閴存潈淇℃伅
                 if (param.getOriginType() == OriginType.RTMP_PUSH.ordinal()
                         || param.getOriginType() == OriginType.RTSP_PUSH.ordinal()
                         || param.getOriginType() == OriginType.RTC_PUSH.ordinal()) {
-
                     StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(param.getApp(), param.getStream());
                     if (streamAuthorityInfo == null) {
                         streamAuthorityInfo = StreamAuthorityInfo.getInstanceByHook(param);
@@ -321,7 +333,7 @@
                 redisCatchStorage.removeStreamAuthorityInfo(param.getApp(), param.getStream());
             }
 
-            if ("rtsp".equals(param.getSchema())) {
+            if ("rtmp".equals(param.getSchema())) {
                 // 鏇存柊娴佸獟浣撹礋杞戒俊鎭�
                 if (param.isRegist()) {
                     mediaServerService.addCount(param.getMediaServerId());
@@ -329,7 +341,10 @@
                     mediaServerService.removeCount(param.getMediaServerId());
                 }
                 // 璁剧疆鎷夋祦浠g悊涓婄嚎/绂荤嚎
-                streamProxyService.updateStatus(param.isRegist(), param.getApp(), param.getStream());
+                int updateStatusResult = streamProxyService.updateStatus(param.isRegist(), param.getApp(), param.getStream());
+                if (updateStatusResult > 0) {
+
+                }
 
                 if ("rtp".equals(param.getApp()) && !param.isRegist()) {
                     StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(param.getStream());
@@ -337,7 +352,8 @@
                         redisCatchStorage.stopPlay(streamInfo);
                         storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
                     } else {
-                        streamInfo = redisCatchStorage.queryPlayback(null, null, param.getStream(), null);
+                        streamInfo = redisCatchStorage.queryPlayback(null, null,
+                                param.getStream(), null);
                         if (streamInfo != null) {
                             redisCatchStorage.stopPlayback(streamInfo.getDeviceID(), streamInfo.getChannelId(),
                                     streamInfo.getStream(), null);
@@ -346,48 +362,54 @@
                 } else {
                     if (!"rtp".equals(param.getApp())) {
                         String type = OriginType.values()[param.getOriginType()].getType();
-                        MediaServerItem mediaServerItem = mediaServerService.getOne(param.getMediaServerId());
+                        if (param.isRegist()) {
+                            StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(
+                                    param.getApp(), param.getStream());
+                            String callId = null;
+                            if (streamAuthorityInfo != null) {
+                                callId = streamAuthorityInfo.getCallId();
+                            }
+                            StreamInfo streamInfoByAppAndStream = mediaService.getStreamInfoByAppAndStream(mediaInfo,
+                                    param.getApp(), param.getStream(), tracks, callId);
+                            param.setStreamInfo(new StreamContent(streamInfoByAppAndStream));
+                            // 濡傛灉鏄媺娴佷唬鐞嗕骇鐢熺殑锛屼笉闇�瑕佸啓鍏ユ帹娴�
 
-                        if (mediaServerItem != null) {
-                            if (param.isRegist()) {
-                                StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(param.getApp(), param.getStream());
-                                String callId = null;
-                                if (streamAuthorityInfo != null) {
-                                    callId = streamAuthorityInfo.getCallId();
-                                }
-                                StreamInfo streamInfoByAppAndStream = mediaService.getStreamInfoByAppAndStream(mediaServerItem,
-                                        param.getApp(), param.getStream(), tracks, callId);
-                                param.setStreamInfo(new StreamContent(streamInfoByAppAndStream));
-                                redisCatchStorage.addStream(mediaServerItem, type, param.getApp(), param.getStream(), param);
-                                if (param.getOriginType() == OriginType.RTSP_PUSH.ordinal()
-                                        || param.getOriginType() == OriginType.RTMP_PUSH.ordinal()
-                                        || param.getOriginType() == OriginType.RTC_PUSH.ordinal()) {
-                                    param.setSeverId(userSetting.getServerId());
-                                    zlmMediaListManager.addPush(param);
-                                }
-                            } else {
-                                // 鍏煎娴佹敞閿�鏃剁被鍨嬩粠redis璁板綍鑾峰彇
-                                OnStreamChangedHookParam onStreamChangedHookParam = redisCatchStorage.getStreamInfo(param.getApp(), param.getStream(), param.getMediaServerId());
-                                if (onStreamChangedHookParam != null) {
-                                    type = OriginType.values()[onStreamChangedHookParam.getOriginType()].getType();
-                                    redisCatchStorage.removeStream(mediaServerItem.getId(), type, param.getApp(), param.getStream());
-                                }
-                                GbStream gbStream = storager.getGbStream(param.getApp(), param.getStream());
-                                if (gbStream != null) {
+                            redisCatchStorage.addStream(mediaInfo, type, param.getApp(), param.getStream(), param);
+                            if (param.getOriginType() == OriginType.RTSP_PUSH.ordinal()
+                                    || param.getOriginType() == OriginType.RTMP_PUSH.ordinal()
+                                    || param.getOriginType() == OriginType.RTC_PUSH.ordinal()) {
+                                param.setSeverId(userSetting.getServerId());
+                                zlmMediaListManager.addPush(param);
+                            }
+                        } else {
+                            // 鍏煎娴佹敞閿�鏃剁被鍨嬩粠redis璁板綍鑾峰彇
+                            OnStreamChangedHookParam onStreamChangedHookParam = redisCatchStorage.getStreamInfo(
+                                    param.getApp(), param.getStream(), param.getMediaServerId());
+                            if (onStreamChangedHookParam != null) {
+                                type = OriginType.values()[onStreamChangedHookParam.getOriginType()].getType();
+                                redisCatchStorage.removeStream(mediaInfo.getId(), type, param.getApp(), param.getStream());
+                            }
+                            GbStream gbStream = storager.getGbStream(param.getApp(), param.getStream());
+                            if (gbStream != null) {
 //									eventPublisher.catalogEventPublishForStream(null, gbStream, CatalogEvent.OFF);
-                                }
-                                zlmMediaListManager.removeMedia(param.getApp(), param.getStream());
                             }
-                            if (type != null) {
-                                // 鍙戦�佹祦鍙樺寲redis娑堟伅
-                                JSONObject jsonObject = new JSONObject();
-                                jsonObject.put("serverId", userSetting.getServerId());
-                                jsonObject.put("app", param.getApp());
-                                jsonObject.put("stream", param.getStream());
-                                jsonObject.put("register", param.isRegist());
-                                jsonObject.put("mediaServerId", param.getMediaServerId());
-                                redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
+                            zlmMediaListManager.removeMedia(param.getApp(), param.getStream());
+                        }
+                        GbStream gbStream = storager.getGbStream(param.getApp(), param.getStream());
+                        if (gbStream != null) {
+                            if (userSetting.isUsePushingAsStatus()) {
+                                eventPublisher.catalogEventPublishForStream(null, gbStream, param.isRegist()?CatalogEvent.ON:CatalogEvent.OFF);
                             }
+                        }
+                        if (type != null) {
+                            // 鍙戦�佹祦鍙樺寲redis娑堟伅
+                            JSONObject jsonObject = new JSONObject();
+                            jsonObject.put("serverId", userSetting.getServerId());
+                            jsonObject.put("app", param.getApp());
+                            jsonObject.put("stream", param.getStream());
+                            jsonObject.put("register", param.isRegist());
+                            jsonObject.put("mediaServerId", param.getMediaServerId());
+                            redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
                         }
                     }
                 }
@@ -395,7 +417,7 @@
                     List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByStream(param.getStream());
                     if (sendRtpItems.size() > 0) {
                         for (SendRtpItem sendRtpItem : sendRtpItems) {
-                            if (sendRtpItem.getApp().equals(param.getApp())) {
+                            if (sendRtpItem != null && sendRtpItem.getApp().equals(param.getApp())) {
                                 String platformId = sendRtpItem.getPlatformId();
                                 ParentPlatform platform = storager.queryParentPlatByServerGBId(platformId);
                                 Device device = deviceService.getDevice(platformId);
@@ -403,6 +425,8 @@
                                 try {
                                     if (platform != null) {
                                         commanderFroPlatform.streamByeCmd(platform, sendRtpItem);
+                                        redisCatchStorage.deleteSendRTPServer(platformId, sendRtpItem.getChannelId(),
+                                                sendRtpItem.getCallId(), sendRtpItem.getStreamId());
                                     } else {
                                         cmder.streamByeCmd(device, sendRtpItem.getChannelId(), param.getStream(), sendRtpItem.getCallId());
                                     }
@@ -427,7 +451,8 @@
     @PostMapping(value = "/on_stream_none_reader", produces = "application/json;charset=UTF-8")
     public JSONObject onStreamNoneReader(@RequestBody OnStreamNoneReaderHookParam param) {
 
-        logger.info("[ZLM HOOK]娴佹棤浜鸿鐪嬶細{]->{}->{}/{}" + param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream());
+        logger.info("[ZLM HOOK]娴佹棤浜鸿鐪嬶細{]->{}->{}/{}" + param.getMediaServerId(), param.getSchema(),
+                param.getApp(), param.getStream());
         JSONObject ret = new JSONObject();
         ret.put("code", 0);
         // 鍥芥爣绫诲瀷鐨勬祦
@@ -439,7 +464,8 @@
             if (streamInfoForPlayCatch != null) {
                 // 鏀跺埌鏃犱汉瑙傜湅璇存槑娴佷篃娌℃湁鍦ㄥ線涓婄骇鎺ㄩ��
                 if (redisCatchStorage.isChannelSendingRTP(streamInfoForPlayCatch.getChannelId())) {
-                    List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByChnnelId(streamInfoForPlayCatch.getChannelId());
+                    List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByChnnelId(
+                            streamInfoForPlayCatch.getChannelId());
                     if (sendRtpItems.size() > 0) {
                         for (SendRtpItem sendRtpItem : sendRtpItems) {
                             ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
@@ -450,6 +476,13 @@
                             }
                             redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), sendRtpItem.getChannelId(),
                                     sendRtpItem.getCallId(), sendRtpItem.getStreamId());
+                            if (InviteStreamType.PUSH == sendRtpItem.getPlayType()) {
+                                MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(0,
+                                        sendRtpItem.getApp(), sendRtpItem.getStreamId(), sendRtpItem.getChannelId(),
+                                        sendRtpItem.getPlatformId(), parentPlatform.getName(), userSetting.getServerId(), sendRtpItem.getMediaServerId());
+                                messageForPushChannel.setPlatFormIndex(parentPlatform.getId());
+                                redisCatchStorage.sendPlatformStopPlayMsg(messageForPushChannel);
+                            }
                         }
                     }
                 }
@@ -469,7 +502,8 @@
                 return ret;
             }
             // 褰曞儚鍥炴斁
-            StreamInfo streamInfoForPlayBackCatch = redisCatchStorage.queryPlayback(null, null, param.getStream(), null);
+            StreamInfo streamInfoForPlayBackCatch = redisCatchStorage.queryPlayback(null, null,
+                    param.getStream(), null);
             if (streamInfoForPlayBackCatch != null) {
                 if (streamInfoForPlayBackCatch.isPause()) {
                     ret.put("close", false);
@@ -490,7 +524,8 @@
                 return ret;
             }
             // 褰曞儚涓嬭浇
-            StreamInfo streamInfoForDownload = redisCatchStorage.queryDownload(null, null, param.getStream(), null);
+            StreamInfo streamInfoForDownload = redisCatchStorage.queryDownload(null, null,
+                    param.getStream(), null);
             // 杩涜褰曞儚涓嬭浇鏃舵棤浜鸿鐪嬩笉鏂祦
             if (streamInfoForDownload != null) {
                 ret.put("close", false);
@@ -518,7 +553,7 @@
                 }
                 return ret;
             }
-            // 鎺ㄦ祦鍏锋湁涓诲姩鎬э紝鏆傛椂涓嶅仛澶勭悊
+            // TODO 鎺ㄦ祦鍏锋湁涓诲姩鎬э紝鏆傛椂涓嶅仛澶勭悊
 //			StreamPushItem streamPushItem = streamPushService.getPush(app, streamId);
 //			if (streamPushItem != null) {
 //				// TODO 鍙戦�佸仠姝�

--
Gitblit v1.8.0