兼容流注销时流产生类型错误的问题

648540858
1 parent 963a74d2
Showing 11 changed files with 106 additions and 49 deletions
src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java
src/main/java/com/genersoft/iot/vmp/conf/ThreadPoolTaskConfig.java
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookSubscribe.java
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRunner.java
src/main/java/com/genersoft/iot/vmp/media/zlm/dto/OriginType.java
src/main/java/com/genersoft/iot/vmp/service/IStreamPushService.java
src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java
src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java
src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java
src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java
@@ -56,5 +56,5 @@ public class VideoManagerConstants {
 	public static final String MEDIA_TRANSACTION_USED_PREFIX = "VMP_media_transaction_";
 	//************************** redis 消息*********************************
-	public static final String WVP_MSG_STREAM_PUSH_CHANGE_PREFIX = "WVP_MSG_STREAM_PUSH_CHANGE";
+	public static final String WVP_MSG_STREAM_PUSH_CHANGE_PREFIX = "WVP_MSG_STREAM_CHANGE_";
 }
@@ -38,7 +38,7 @@ public class ThreadPoolTaskConfig {
     /**
      * 线程池名前缀
      */
-    private static final String threadNamePrefix = "wvp-sip-";
+    private static final String threadNamePrefix = "wvp-";
     @Bean("taskExecutor") // bean的名称，默认为首字母小写的方法名
     public ThreadPoolTaskExecutor taskExecutor() {
@@ -9,17 +9,11 @@ import com.genersoft.iot.vmp.common.StreamInfo;
 import com.genersoft.iot.vmp.conf.MediaConfig;
 import com.genersoft.iot.vmp.conf.UserSetup;
 import com.genersoft.iot.vmp.gb28181.bean.Device;
-import com.genersoft.iot.vmp.media.zlm.dto.MediaItem;
-import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
-import com.genersoft.iot.vmp.media.zlm.dto.OriginType;
-import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
-import com.genersoft.iot.vmp.service.IMediaServerService;
-import com.genersoft.iot.vmp.service.IMediaService;
-import com.genersoft.iot.vmp.service.IStreamProxyService;
+import com.genersoft.iot.vmp.media.zlm.dto.*;
+import com.genersoft.iot.vmp.service.*;
 import com.genersoft.iot.vmp.service.bean.SSRCInfo;
 import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
 import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
-import com.genersoft.iot.vmp.service.IPlayService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -66,6 +60,9 @@ public class ZLMHttpHookListener {
 	private IStreamProxyService streamProxyService;
 	@Autowired
+	private IStreamPushService streamPushService;
+
+	@Autowired
 	private IMediaService mediaService;
 	@Autowired
@@ -84,6 +81,32 @@ public class ZLMHttpHookListener {
 	private MediaConfig mediaConfig;
 	/**
+	 * 服务器定时上报时间，上报间隔可配置，默认10s上报一次
+	 *
+	 */
+	@ResponseBody
+	@PostMapping(value = "/on_server_keepalive", produces = "application/json;charset=UTF-8")
+	public ResponseEntity<String> onServerKeepalive(@RequestBody JSONObject json){
+
+		if (logger.isDebugEnabled()) {
+			logger.debug("[ ZLM HOOK ]on_server_keepalive API调用，参数：" + json.toString());
+		}
+		String mediaServerId = json.getString("mediaServerId");
+
+		List<ZLMHttpHookSubscribe.Event> subscribes = this.subscribe.getSubscribes(ZLMHttpHookSubscribe.HookType.on_server_keepalive);
+		if (subscribes != null  && subscribes.size() > 0) {
+			for (ZLMHttpHookSubscribe.Event subscribe : subscribes) {
+				subscribe.response(null, json);
+			}
+		}
+
+		JSONObject ret = new JSONObject();
+		ret.put("code", 0);
+		ret.put("msg", "success");
+		return new ResponseEntity<String>(ret.toString(),HttpStatus.OK);
+	}
+
+	/**
 	 * 流量统计事件，播放器或推流器断开时并且耗用流量超过特定阈值时会触发此事件，阈值通过配置文件general.flowThreshold配置；此事件对回复不敏感。
 	 *  
 	 */
@@ -92,7 +115,7 @@ public class ZLMHttpHookListener {
 	public ResponseEntity<String> onFlowReport(@RequestBody JSONObject json){
 		if (logger.isDebugEnabled()) {
-			logger.debug("ZLM HOOK on_flow_report API调用，参数：" + json.toString());
+			logger.debug("[ ZLM HOOK ]on_flow_report API调用，参数：" + json.toString());
 		}
 		String mediaServerId = json.getString("mediaServerId");
 		JSONObject ret = new JSONObject();
@@ -110,7 +133,7 @@ public class ZLMHttpHookListener {
 	public ResponseEntity<String> onHttpAccess(@RequestBody JSONObject json){
 		if (logger.isDebugEnabled()) {
-			logger.debug("ZLM HOOK on_http_access API 调用，参数：" + json.toString());
+			logger.debug("[ ZLM HOOK ]on_http_access API 调用，参数：" + json.toString());
 		}
 		String mediaServerId = json.getString("mediaServerId");
 		JSONObject ret = new JSONObject();
@@ -130,7 +153,7 @@ public class ZLMHttpHookListener {
 	public ResponseEntity<String> onPlay(@RequestBody JSONObject json){
 		if (logger.isDebugEnabled()) {
-			logger.debug("ZLM HOOK on_play API调用，参数：" + json.toString());
+			logger.debug("[ ZLM HOOK ]on_play API调用，参数：" + json.toString());
 		}
 		String mediaServerId = json.getString("mediaServerId");
 		ZLMHttpHookSubscribe.Event subscribe = this.subscribe.getSubscribe(ZLMHttpHookSubscribe.HookType.on_play, json);
@@ -155,7 +178,7 @@ public class ZLMHttpHookListener {
 	@PostMapping(value = "/on_publish", produces = "application/json;charset=UTF-8")
 	public ResponseEntity<String> onPublish(@RequestBody JSONObject json) {
-		logger.debug("ZLM HOOK on_publish API调用，参数：" + json.toString());
+		logger.debug("[ ZLM HOOK ]on_publish API调用，参数：" + json.toString());
 		String mediaServerId = json.getString("mediaServerId");
 		ZLMHttpHookSubscribe.Event subscribe = this.subscribe.getSubscribe(ZLMHttpHookSubscribe.HookType.on_publish, json);
@@ -191,7 +214,7 @@ public class ZLMHttpHookListener {
 	public ResponseEntity<String> onRecordMp4(@RequestBody JSONObject json){
 		if (logger.isDebugEnabled()) {
-			logger.debug("ZLM HOOK on_record_mp4 API调用，参数：" + json.toString());
+			logger.debug("[ ZLM HOOK ]on_record_mp4 API调用，参数：" + json.toString());
 		}
 		String mediaServerId = json.getString("mediaServerId");
 		JSONObject ret = new JSONObject();
@@ -209,7 +232,7 @@ public class ZLMHttpHookListener {
 	public ResponseEntity<String> onRtspRealm(@RequestBody JSONObject json){
 		if (logger.isDebugEnabled()) {
-			logger.debug("ZLM HOOK on_rtsp_realm API调用，参数：" + json.toString());
+			logger.debug("[ ZLM HOOK ]on_rtsp_realm API调用，参数：" + json.toString());
 		}
 		String mediaServerId = json.getString("mediaServerId");
 		JSONObject ret = new JSONObject();
@@ -228,7 +251,7 @@ public class ZLMHttpHookListener {
 	public ResponseEntity<String> onRtspAuth(@RequestBody JSONObject json){
 		if (logger.isDebugEnabled()) {
-			logger.debug("ZLM HOOK on_rtsp_auth API调用，参数：" + json.toString());
+			logger.debug("[ ZLM HOOK ]on_rtsp_auth API调用，参数：" + json.toString());
 		}
 		String mediaServerId = json.getString("mediaServerId");
 		JSONObject ret = new JSONObject();
@@ -247,7 +270,7 @@ public class ZLMHttpHookListener {
 	public ResponseEntity<String> onShellLogin(@RequestBody JSONObject json){
 		if (logger.isDebugEnabled()) {
-			logger.debug("ZLM HOOK on_shell_login API调用，参数：" + json.toString());
+			logger.debug("[ ZLM HOOK ]on_shell_login API调用，参数：" + json.toString());
 		}
 		// TODO 如果是带有rtpstream则开启按需拉流
 		// String app = json.getString("app");
@@ -277,7 +300,7 @@ public class ZLMHttpHookListener {
 	public ResponseEntity<String> onStreamChanged(@RequestBody MediaItem item){
 		if (logger.isDebugEnabled()) {
-			logger.debug("ZLM HOOK on_stream_changed API调用，参数：" + JSONObject.toJSONString(item));
+			logger.debug("[ ZLM HOOK ]on_stream_changed API调用，参数：" + JSONObject.toJSONString(item));
 		}
 		String mediaServerId = item.getMediaServerId();
 		JSONObject json = (JSONObject) JSON.toJSON(item);
@@ -315,33 +338,39 @@ public class ZLMHttpHookListener {
 				}
 			}else {
 				if (!"rtp".equals(app)){
-
+					String type = OriginType.values()[item.getOriginType()].getType();
 					MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
 					if (regist) {
 						StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(mediaServerItem, app, streamId, tracks);
-						redisCatchStorage.addStream(mediaServerItem, OriginType.values()[item.getOriginType()].getType(), app, streamId, streamInfo);
+						redisCatchStorage.addStream(mediaServerItem, type, app, streamId, streamInfo);
 						if (item.getOriginType() == OriginType.RTSP_PUSH.ordinal()
 								|| item.getOriginType() == OriginType.RTMP_PUSH.ordinal()
 								|| item.getOriginType() == OriginType.RTC_PUSH.ordinal() ) {
 							zlmMediaListManager.addMedia(item);
 						}
 					}else {
+						// 兼容流注销时类型错误的问题，等zlm更新后删除
+						StreamPushItem streamPushItem = streamPushService.getPush(app, streamId);
+						if (streamPushItem != null) {
+							type = "PUSH";
+						}else {
+							StreamProxyItem streamProxyByAppAndStream = streamProxyService.getStreamProxyByAppAndStream(app, streamId);
+							if (streamProxyByAppAndStream != null) {
+								type = "PULL";
+							}
+						}
 						zlmMediaListManager.removeMedia( app, streamId);
 						redisCatchStorage.removeStream(mediaServerItem, OriginType.values()[item.getOriginType()].getType(), app, streamId);
-
-					}
-					if (item.getOriginType() == OriginType.RTSP_PUSH.ordinal()
-							|| item.getOriginType() == OriginType.RTMP_PUSH.ordinal()
-							|| item.getOriginType() == OriginType.RTC_PUSH.ordinal() ) {
-						// 发送流变化redis消息
-						JSONObject jsonObject = new JSONObject();
-						jsonObject.put("serverId", userSetup.getServerId());
-						jsonObject.put("app", app);
-						jsonObject.put("stream", streamId);
-						jsonObject.put("register", regist);
-						jsonObject.put("mediaServerId", mediaServerId);
-						redisCatchStorage.sendStreamChangeMsg(jsonObject);
 					}
+
+					// 发送流变化redis消息
+					JSONObject jsonObject = new JSONObject();
+					jsonObject.put("serverId", userSetup.getServerId());
+					jsonObject.put("app", app);
+					jsonObject.put("stream", streamId);
+					jsonObject.put("register", regist);
+					jsonObject.put("mediaServerId", mediaServerId);
+					redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
 				}
 			}
 		}
@@ -361,7 +390,7 @@ public class ZLMHttpHookListener {
 	public ResponseEntity<String> onStreamNoneReader(@RequestBody JSONObject json){
 		if (logger.isDebugEnabled()) {
-			logger.debug("ZLM HOOK on_stream_none_reader API调用，参数：" + json.toString());
+			logger.debug("[ ZLM HOOK ]on_stream_none_reader API调用，参数：" + json.toString());
 		}
 		String mediaServerId = json.getString("mediaServerId");
 		String streamId = json.getString("stream");
@@ -421,7 +450,7 @@ public class ZLMHttpHookListener {
 	@PostMapping(value = "/on_stream_not_found", produces = "application/json;charset=UTF-8")
 	public ResponseEntity<String> onStreamNotFound(@RequestBody JSONObject json){
 		if (logger.isDebugEnabled()) {
-			logger.debug("ZLM HOOK on_stream_not_found API调用，参数：" + json.toString());
+			logger.debug("[ ZLM HOOK ]on_stream_not_found API调用，参数：" + json.toString());
 		}
 		String mediaServerId = json.getString("mediaServerId");
 		MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId);
@@ -468,7 +497,7 @@ public class ZLMHttpHookListener {
 	public ResponseEntity<String> onServerStarted(HttpServletRequest request, @RequestBody JSONObject jsonObject){
 		if (logger.isDebugEnabled()) {
-			logger.debug("ZLM HOOK on_server_started API调用，参数：" + jsonObject.toString());
+			logger.debug("[ ZLM HOOK ]on_server_started API调用，参数：" + jsonObject.toString());
 		}
 		String remoteAddr = request.getRemoteAddr();
 		jsonObject.put("ip", remoteAddr);
@@ -27,7 +27,8 @@ public class ZLMHttpHookSubscribe {
         on_stream_changed,
         on_stream_none_reader,
         on_stream_not_found,
-        on_server_started
+        on_server_started,
+        on_server_keepalive
     }
     public interface Event{
@@ -3,7 +3,6 @@ package com.genersoft.iot.vmp.media.zlm;
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
-import com.genersoft.iot.vmp.common.VideoManagerConstants;
 import com.genersoft.iot.vmp.conf.MediaConfig;
 import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
 import com.genersoft.iot.vmp.service.IMediaServerService;
@@ -11,9 +10,11 @@ import com.genersoft.iot.vmp.service.IStreamProxyService;
 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.boot.CommandLineRunner;
 import org.springframework.core.annotation.Order;
 import org.springframework.scheduling.annotation.Async;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.stereotype.Component;
 import java.util.*;
@@ -41,8 +42,13 @@ public class ZLMRunner implements CommandLineRunner {
     @Autowired
     private MediaConfig mediaConfig;
+    @Qualifier("taskExecutor")
+    @Autowired
+    private ThreadPoolTaskExecutor taskExecutor;
+
     @Override
     public void run(String... strings) throws Exception {
+        mediaServerService.clearMediaServerForOnline();
         if (mediaServerService.getDefaultMediaServer() == null) {
             mediaServerService.addToDatabase(mediaConfig.getMediaSerItem());
         }
@@ -59,6 +65,9 @@ public class ZLMRunner implements CommandLineRunner {
             }
         });
+        // TODO 订阅 zlm保活事件, 当zlm离线时做业务的处理
+
+
         // 获取zlm信息
         logger.info("等待默认zlm接入...");
@@ -70,7 +79,9 @@ public class ZLMRunner implements CommandLineRunner {
         for (MediaServerItem mediaServerItem : all) {
             if (startGetMedia == null) startGetMedia = new HashMap<>();
             startGetMedia.put(mediaServerItem.getId(), true);
-            connectZlmServer(mediaServerItem);
+            taskExecutor.execute(()->{
+                connectZlmServer(mediaServerItem);
+            });
         }
         Timer timer = new Timer();
         // 2分钟后未连接到则不再去主动连接, TODO 并对重启前使用此在zlm的通道发送bye
 package com.genersoft.iot.vmp.media.zlm.dto;
 public enum OriginType {
+    // 不可调整顺序
     UNKNOWN("UNKNOWN"),
     RTMP_PUSH("PUSH"),
     RTSP_PUSH("PUSH"),
     RTP_PUSH("RTP"),
-    RTC_PUSH("PUSH"),
     PULL("PULL"),
     FFMPEG_PULL("PULL"),
     MP4_VOD("MP4_VOD"),
-    DEVICE_CHN("DEVICE_CHN");
+    DEVICE_CHN("DEVICE_CHN"),
+    RTC_PUSH("PUSH");
     private final String type;
     OriginType(String type) {
@@ -35,4 +35,6 @@ public interface IStreamPushService {
     PageInfo<StreamPushItem> getPushList(Integer page, Integer count);
     StreamPushItem transform(MediaItem item);
+
+    StreamPushItem getPush(String app, String streamId);
 }
@@ -310,7 +310,7 @@ public class MediaServerServiceImpl implements IMediaServerService, CommandLineR
      */
     @Override
     public void handLeZLMServerConfig(ZLMServerConfig zlmServerConfig) {
-        logger.info("[ {} ]-[ {}:{} ]已连接",
+        logger.info("[ ZLM：{} ]-[ {}:{} ]已连接",
                 zlmServerConfig.getGeneralMediaServerId(), zlmServerConfig.getIp(), zlmServerConfig.getHttpPort());
         MediaServerItem serverItem = mediaServerMapper.queryOne(zlmServerConfig.getGeneralMediaServerId());
@@ -469,7 +469,7 @@ public class MediaServerServiceImpl implements IMediaServerService, CommandLineR
      */
     @Override
     public void setZLMConfig(MediaServerItem mediaServerItem) {
-        logger.info("[ {} ]-[ {}:{} ]设置zlm",
+        logger.info("[ ZLM：{} ]-[ {}:{} ]设置zlm",
                 mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
         String protocol = sslEnabled ? "https" : "http";
         String hookPrex = String.format("%s://%s:%s/index/hook", protocol, mediaServerItem.getHookIp(), serverPort);
@@ -494,16 +494,17 @@ public class MediaServerServiceImpl implements IMediaServerService, CommandLineR
         param.put("hook.on_stream_changed",String.format("%s/on_stream_changed", hookPrex));
         param.put("hook.on_stream_none_reader",String.format("%s/on_stream_none_reader", hookPrex));
         param.put("hook.on_stream_not_found",String.format("%s/on_stream_not_found", hookPrex));
+        param.put("hook.on_server_keepalive",String.format("%s/on_server_keepalive", hookPrex));
         param.put("hook.timeoutSec","20");
         param.put("general.streamNoneReaderDelayMS","-1".equals(mediaServerItem.getStreamNoneReaderDelayMS())?"3600000":mediaServerItem.getStreamNoneReaderDelayMS() );
         JSONObject responseJSON = zlmresTfulUtils.setServerConfig(mediaServerItem, param);
         if (responseJSON != null && responseJSON.getInteger("code") == 0) {
-            logger.info("[ {} ]-[ {}:{} ]设置zlm成功",
+            logger.info("[ ZLM：{} ]-[ {}:{} ]设置zlm成功",
                     mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
         }else {
-            logger.info("[ {} ]-[ {}:{} ]设置zlm失败",
+            logger.info("[ ZLM：{} ]-[ {}:{} ]设置zlm失败",
                     mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
         }
     }
@@ -109,4 +109,11 @@ public class StreamPushServiceImpl implements IStreamPushService {
         }
         return del > 0;
     }
+
+
+    @Override
+    public StreamPushItem getPush(String app, String streamId) {
+
+        return streamPushMapper.selectOne(app, streamId);
+    }
 }
@@ -127,7 +127,7 @@ public interface IRedisCatchStorage {
      * 发送推流生成与推流消失消息
      * @param jsonObject 消息内容
      */
-    void sendStreamChangeMsg(JSONObject jsonObject);
+    void sendStreamChangeMsg(String type, JSONObject jsonObject);
     /**
      * 添加流信息到redis
@@ -9,6 +9,8 @@ import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
 import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
 import com.genersoft.iot.vmp.storager.dao.DeviceChannelMapper;
 import com.genersoft.iot.vmp.utils.redis.RedisUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
@@ -19,6 +21,8 @@ import java.util.*;
 @Component
 public class RedisCatchStorageImpl implements IRedisCatchStorage {
+    private Logger logger = LoggerFactory.getLogger(RedisCatchStorageImpl.class);
+
     @Autowired
 	private RedisUtil redis;
@@ -311,8 +315,9 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
     }
     @Override
-    public void sendStreamChangeMsg(JSONObject jsonObject) {
-        String key = VideoManagerConstants.WVP_MSG_STREAM_PUSH_CHANGE_PREFIX;
+    public void sendStreamChangeMsg(String type, JSONObject jsonObject) {
+        String key = VideoManagerConstants.WVP_MSG_STREAM_PUSH_CHANGE_PREFIX + type;
+        logger.debug("[redis 流变化事件] {}: {}", key, jsonObject.toString());
         redis.convertAndSend(key, jsonObject);
     }