Merge remote-tracking branch 'origin/1078' into 1078

src/main/java/com/genersoft/iot/vmp/jt1078/bean/JTMediaStreamType.java

@@ -0,0 +1,5 @@
+package com.genersoft.iot.vmp.jt1078.bean;
+
+public enum JTMediaStreamType {
+    PLAY,PLAYBACK,TALK
+}

src/main/java/com/genersoft/iot/vmp/jt1078/service/Ijt1078Service.java

@@ -9,6 +9,8 @@ import com.github.pagehelper.PageInfo;
 import java.util.List;
 
 public interface Ijt1078Service {
+    JTMediaStreamType checkStreamFromJt(String stream);
+
     JTDevice getDevice(String phoneNumber);
 
     JTChannel getChannel(Integer terminalDbId, Integer channelId);

src/main/java/com/genersoft/iot/vmp/jt1078/service/impl/jt1078ServiceImpl.java

@@ -1,6 +1,5 @@
 package com.genersoft.iot.vmp.jt1078.service.impl;
 
-import com.alibaba.fastjson2.JSONObject;
 import com.genersoft.iot.vmp.common.GeneralCallback;
 import com.genersoft.iot.vmp.common.StreamInfo;
 import com.genersoft.iot.vmp.common.VideoManagerConstants;
@@ -14,7 +13,6 @@ import com.genersoft.iot.vmp.jt1078.bean.common.ConfigAttribute;
 import com.genersoft.iot.vmp.jt1078.cmd.JT1078Template;
 import com.genersoft.iot.vmp.jt1078.dao.JTChannelMapper;
 import com.genersoft.iot.vmp.jt1078.dao.JTTerminalMapper;
-import com.genersoft.iot.vmp.jt1078.event.CallbackManager;
 import com.genersoft.iot.vmp.jt1078.event.FtpUploadEvent;
 import com.genersoft.iot.vmp.jt1078.proc.request.J1205;
 import com.genersoft.iot.vmp.jt1078.proc.response.*;
@@ -25,14 +23,16 @@ import com.genersoft.iot.vmp.media.event.hook.Hook;
 import com.genersoft.iot.vmp.media.event.hook.HookData;
 import com.genersoft.iot.vmp.media.event.hook.HookSubscribe;
 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.event.mediaServer.MediaSendRtpStoppedEvent;
 import com.genersoft.iot.vmp.media.service.IMediaServerService;
-import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
-import com.genersoft.iot.vmp.service.IMediaService;
 import com.genersoft.iot.vmp.service.bean.InviteErrorCode;
 import com.genersoft.iot.vmp.service.bean.SSRCInfo;
 import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
 import com.genersoft.iot.vmp.utils.DateUtil;
+import com.genersoft.iot.vmp.utils.MediaServerUtils;
 import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
 import com.github.pagehelper.PageHelper;
 import com.github.pagehelper.PageInfo;
@@ -42,6 +42,7 @@ import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.event.EventListener;
 import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 
 import java.lang.reflect.Field;
@@ -71,27 +72,95 @@ public class jt1078ServiceImpl implements Ijt1078Service {
     @Autowired
     private IMediaServerService mediaServerService;
 
-    @Autowired
-    private IMediaService mediaService;
-
     @Autowired
     private DynamicTask dynamicTask;
 
     @Autowired
     private UserSetting userSetting;
 
-    @Autowired
-    private ZLMRESTfulUtils zlmresTfulUtils;
-
-    @Autowired
-    private CallbackManager callbackManager;
-
     @Autowired
     private IRedisCatchStorage redisCatchStorage;
 
     @Autowired
     private FtpSetting ftpSetting;
 
+    /**
+     * 流到来的处理
+     */
+    @Async("taskExecutor")
+    @org.springframework.context.event.EventListener
+    public void onApplicationEvent(MediaArrivalEvent event) {
+
+    }
+
+    /**
+     * 流离开的处理
+     */
+    @Async("taskExecutor")
+    @EventListener
+    public void onApplicationEvent(MediaDepartureEvent event) {
+
+    }
+
+    /**
+     * 流未找到的处理
+     */
+    @Async("taskExecutor")
+    @EventListener
+    public void onApplicationEvent(MediaNotFoundEvent event) {
+        if (!userSetting.isAutoApplyPlay()) {
+            return;
+        }
+        JTMediaStreamType jtMediaStreamType = checkStreamFromJt(event.getStream());
+        if (jtMediaStreamType == null){
+            return;
+        }
+        String[] streamParamArray = event.getStream().split("_");
+        String phoneNumber = streamParamArray[1];
+        int channelId = Integer.parseInt(streamParamArray[2]);
+        String params = event.getParams();
+        Map<String, String> paramMap = MediaServerUtils.urlParamToMap(params);
+        int type = 0;
+        try {
+            type = Integer.parseInt(paramMap.get("type"));
+        }catch (NumberFormatException ignored) {}
+        if (jtMediaStreamType.equals(JTMediaStreamType.PLAY)) {
+            play(phoneNumber, channelId, 0, null);
+        }else if (jtMediaStreamType.equals(JTMediaStreamType.PLAYBACK)) {
+            String startTimeParam = DateUtil.jt1078Toyyyy_MM_dd_HH_mm_ss(streamParamArray[3]);
+            String endTimeParam = DateUtil.jt1078Toyyyy_MM_dd_HH_mm_ss(streamParamArray[4]);
+            int rate = 0;
+            int playbackType = 0;
+            int playbackSpeed = 0;
+            try {
+                rate = Integer.parseInt(paramMap.get("rate"));
+                playbackType = Integer.parseInt(paramMap.get("playbackType"));
+                playbackSpeed = Integer.parseInt(paramMap.get("playbackSpeed"));
+            }catch (NumberFormatException ignored) {}
+            playback(phoneNumber, channelId, startTimeParam, endTimeParam, type, rate, playbackType, playbackSpeed, null);
+        }
+    }
+
+
+    /**
+     * 校验流是否是属于部标的
+     */
+    @Override
+    public JTMediaStreamType checkStreamFromJt(String stream) {
+        if (!stream.startsWith("jt_")) {
+            return null;
+        }
+        String[] streamParamArray = stream.split("_");
+        if (streamParamArray.length == 3) {
+            return JTMediaStreamType.PLAY;
+        }else if (streamParamArray.length == 5) {
+            return JTMediaStreamType.PLAYBACK;
+        }else if (streamParamArray.length == 4) {
+            return JTMediaStreamType.TALK;
+        }else {
+            return null;
+        }
+    }
 
     @Override
     public JTDevice getDevice(String phoneNumber) {
@@ -172,7 +241,7 @@ public class jt1078ServiceImpl implements Ijt1078Service {
             // 清理数据
             redisTemplate.delete(playKey);
         }
-        String stream = "jt_play_" + phoneNumber + "_" + channelId;
+        String stream = "jt_" + phoneNumber + "_" + channelId;
         MediaServer mediaServer = mediaServerService.getMediaServerForMinimumLoad(null);
         if (mediaServer == null) {
             for (GeneralCallback<StreamInfo> errorCallback : errorCallbacks) {
@@ -189,6 +258,9 @@ public class jt1078ServiceImpl implements Ijt1078Service {
             StreamInfo info = onPublishHandler(mediaServer, hookData, phoneNumber, channelId);
 
             for (GeneralCallback<StreamInfo> errorCallback : errorCallbacks) {
+                if (errorCallback == null) {
+                    continue;
+                }
                 errorCallback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), info);
             }
             subscribe.removeSubscribe(hook);
@@ -208,6 +280,10 @@ public class jt1078ServiceImpl implements Ijt1078Service {
         });
         // 开启收流端口
         SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServer, stream, "000", false, false, 0, false, !channel.getHasAudio(), false, 1);
+        if (ssrcInfo == null) {
+            stopPlay(phoneNumber, channelId);
+            return;
+        }
         // 设置超时监听
         dynamicTask.startDelay(playKey, () -> {
             logger.info("[1078-点播] 超时， phoneNumber： {}， channelId： {}", phoneNumber, channelId);
@@ -354,7 +430,7 @@ public class jt1078ServiceImpl implements Ijt1078Service {
         }
         String startTimeParam = DateUtil.yyyy_MM_dd_HH_mm_ssTo1078(startTime);
         String endTimeParam = DateUtil.yyyy_MM_dd_HH_mm_ssTo1078(endTime);
-        String stream = phoneNumber + "_" + channelId + "_" + startTimeParam + "_" + endTimeParam;
+        String stream = "jt_" + phoneNumber + "_" + channelId + "_" + startTimeParam + "_" + endTimeParam;
         MediaServer mediaServer = mediaServerService.getMediaServerForMinimumLoad(null);
         if (mediaServer == null) {
             for (GeneralCallback<StreamInfo> errorCallback : errorCallbacks) {
@@ -904,7 +980,7 @@ public class jt1078ServiceImpl implements Ijt1078Service {
             throw new ControllerException(ErrorCode.ERROR100.getCode(), "对讲进行中");
         }
 
-        String receiveStream = "1078" + "_" + phoneNumber + "_" + channelId;
+        String receiveStream = "jt_" + phoneNumber + "_" + channelId + "_talk";
         MediaServer mediaServer = mediaServerService.getOne(mediaServerId);
         if (mediaServer == null) {
             for (GeneralCallback<StreamInfo> errorCallback : errorCallbacks) {

src/main/java/com/genersoft/iot/vmp/media/event/media/MediaEvent.java

@@ -20,6 +20,16 @@ public class MediaEvent extends ApplicationEvent {
 
     private String schema;
 
+    private String params;
+
+    public String getParams() {
+        return params;
+    }
+
+    public void setParams(String params) {
+        this.params = params;
+    }
+
 
     public String getApp() {
         return app;

src/main/java/com/genersoft/iot/vmp/media/event/media/MediaNotFoundEvent.java

@@ -17,6 +17,7 @@ public class MediaNotFoundEvent extends MediaEvent {
         mediaDepartureEven.setStream(hookParam.getStream());
         mediaDepartureEven.setSchema(hookParam.getSchema());
         mediaDepartureEven.setMediaServer(mediaServer);
+        mediaDepartureEven.setParams(hookParam.getParams());
         return mediaDepartureEven;
     }
 }

src/main/java/com/genersoft/iot/vmp/media/event/media/MediaPublishEvent.java

@@ -21,13 +21,4 @@ public class MediaPublishEvent extends MediaEvent {
         return mediaPublishEvent;
     }
 
-    private String params;
-
-    public String getParams() {
-        return params;
-    }
-
-    public void setParams(String params) {
-        this.params = params;
-    }
 }

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

@@ -10,6 +10,8 @@ import com.genersoft.iot.vmp.gb28181.bean.*;
 import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
 import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
 import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
+import com.genersoft.iot.vmp.jt1078.bean.JTMediaStreamType;
+import com.genersoft.iot.vmp.jt1078.service.Ijt1078Service;
 import com.genersoft.iot.vmp.media.bean.MediaServer;
 import com.genersoft.iot.vmp.media.bean.ResultForOnPublish;
 import com.genersoft.iot.vmp.media.zlm.dto.StreamAuthorityInfo;
@@ -72,6 +74,9 @@ public class MediaServiceImpl implements IMediaService {
     @Autowired
     private ISIPCommander commander;
 
+    @Autowired
+    private Ijt1078Service ijt1078Service;
+
     @Override
     public boolean authenticatePlay(String app, String stream, String callId) {
         if (app == null || stream == null) {
@@ -267,10 +272,19 @@ public class MediaServiceImpl implements IMediaService {
                         inviteInfo.getChannelId(), inviteInfo.getStream());
                 storager.stopPlay(inviteInfo.getDeviceId(), inviteInfo.getChannelId());
                 return result;
+            }else {
+                // 判断是否是1078点播
+                JTMediaStreamType jtMediaStreamType = ijt1078Service.checkStreamFromJt(stream);
+                if (jtMediaStreamType != null) {
+                    String[] streamParamArray = stream.split("_");
+                    if (jtMediaStreamType.equals(JTMediaStreamType.PLAY)) {
+                        ijt1078Service.stopPlay(streamParamArray[1], Integer.parseInt(streamParamArray[2]));
+                    }else if (jtMediaStreamType.equals(JTMediaStreamType.PLAYBACK)) {
+                        ijt1078Service.stopPlayback(streamParamArray[1], Integer.parseInt(streamParamArray[2]));
+                    }
+                }
             }
-            if (stream.startsWith("1078")) {
+
-                return false;
-            }
             SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(null, null, stream, null);
             if (sendRtpItem != null && "talk".equals(sendRtpItem.getApp() )) {
                 return false;