修复stream-none-reader-delay-ms为-1时不自动关闭流

64850858
1 parent ea32cd26
Showing 11 changed files with 70 additions and 23 deletions
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorFactory.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/ByeRequestProcessor.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/MessageRequestProcessor.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/response/impl/ByeResponseProcessor.java
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java
src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
src/main/java/com/genersoft/iot/vmp/storager/IVideoManagerStorager.java
src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java
src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStoragerImpl.java
src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java
@@ -159,6 +159,7 @@ public class SIPProcessorFactory {
 			ByeRequestProcessor processor = new ByeRequestProcessor();
 			processor.setRequestEvent(evt);
 			processor.setRedisCatchStorage(redisCatchStorage);
+			processor.setStorager(storager);
 			processor.setZlmrtpServerFactory(zlmrtpServerFactory);
 			processor.setSIPCommander(cmder);
 			processor.setMediaServerService(mediaServerService);
@@ -11,6 +11,8 @@ import javax.sip.header.HeaderAddress;
 import javax.sip.header.ToHeader;
 import javax.sip.message.Response;
+import com.genersoft.iot.vmp.common.StreamInfo;
+import com.genersoft.iot.vmp.gb28181.bean.Device;
 import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
 import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
 import com.genersoft.iot.vmp.gb28181.transmit.request.SIPRequestAbstractProcessor;
@@ -18,6 +20,7 @@ import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
 import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
 import com.genersoft.iot.vmp.service.IMediaServerService;
 import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
+import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -38,6 +41,8 @@ public class ByeRequestProcessor extends SIPRequestAbstractProcessor {
 	private IRedisCatchStorage redisCatchStorage;
+	private IVideoManagerStorager storager;
+
 	private ZLMRTPServerFactory zlmrtpServerFactory;
 	private IMediaServerService mediaServerService;
@@ -56,20 +61,32 @@ public class ByeRequestProcessor extends SIPRequestAbstractProcessor {
 				String platformGbId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(FromHeader.NAME)).getAddress().getURI()).getUser();
 				String channelId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(ToHeader.NAME)).getAddress().getURI()).getUser();
 				SendRtpItem sendRtpItem =  redisCatchStorage.querySendRTPServer(platformGbId, channelId);
-				if (sendRtpItem == null) return;
-				String streamId = sendRtpItem.getStreamId();
-				Map<String, Object> param = new HashMap<>();
-				param.put("vhost","__defaultVhost__");
-				param.put("app",sendRtpItem.getApp());
-				param.put("stream",streamId);
-				param.put("ssrc",sendRtpItem.getSsrc());
-				logger.info("停止向上级推流：" + streamId);
-				MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
-				zlmrtpServerFactory.stopSendRtpStream(mediaInfo, param);
-				redisCatchStorage.deleteSendRTPServer(platformGbId, channelId);
-				if (zlmrtpServerFactory.totalReaderCount(mediaInfo, sendRtpItem.getApp(), streamId) == 0) {
-					logger.info(streamId + "无其它观看者，通知设备停止推流");
-					cmder.streamByeCmd(sendRtpItem.getDeviceId(), channelId);
+				logger.info("收到bye, [{}/{}]", platformGbId, channelId);
+				if (sendRtpItem != null){
+					String streamId = sendRtpItem.getStreamId();
+					Map<String, Object> param = new HashMap<>();
+					param.put("vhost","__defaultVhost__");
+					param.put("app",sendRtpItem.getApp());
+					param.put("stream",streamId);
+					param.put("ssrc",sendRtpItem.getSsrc());
+					logger.info("停止向上级推流：" + streamId);
+					MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
+					zlmrtpServerFactory.stopSendRtpStream(mediaInfo, param);
+					redisCatchStorage.deleteSendRTPServer(platformGbId, channelId);
+					if (zlmrtpServerFactory.totalReaderCount(mediaInfo, sendRtpItem.getApp(), streamId) == 0) {
+						logger.info(streamId + "无其它观看者，通知设备停止推流");
+						cmder.streamByeCmd(sendRtpItem.getDeviceId(), channelId);
+					}
+				}
+				// 可能是设备主动停止
+				Device device = storager.queryVideoDeviceByChannelId(platformGbId);
+				if (device != null) {
+					StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(device.getDeviceId(), channelId);
+					if (streamInfo != null) {
+						redisCatchStorage.stopPlay(streamInfo);
+					}
+					storager.stopPlay(device.getDeviceId(), channelId);
+					mediaServerService.closeRTPServer(device, channelId);
 				}
 			}
 		} catch (SipException e) {
@@ -124,4 +141,12 @@ public class ByeRequestProcessor extends SIPRequestAbstractProcessor {
 	public void setMediaServerService(IMediaServerService mediaServerService) {
 		this.mediaServerService = mediaServerService;
 	}
+
+	public IVideoManagerStorager getStorager() {
+		return storager;
+	}
+
+	public void setStorager(IVideoManagerStorager storager) {
+		this.storager = storager;
+	}
 }
@@ -756,7 +756,6 @@ public class MessageRequestProcessor extends SIPRequestAbstractProcessor {
 				deferredResultHolder.invokeResult(msg);
 			}
 		} catch (DocumentException | SipException | InvalidArgumentException | ParseException e) {
-			// } catch (DocumentException e) {
 			e.printStackTrace();
 		}
 	}
@@ -27,7 +27,7 @@ public class ByeResponseProcessor implements ISIPResponseProcessor {
 	@Override
 	public void process(ResponseEvent evt, SipLayer layer, SipConfig config) {
 		// TODO Auto-generated method stub
-		
+		System.out.println(	);
 	}
 }
@@ -346,6 +346,10 @@ public class ZLMHttpHookListener {
 					redisCatchStorage.stopPlayback(streamInfoForPlayBackCatch);
 				}
 			}
+			MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
+			if (mediaServerItem != null && "-1".equals(mediaServerItem.getStreamNoneReaderDelayMS())) {
+				ret.put("close", false);
+			}
 			return new ResponseEntity<String>(ret.toString(),HttpStatus.OK);
 		}else {
 			JSONObject ret = new JSONObject();
@@ -371,7 +375,7 @@ public class ZLMHttpHookListener {
 		if (userSetup.isAutoApplyPlay() && mediaInfo != null) {
 			String app = json.getString("app");
 			String streamId = json.getString("stream");
-			if ("rtp".equals(app) && streamId.contains("gb_play") ) {
+			if ("rtp".equals(app)) {
 				String[] s = streamId.split("_");
 				if (s.length == 4) {
 					String deviceId = s[2];
@@ -382,7 +386,7 @@ public class ZLMHttpHookListener {
 						SSRCInfo ssrcInfo;
 						String streamId2 = null;
 						if (mediaInfo.isRtpEnable()) {
-							streamId2 = String.format("gb_play_%s_%s", device.getDeviceId(), channelId);
+							streamId2 = String.format("%s/%s", device.getDeviceId(), channelId);
 						}
 						ssrcInfo = mediaServerService.openRTPServer(mediaInfo, streamId2);
 						cmder.playStreamCmd(mediaInfo, ssrcInfo, device, channelId, (MediaServerItem mediaServerItemInuse, JSONObject response) -> {
@@ -123,7 +123,7 @@ public class MediaServerServiceImpl implements IMediaServerService, CommandLineR
         String mediaServerId = streamSession.getMediaServerId(device.getDeviceId(), channelId);
         MediaServerItem mediaServerItem = this.getOne(mediaServerId);
         if (mediaServerItem != null) {
-            String streamId = String.format("gb_play_%s_%s", device.getDeviceId(), channelId);
+            String streamId = String.format("%s/%s", device.getDeviceId(), channelId);
             zlmrtpServerFactory.closeRTPServer(mediaServerItem, streamId);
             releaseSsrc(mediaServerItem, streamSession.getSSRC(device.getDeviceId(), channelId));
         }
@@ -380,7 +380,7 @@ public class MediaServerServiceImpl implements IMediaServerService, CommandLineR
         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.timeoutSec","20");
-        param.put("general.streamNoneReaderDelayMS",mediaServerItem.getStreamNoneReaderDelayMS());
+        param.put("general.streamNoneReaderDelayMS","-1".equals(mediaServerItem.getStreamNoneReaderDelayMS())?"3600000":mediaServerItem.getStreamNoneReaderDelayMS() );
         JSONObject responseJSON = zlmresTfulUtils.setServerConfig(mediaServerItem, param);
@@ -156,7 +156,7 @@ public class PlayServiceImpl implements IPlayService {
             SSRCInfo ssrcInfo;
             String streamId = null;
             if (mediaServerItem.isRtpEnable()) {
-                streamId = String.format("gb_play_%s_%s", device.getDeviceId(), channelId);
+                streamId = String.format("%s/%s", device.getDeviceId(), channelId);
             }
             ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId);
@@ -221,7 +221,7 @@ public class PlayServiceImpl implements IPlayService {
                 SSRCInfo ssrcInfo;
                 String streamId2 = null;
                 if (mediaServerItem.isRtpEnable()) {
-                    streamId2 = String.format("gb_play_%s_%s", device.getDeviceId(), channelId);
+                    streamId2 = String.format("%s/%s", device.getDeviceId(), channelId);
                 }
                 ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId2);
@@ -374,4 +374,6 @@ public interface IVideoManagerStorager {
 	void updateMediaServer(MediaServerItem mediaServerItem);
 	List<StreamProxyItem> getStreamProxyListForEnableInMediaServer(String id, boolean b);
+
+    Device queryVideoDeviceByChannelId(String platformGbId);
 }
@@ -102,4 +102,7 @@ public interface DeviceChannelMapper {
             " </script>"})
     List<ChannelReduce> queryChannelListInAll(String query, Boolean online, Boolean hasSubChannel, String platformId, Boolean inPlatform);
+
+    @Select("SELECT * FROM device_channel WHERE channelId=#{channelId}")
+    List<DeviceChannel> queryChannelByChannelId(String channelId);
 }
@@ -623,4 +623,14 @@ public class VideoManagerStoragerImpl implements IVideoManagerStorager {
 	public List<StreamProxyItem> getStreamProxyListForEnableInMediaServer(String id, boolean enable) {
 		return streamProxyMapper.selectForEnableInMediaServer(id, enable);
 	}
+
+	@Override
+	public Device queryVideoDeviceByChannelId(String channelId) {
+		Device result = null;
+		List<DeviceChannel> channelList = deviceChannelMapper.queryChannelByChannelId(channelId);
+		if (channelList.size() == 1) {
+			result = deviceMapper.getDeviceByDeviceId(channelList.get(0).getDeviceId());
+		}
+		return result;
+	}
 }
@@ -168,7 +168,10 @@ public class PlayController {
 	})
 	@PostMapping("/convert/{streamId}")
 	public ResponseEntity<String> playConvert(@PathVariable String streamId) {
-		StreamInfo streamInfo = streamId.startsWith("gb_play_") ? redisCatchStorage.queryPlayByStreamId(streamId) : redisCatchStorage.queryPlaybackByStreamId(streamId);
+		StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(streamId);
+		if (streamInfo == null) {
+			streamInfo = redisCatchStorage.queryPlaybackByStreamId(streamId);
+		}
 		if (streamInfo == null) {
 			logger.warn("视频转码API调用失败！, 视频流已经停止!");
 			return new ResponseEntity<String>("未找到视频流信息, 视频流可能已经停止", HttpStatus.OK);