支持处理流没找到事件

src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java

@@ -253,123 +253,17 @@ public class ZLMHttpHookListener {
      */
     @ResponseBody
     @PostMapping(value = "/on_stream_not_found", produces = "application/json;charset=UTF-8")
-    public DeferredResult<HookResult> onStreamNotFound(@RequestBody OnStreamNotFoundHookParam param) {
+    public HookResult onStreamNotFound(@RequestBody OnStreamNotFoundHookParam param) {
         logger.info("[ZLM HOOK] 流未找到：{}->{}->{}/{}", param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream());
 
-        DeferredResult<HookResult> defaultResult = new DeferredResult<>();
 
         MediaServer mediaServer = mediaServerService.getOne(param.getMediaServerId());
         if (!userSetting.isAutoApplyPlay() || mediaServer == null) {
-            defaultResult.setResult(new HookResult(ErrorCode.ERROR404.getCode(), ErrorCode.ERROR404.getMsg()));
-            return defaultResult;
+            return HookResult.SUCCESS();
         }
         MediaNotFoundEvent mediaNotFoundEvent = MediaNotFoundEvent.getInstance(this, param, mediaServer);
         applicationEventPublisher.publishEvent(mediaNotFoundEvent);
-
-
-
-        if ("rtp".equals(param.getApp())) {
-            String[] s = param.getStream().split("_");
-            if ((s.length != 2 && s.length != 4)) {
-                defaultResult.setResult(HookResult.SUCCESS());
-                return defaultResult;
-            }
-            String deviceId = s[0];
-            String channelId = s[1];
-            Device device = redisCatchStorage.getDevice(deviceId);
-            if (device == null || !device.isOnLine()) {
-                defaultResult.setResult(new HookResult(ErrorCode.ERROR404.getCode(), ErrorCode.ERROR404.getMsg()));
-                return defaultResult;
-            }
-            DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
-            if (deviceChannel == null) {
-                defaultResult.setResult(new HookResult(ErrorCode.ERROR404.getCode(), ErrorCode.ERROR404.getMsg()));
-                return defaultResult;
-            }
-            if (s.length == 2) {
-                logger.info("[ZLM HOOK] 预览流未找到, 发起自动点播：{}->{}->{}/{}", param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream());
-
-                RequestMessage msg = new RequestMessage();
-                String key = DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId;
-                boolean exist = resultHolder.exist(key, null);
-                msg.setKey(key);
-                String uuid = UUID.randomUUID().toString();
-                msg.setId(uuid);
-                DeferredResult<HookResult> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue());
-
-                result.onTimeout(() -> {
-                    logger.info("[ZLM HOOK] 预览流自动点播, 等待超时");
-                    msg.setData(new HookResult(ErrorCode.ERROR100.getCode(), "点播超时"));
-                    resultHolder.invokeAllResult(msg);
-                    inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId);
-                    storager.stopPlay(deviceId, channelId);
-                });
-
-                resultHolder.put(key, uuid, result);
-
-                if (!exist) {
-                    playService.play(mediaServer, deviceId, channelId, null, (code, message, data) -> {
-                        msg.setData(new HookResult(code, message));
-                        resultHolder.invokeResult(msg);
-                    });
-                }
-                return result;
-            } else if (s.length == 4) {
-                // 此时为录像回放， 录像回放格式为> 设备ID_通道ID_开始时间_结束时间
-                String startTimeStr = s[2];
-                String endTimeStr = s[3];
-                if (startTimeStr == null || endTimeStr == null || startTimeStr.length() != 14 || endTimeStr.length() != 14) {
-                    defaultResult.setResult(HookResult.SUCCESS());
-                    return defaultResult;
-                }
-                String startTime = DateUtil.urlToyyyy_MM_dd_HH_mm_ss(startTimeStr);
-                String endTime = DateUtil.urlToyyyy_MM_dd_HH_mm_ss(endTimeStr);
-                logger.info("[ZLM HOOK] 回放流未找到, 发起自动点播：{}->{}->{}/{}-{}-{}",
-                        param.getMediaServerId(), param.getSchema(),
-                        param.getApp(), param.getStream(),
-                        startTime, endTime
-                );
-                RequestMessage msg = new RequestMessage();
-                String key = DeferredResultHolder.CALLBACK_CMD_PLAYBACK + deviceId + channelId;
-                boolean exist = resultHolder.exist(key, null);
-                msg.setKey(key);
-                String uuid = UUID.randomUUID().toString();
-                msg.setId(uuid);
-                DeferredResult<HookResult> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue());
-
-                result.onTimeout(() -> {
-                    logger.info("[ZLM HOOK] 回放流自动点播, 等待超时");
-                    // 释放rtpserver
-                    msg.setData(new HookResult(ErrorCode.ERROR100.getCode(), "点播超时"));
-                    resultHolder.invokeResult(msg);
-                });
-
-                resultHolder.put(key, uuid, result);
-
-                if (!exist) {
-                    SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServer, param.getStream(), null,
-                            device.isSsrcCheck(), true, 0, false, false, device.getStreamModeForParam());
-                    playService.playBack(mediaServer, ssrcInfo, deviceId, channelId, startTime, endTime, (code, message, data) -> {
-                        msg.setData(new HookResult(code, message));
-                        resultHolder.invokeResult(msg);
-                    });
-                }
-                return result;
-            } else {
-                defaultResult.setResult(HookResult.SUCCESS());
-                return defaultResult;
-            }
-
-        } else {
-            // 拉流代理
-            StreamProxyItem streamProxyByAppAndStream = streamProxyService.getStreamProxyByAppAndStream(param.getApp(), param.getStream());
-            if (streamProxyByAppAndStream != null && streamProxyByAppAndStream.isEnableDisableNoneReader()) {
-                streamProxyService.start(param.getApp(), param.getStream());
-            }
-            DeferredResult<HookResult> result = new DeferredResult<>();
-            result.setResult(HookResult.SUCCESS());
-            return result;
-        }
+        return HookResult.SUCCESS();
     }
 
     /**

src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java

@@ -13,12 +13,15 @@ import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
 import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager;
 import com.genersoft.iot.vmp.gb28181.session.SSRCFactory;
 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.ISIPCommander;
 import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
 import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
 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.SendRtpPortManager;
 import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory;
@@ -28,6 +31,7 @@ import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForRecordMp4;
 import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange;
 import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
 import com.genersoft.iot.vmp.media.zlm.dto.hook.HookParam;
+import com.genersoft.iot.vmp.media.zlm.dto.hook.HookResult;
 import com.genersoft.iot.vmp.media.zlm.dto.hook.OnRecordMp4HookParam;
 import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
 import com.genersoft.iot.vmp.service.*;
@@ -49,6 +53,7 @@ import org.springframework.context.event.EventListener;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 import org.springframework.util.ObjectUtils;
+import org.springframework.web.context.request.async.DeferredResult;
 
 import javax.sdp.*;
 import javax.sip.InvalidArgumentException;
@@ -195,6 +200,53 @@ public class PlayServiceImpl implements IPlayService {
         }
     }
 
+    /**
+     * 流未找到的处理
+     */
+    @Async("taskExecutor")
+    @EventListener
+    public void onApplicationEvent(MediaNotFoundEvent event) {
+        if (!"rtp".equals(event.getApp())) {
+            return;
+        }
+        String[] s = event.getStream().split("_");
+        if ((s.length != 2 && s.length != 4)) {
+            return;
+        }
+        String deviceId = s[0];
+        String channelId = s[1];
+        Device device = redisCatchStorage.getDevice(deviceId);
+        if (device == null || !device.isOnLine()) {
+            return;
+        }
+        DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
+        if (deviceChannel == null) {
+            return;
+        }
+        if (s.length == 2) {
+            logger.info("[ZLM HOOK] 预览流未找到, 发起自动点播：{}->{}->{}/{}", event.getMediaServer().getId(), event.getSchema(), event.getApp(), event.getStream());
+            play(event.getMediaServer(), deviceId, channelId, null, null);
+        } else if (s.length == 4) {
+            // 此时为录像回放， 录像回放格式为> 设备ID_通道ID_开始时间_结束时间
+            String startTimeStr = s[2];
+            String endTimeStr = s[3];
+            if (startTimeStr == null || endTimeStr == null || startTimeStr.length() != 14 || endTimeStr.length() != 14) {
+                return;
+            }
+            String startTime = DateUtil.urlToyyyy_MM_dd_HH_mm_ss(startTimeStr);
+            String endTime = DateUtil.urlToyyyy_MM_dd_HH_mm_ss(endTimeStr);
+            logger.info("[ZLM HOOK] 回放流未找到, 发起自动点播：{}->{}->{}/{}-{}-{}",
+                    event.getMediaServer().getId(), event.getSchema(),
+                    event.getApp(), event.getStream(),
+                    startTime, endTime
+            );
+
+            SSRCInfo ssrcInfo = mediaServerService.openRTPServer(event.getMediaServer(), event.getStream(), null,
+                    device.isSsrcCheck(), true, 0, false, false, device.getStreamModeForParam());
+            playBack(event.getMediaServer(), ssrcInfo, deviceId, channelId, startTime, endTime, null);
+        }
+    }
+
 
     @Override
     public SSRCInfo play(MediaServer mediaServerItem, String deviceId, String channelId, String ssrc, ErrorCallback<Object> callback) {

src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java

@@ -11,6 +11,7 @@ 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.zlm.ZlmHttpHookSubscribe;
@@ -18,6 +19,7 @@ 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.MediaServer;
 import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
+import com.genersoft.iot.vmp.media.zlm.dto.hook.HookResult;
 import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
 import com.genersoft.iot.vmp.service.IGbStreamService;
 import com.genersoft.iot.vmp.service.IMediaService;
@@ -44,6 +46,7 @@ import org.springframework.transaction.TransactionDefinition;
 import org.springframework.transaction.TransactionStatus;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.ObjectUtils;
+import org.springframework.web.context.request.async.DeferredResult;
 
 import java.util.HashMap;
 import java.util.List;
@@ -128,6 +131,22 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
         }
     }
 
+    /**
+     * 流离开的处理
+     */
+    @Async("taskExecutor")
+    @EventListener
+    public void onApplicationEvent(MediaNotFoundEvent event) {
+        if ("rtp".equals(event.getApp())) {
+            return;
+        }
+        // 拉流代理
+        StreamProxyItem streamProxyByAppAndStream = getStreamProxyByAppAndStream(event.getApp(), event.getStream());
+        if (streamProxyByAppAndStream != null && streamProxyByAppAndStream.isEnableDisableNoneReader()) {
+            start(event.getApp(), event.getStream());
+        }
+    }
+
 
     @Override
     public void save(StreamProxyItem param, GeneralCallback<StreamInfo> callback) {