From ab491a079ba4ab85ffef35d14c0767eba01455d8 Mon Sep 17 00:00:00 2001
From: fuliqi <fuliqi@qq.com>
Date: 星期三, 15 一月 2025 09:18:36 +0800
Subject: [PATCH] 切正式服

---
 src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java |   80 ++++++++++++++++++++++-----------------
 1 files changed, 45 insertions(+), 35 deletions(-)

diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java
index c768ae1..920228d 100755
--- a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java
@@ -7,21 +7,19 @@
 import com.genersoft.iot.vmp.conf.DynamicTask;
 import com.genersoft.iot.vmp.conf.UserSetting;
 import com.genersoft.iot.vmp.conf.exception.ControllerException;
+import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
 import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
 import com.genersoft.iot.vmp.media.bean.MediaInfo;
-import com.genersoft.iot.vmp.media.event.MediaArrivalEvent;
-import com.genersoft.iot.vmp.media.event.MediaDepartureEvent;
-import com.genersoft.iot.vmp.media.event.MediaNotFoundEvent;
-import com.genersoft.iot.vmp.media.service.IMediaServerService;
-import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory;
+import com.genersoft.iot.vmp.media.bean.MediaServer;
+import com.genersoft.iot.vmp.media.event.hook.Hook;
 import com.genersoft.iot.vmp.media.event.hook.HookSubscribe;
-import com.genersoft.iot.vmp.media.event.hook.HookSubscribeFactory;
-import com.genersoft.iot.vmp.media.event.hook.HookSubscribeForStreamChange;
-import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
+import com.genersoft.iot.vmp.media.event.hook.HookType;
+import com.genersoft.iot.vmp.media.event.media.MediaArrivalEvent;
+import com.genersoft.iot.vmp.media.event.media.MediaDepartureEvent;
+import com.genersoft.iot.vmp.media.event.media.MediaNotFoundEvent;
+import com.genersoft.iot.vmp.media.service.IMediaServerService;
 import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
-import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
 import com.genersoft.iot.vmp.service.IGbStreamService;
-import com.genersoft.iot.vmp.service.IMediaService;
 import com.genersoft.iot.vmp.service.IStreamProxyService;
 import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
 import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
@@ -64,12 +62,6 @@
 
     @Autowired
     private IVideoManagerStorage videoManagerStorager;
-
-    @Autowired
-    private IMediaService mediaService;
-
-    @Autowired
-    private ZLMServerFactory zlmServerFactory;
 
     @Autowired
     private StreamProxyMapper streamProxyMapper;
@@ -203,9 +195,9 @@
             callback.run(ErrorCode.ERROR100.getCode(), "淇濆瓨澶辫触", null);
             return;
         }
-        HookSubscribeForStreamChange hookSubscribeForStreamChange = HookSubscribeFactory.on_stream_changed(param.getApp(), param.getStream(), true, "rtsp", mediaServer.getId());
-        hookSubscribe.addSubscribe(hookSubscribeForStreamChange, (mediaServerItem, response) -> {
-            StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(
+        Hook hook = Hook.getInstance(HookType.on_media_arrival, param.getApp(), param.getStream(), mediaServer.getId());
+        hookSubscribe.addSubscribe(hook, (hookData) -> {
+            StreamInfo streamInfo = mediaServerService.getStreamInfoByAppAndStream(
                     mediaServer, param.getApp(), param.getStream(), null, null);
             callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(), streamInfo);
         });
@@ -213,7 +205,7 @@
             String talkKey = UUID.randomUUID().toString();
             String delayTalkKey = UUID.randomUUID().toString();
             dynamicTask.startDelay(delayTalkKey, ()->{
-                StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStreamWithCheck(param.getApp(), param.getStream(), mediaServer.getId(), false);
+                StreamInfo streamInfo = mediaServerService.getStreamInfoByAppAndStreamWithCheck(param.getApp(), param.getStream(), mediaServer.getId(), false);
                 if (streamInfo != null) {
                     callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(), streamInfo);
                 }else {
@@ -223,9 +215,9 @@
             }, 7000);
             WVPResult<String> result = addStreamProxyToZlm(param);
             if (result != null && result.getCode() == 0) {
-                hookSubscribe.removeSubscribe(hookSubscribeForStreamChange);
+                hookSubscribe.removeSubscribe(hook);
                 dynamicTask.stop(talkKey);
-                StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(
+                StreamInfo streamInfo = mediaServerService.getStreamInfoByAppAndStream(
                         mediaServer, param.getApp(), param.getStream(), null, null);
                 callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(), streamInfo);
             }else {
@@ -244,7 +236,7 @@
             }
         }
         else{
-            StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(
+            StreamInfo streamInfo = mediaServerService.getStreamInfoByAppAndStream(
                     mediaServer, param.getApp(), param.getStream(), null, null);
             callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(), streamInfo);
         }
@@ -353,11 +345,14 @@
         if (mediaServer == null) {
             return null;
         }
-        if (zlmServerFactory.isStreamReady(mediaServer, param.getApp(), param.getStream())) {
+        if (mediaServerService.isStreamReady(mediaServer, param.getApp(), param.getStream())) {
             mediaServerService.closeStreams(mediaServer, param.getApp(), param.getStream());
         }
         String msgResult;
         if ("ffmpeg".equalsIgnoreCase(param.getType())){
+            if (param.getTimeoutMs() == 0) {
+                param.setTimeoutMs(15);
+            }
             result = mediaServerService.addFFmpegSource(mediaServer, param.getSrcUrl().trim(), param.getDstUrl(),
                     param.getTimeoutMs(), param.isEnableAudio(), param.isEnableMp4(),
                     param.getFfmpegCmdKey());
@@ -365,8 +360,6 @@
             result = mediaServerService.addStreamProxy(mediaServer, param.getApp(), param.getStream(), param.getUrl().trim(),
                     param.isEnableAudio(), param.isEnableMp4(), param.getRtpType());
         }
-        System.out.println("addStreamProxyToZlm====");
-        System.out.println(result);
         if (result != null && result.getCode() == 0) {
             String key = result.getData();
             if (key == null) {
@@ -385,6 +378,13 @@
             return null;
         }
         MediaServer mediaServer = mediaServerService.getOne(param.getMediaServerId());
+        if (mediaServer == null) {
+            return null;
+        }
+        List<StreamInfo> mediaList = mediaServerService.getMediaList(mediaServer, param.getApp(), param.getStream(), null);
+        if (mediaList == null || mediaList.isEmpty()) {
+            return true;
+        }
         Boolean result = false;
         if ("ffmpeg".equalsIgnoreCase(param.getType())){
             result = mediaServerService.delFFmpegSource(mediaServer, param.getStreamKey());
@@ -410,6 +410,7 @@
             gbStreamMapper.del(app, stream);
             videoManagerStorager.deleteStreamProxy(app, stream);
             redisCatchStorage.removeStream(streamProxyItem.getMediaServerId(), "PULL", app, stream);
+            redisCatchStorage.removeStream(streamProxyItem.getMediaServerId(), "PUSH", app, stream);
             Boolean result = removeStreamProxyFromZlm(streamProxyItem);
             if (result != null && result) {
                 logger.info("[绉婚櫎浠g悊]锛� 浠g悊锛� {}/{}, 浠巣lm绉婚櫎鎴愬姛", app, stream);
@@ -508,18 +509,18 @@
         String type = "PULL";
 
         // 鍙戦�乺edis娑堟伅
-        List<OnStreamChangedHookParam> onStreamChangedHookParams = redisCatchStorage.getStreams(mediaServerId, type);
-        if (onStreamChangedHookParams.size() > 0) {
-            for (OnStreamChangedHookParam onStreamChangedHookParam : onStreamChangedHookParams) {
+        List<MediaInfo> mediaInfoList = redisCatchStorage.getStreams(mediaServerId, type);
+        if (mediaInfoList.size() > 0) {
+            for (MediaInfo mediaInfo : mediaInfoList) {
                 JSONObject jsonObject = new JSONObject();
                 jsonObject.put("serverId", userSetting.getServerId());
-                jsonObject.put("app", onStreamChangedHookParam.getApp());
-                jsonObject.put("stream", onStreamChangedHookParam.getStream());
+                jsonObject.put("app", mediaInfo.getApp());
+                jsonObject.put("stream", mediaInfo.getStream());
                 jsonObject.put("register", false);
                 jsonObject.put("mediaServerId", mediaServerId);
                 redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
                 // 绉婚櫎redis鍐呮祦鐨勪俊鎭�
-                redisCatchStorage.removeStream(mediaServerId, type, onStreamChangedHookParam.getApp(), onStreamChangedHookParam.getStream());
+                redisCatchStorage.removeStream(mediaServerId, type, mediaInfo.getApp(), mediaInfo.getStream());
             }
         }
     }
@@ -531,14 +532,23 @@
 
     @Override
     public int updateStatus(boolean status, String app, String stream) {
-        return streamProxyMapper.updateStatus(app, stream, status);
+        // 鐘舵�佸彉鍖栨椂鎺ㄩ�佸埌鍥芥爣涓婄骇
+        StreamProxyItem streamProxyItem = streamProxyMapper.selectOne(app, stream);
+        if (streamProxyItem == null) {
+            return 0;
+        }
+        int result = streamProxyMapper.updateStatus(app, stream, status);
+        if (!ObjectUtils.isEmpty(streamProxyItem.getGbId())) {
+            gbStreamService.sendCatalogMsg(streamProxyItem, status?CatalogEvent.ON:CatalogEvent.OFF);
+        }
+        return result;
     }
 
     private void syncPullStream(String mediaServerId){
         MediaServer mediaServer = mediaServerService.getOne(mediaServerId);
         if (mediaServer != null) {
-            List<OnStreamChangedHookParam> allPullStream = redisCatchStorage.getStreams(mediaServerId, "PULL");
-            if (!allPullStream.isEmpty()) {
+            List<MediaInfo> mediaInfoList = redisCatchStorage.getStreams(mediaServerId, "PULL");
+            if (!mediaInfoList.isEmpty()) {
                 List<StreamInfo> mediaList = mediaServerService.getMediaList(mediaServer, null, null, null);
                 Map<String, StreamInfo> stringStreamInfoMap = new HashMap<>();
                 if (mediaList != null && !mediaList.isEmpty()) {

--
Gitblit v1.8.0