实现国标录像级联播放，优化点播流程，加快点播速度

648540858
1 parent 10f77667
Showing 19 changed files with 209 additions and 101 deletions
src/main/java/com/genersoft/iot/vmp/gb28181/bean/SendRtpItem.java
src/main/java/com/genersoft/iot/vmp/gb28181/session/SsrcConfig.java
src/main/java/com/genersoft/iot/vmp/gb28181/task/GPSSubscribeTask.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorObserver.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/response/impl/InviteResponseProcessor.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/ZLMRESTfulUtils.java
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java
src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
src/main/java/com/genersoft/iot/vmp/storager/dao/ParentPlatformMapper.java
src/main/java/com/genersoft/iot/vmp/storager/dao/PlatformChannelMapper.java
src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java
src/test/java/com/genersoft/iot/vmp/service/impl/DeviceAlarmServiceImplTest.java
@@ -71,6 +71,16 @@ public class SendRtpItem {
      */
     private String mediaServerId;
+    /**
+     *  invite的callId
+     */
+    private String CallId;
+
+    /**
+     * 是否是play， false是playback
+     */
+    private boolean isPlay;
+
     public String getIp() {
         return ip;
     }
@@ -174,4 +184,20 @@ public class SendRtpItem {
     public void setMediaServerId(String mediaServerId) {
         this.mediaServerId = mediaServerId;
     }
+
+    public String getCallId() {
+        return CallId;
+    }
+
+    public void setCallId(String callId) {
+        CallId = callId;
+    }
+
+    public boolean isPlay() {
+        return isPlay;
+    }
+
+    public void setPlay(boolean play) {
+        isPlay = play;
+    }
 }
@@ -81,7 +81,6 @@ public class SsrcConfig {
             isUsed.remove(sn);
             notUsed.add(sn);
         }catch (NullPointerException e){
-            System.out.printf("11111");
         }
     }
@@ -36,7 +36,6 @@ public class GPSSubscribeTask implements Runnable{
         SubscribeInfo subscribe = redisCatchStorage.getSubscribe(key);
         if (subscribe != null) {
-            System.out.println("发送GPS消息");
             ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(platformId);
             if (parentPlatform == null || parentPlatform.isStatus()) {
                 // TODO 暂时只处理视频流的回复,后续增加对国标设备的支持
@@ -141,7 +141,6 @@ public class SIPProcessorObserver implements ISIPProcessorObserver {
      */
     @Override
     public void processTimeout(TimeoutEvent timeoutEvent) {
-        System.out.println("processTimeout");
         if(timeoutProcessor != null) {
             timeoutProcessor.process(timeoutEvent);
         }
@@ -173,7 +172,6 @@ public class SIPProcessorObserver implements ISIPProcessorObserver {
     @Override
     public void processDialogTerminated(DialogTerminatedEvent dialogTerminatedEvent) {
-        System.out.println("processDialogTerminated");
         CallIdHeader callId = dialogTerminatedEvent.getDialog().getCallId();
     }
@@ -346,8 +346,11 @@ public class SIPCommander implements ISIPCommander {
 			subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,
 					(MediaServerItem mediaServerItemInUse, JSONObject json)->{
 				if (userSetup.isWaitTrack() && json.getJSONArray("tracks") == null) return;
-				event.response(mediaServerItemInUse, json);
-				subscribe.removeSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey);
+				if (event != null) {
+					event.response(mediaServerItemInUse, json);
+				}
+
+//				subscribe.removeSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey);
 			});
 			//
 			StringBuffer content = new StringBuffer(200);
@@ -452,9 +455,11 @@ public class SIPCommander implements ISIPCommander {
 			logger.debug("录像回放添加订阅，订阅内容：" + subscribeKey.toString());
 			subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,
 					(MediaServerItem mediaServerItemInUse, JSONObject json)->{
+						System.out.println(344444);
 				if (userSetup.isWaitTrack() && json.getJSONArray("tracks") == null) return;
-				event.response(mediaServerItemInUse, json);
-				subscribe.removeSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey);
+				if (event != null) {
+					event.response(mediaServerItemInUse, json);
+				}
 			});
 			StringBuffer content = new StringBuffer(200);
@@ -466,8 +471,6 @@ public class SIPCommander implements ISIPCommander {
 	        content.append("t="+DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(startTime)+" "
 					+DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(endTime) +"\r\n");
-
-
 			String streamMode = device.getStreamMode().toUpperCase();
 			if (userSetup.isSeniorSdp()) {
@@ -1202,7 +1205,6 @@ public class SIPCommander implements ISIPCommander {
 		if (type == null) {
 			type = "all";
 		}
-
 		try {
 			StringBuffer recordInfoXml = new StringBuffer(200);
 			recordInfoXml.append("<?xml version=\"1.0\" encoding=\"GB2312\"?>\r\n");
@@ -508,9 +508,6 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
             // callid
             CallIdHeader callIdHeader = parentPlatform.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()
                     : udpSipProvider.getNewCallId();
-            System.out.println(
-                    recordXml.toString()
-            );
             Request request = headerProviderPlarformProvider.createMessageRequest(parentPlatform, recordXml.toString(), fromTag, callIdHeader);
             transmitRequest(parentPlatform, request);
 package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl;
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
 import com.genersoft.iot.vmp.common.StreamInfo;
 import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
 import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver;
 import com.genersoft.iot.vmp.gb28181.transmit.event.request.ISIPRequestProcessor;
 import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
+import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
 import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
 import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
 import com.genersoft.iot.vmp.service.IMediaServerService;
@@ -24,6 +27,8 @@ import javax.sip.header.HeaderAddress;
 import javax.sip.header.ToHeader;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Timer;
+import java.util.TimerTask;
 /**
  * SIP命令类型： ACK请求
@@ -52,6 +57,9 @@ public class AckRequestProcessor extends SIPRequestProcessorParent implements In
 	@Autowired
 	private IMediaServerService mediaServerService;
+	@Autowired
+	private ZLMHttpHookSubscribe subscribe;
+
 	/**   
 	 * 处理  ACK请求
@@ -60,6 +68,7 @@ public class AckRequestProcessor extends SIPRequestProcessorParent implements In
 	 */
 	@Override
 	public void process(RequestEvent evt) {
+		logger.debug("ACK请求： {}", ((System.currentTimeMillis())));
 		Dialog dialog = evt.getDialog();
 		if (dialog == null) return;
 		if (dialog.getState()== DialogState.CONFIRMED) {
@@ -69,16 +78,17 @@ public class AckRequestProcessor extends SIPRequestProcessorParent implements In
 			String is_Udp = sendRtpItem.isTcp() ? "0" : "1";
 			String deviceId = sendRtpItem.getDeviceId();
 			StreamInfo streamInfo = null;
-			if (deviceId == null) {
+			if (sendRtpItem.isPlay()) {
+				streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId);
+			}else {
+				streamInfo = redisCatchStorage.queryPlaybackByDevice(deviceId, channelId);
+			}
+			System.out.println(JSON.toJSON(streamInfo));
+			if (streamInfo == null) {
 				streamInfo = new StreamInfo();
 				streamInfo.setApp(sendRtpItem.getApp());
 				streamInfo.setStreamId(sendRtpItem.getStreamId());
-			}else {
-				streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId);
-				sendRtpItem.setStreamId(streamInfo.getStreamId());
-				streamInfo.setApp("rtp");
 			}
-
 			redisCatchStorage.updateSendRTPSever(sendRtpItem);
 			logger.info(platformGbId);
 			logger.info(channelId);
@@ -90,34 +100,42 @@ public class AckRequestProcessor extends SIPRequestProcessorParent implements In
 			param.put("dst_url",sendRtpItem.getIp());
 			param.put("dst_port", sendRtpItem.getPort());
 			param.put("is_udp", is_Udp);
-			//param.put ("src_port", sendRtpItem.getLocalPort());
 			// 设备推流查询，成功后才能转推
-			boolean rtpPushed = false;
-			long startTime = System.currentTimeMillis();
-			while (!rtpPushed) {
-				try {
-					if (System.currentTimeMillis() - startTime < 30 * 1000) {
-						MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
-						if (zlmrtpServerFactory.isStreamReady(mediaInfo, streamInfo.getApp(), streamInfo.getStreamId())) {
-							rtpPushed = true;
-							logger.info("已获取设备推流[{}/{}]，开始向上级推流[{}:{}]",
-									streamInfo.getApp() ,streamInfo.getStreamId(), sendRtpItem.getIp(), sendRtpItem.getPort());
-							zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
-						} else {
-							logger.info("等待设备推流[{}/{}].......",
-									streamInfo.getApp() ,streamInfo.getStreamId());
-							Thread.sleep(1000);
-							continue;
-						}
-					} else {
-						rtpPushed = true;
-						logger.info("设备推流[{}/{}]超时，终止向上级推流",
-								streamInfo.getApp() ,streamInfo.getStreamId());
-					}
-				} catch (InterruptedException e) {
-					e.printStackTrace();
-				}
-			}
+			MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
+			zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
+//			if (zlmrtpServerFactory.isStreamReady(mediaInfo, streamInfo.getApp(), streamInfo.getStreamId())) {
+//				logger.info("已获取设备推流[{}/{}]，开始向上级推流[{}:{}]",
+//						streamInfo.getApp() ,streamInfo.getStreamId(), sendRtpItem.getIp(), sendRtpItem.getPort());
+//				zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
+//			} else {
+//				// 对hook进行订阅
+//				logger.info("等待设备推流[{}/{}].......",
+//						streamInfo.getApp(), streamInfo.getStreamId());
+//				Timer timer = new Timer();
+//				timer.schedule(new TimerTask() {
+//					@Override
+//					public void run() {
+//						logger.info("设备推流[{}/{}]超时，终止向上级推流",
+//								finalStreamInfo.getApp() , finalStreamInfo.getStreamId());
+//
+//					}
+//				}, 30*1000L);
+//				// 添加订阅
+//				JSONObject subscribeKey = new JSONObject();
+//				subscribeKey.put("app", "rtp");
+//				subscribeKey.put("stream", streamInfo.getStreamId());
+//				subscribeKey.put("mediaServerId", streamInfo.getMediaServerId());
+//				subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_publish, subscribeKey,
+//						(MediaServerItem mediaServerItemInUse, JSONObject json) -> {
+//							logger.info("已获取设备推流[{}/{}]，开始向上级推流[{}:{}]",
+//									finalStreamInfo.getApp(), finalStreamInfo.getStreamId(), sendRtpItem.getIp(), sendRtpItem.getPort());
+//							timer.cancel();
+//							zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
+//							subscribe.removeSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey);
+//						});
+//			}
+
+
 		}
 	}
 }
@@ -87,18 +87,29 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
 					MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
 					zlmrtpServerFactory.stopSendRtpStream(mediaInfo, param);
 					redisCatchStorage.deleteSendRTPServer(platformGbId, channelId);
-					if (zlmrtpServerFactory.totalReaderCount(mediaInfo, sendRtpItem.getApp(), streamId) == 0) {
+					int totalReaderCount = zlmrtpServerFactory.totalReaderCount(mediaInfo, sendRtpItem.getApp(), streamId);
+					if (totalReaderCount == 0) {
 						logger.info(streamId + "无其它观看者，通知设备停止推流");
 						cmder.streamByeCmd(sendRtpItem.getDeviceId(), channelId);
+					}else if (totalReaderCount == -1){
+						logger.warn(streamId + " 查找其它观看者失败");
 					}
 				}
 				// 可能是设备主动停止
 				Device device = storager.queryVideoDeviceByChannelId(platformGbId);
 				if (device != null) {
-					StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(device.getDeviceId(), channelId);
-					if (streamInfo != null) {
-						redisCatchStorage.stopPlay(streamInfo);
+					if (sendRtpItem.isPlay()) {
+						StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(device.getDeviceId(), channelId);
+						if (streamInfo != null) {
+							redisCatchStorage.stopPlay(streamInfo);
+						}
+					}else {
+						StreamInfo streamInfo = redisCatchStorage.queryPlaybackByDevice(device.getDeviceId(), channelId);
+						if (streamInfo != null) {
+							redisCatchStorage.stopPlayback(streamInfo);
+						}
 					}
+
 					storager.stopPlay(device.getDeviceId(), channelId);
 					mediaServerService.closeRTPServer(device, channelId);
 				}
 package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl;
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
+import com.genersoft.iot.vmp.common.StreamInfo;
 import com.genersoft.iot.vmp.gb28181.bean.*;
+import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
 import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver;
+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.impl.SIPCommander;
 import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
 import com.genersoft.iot.vmp.gb28181.transmit.event.request.ISIPRequestProcessor;
 import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
+import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
 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.service.IPlayService;
+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.vmanager.gb28181.play.bean.PlayResult;
+import gov.nist.javax.sdp.TimeDescriptionImpl;
+import gov.nist.javax.sdp.fields.TimeField;
 import gov.nist.javax.sip.address.AddressImpl;
 import gov.nist.javax.sip.address.SipUri;
 import org.slf4j.Logger;
@@ -27,10 +37,13 @@ import javax.sip.RequestEvent;
 import javax.sip.ServerTransaction;
 import javax.sip.SipException;
 import javax.sip.address.SipURI;
+import javax.sip.header.CallIdHeader;
 import javax.sip.header.FromHeader;
 import javax.sip.message.Request;
 import javax.sip.message.Response;
 import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.Date;
 import java.util.List;
 import java.util.Vector;
@@ -61,6 +74,9 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
 	private IPlayService playService;
 	@Autowired
+	private ISIPCommander commander;
+
+	@Autowired
 	private ZLMRTPServerFactory zlmrtpServerFactory;
 	@Autowired
@@ -69,6 +85,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
 	@Autowired
 	private SIPProcessorObserver sipProcessorObserver;
+
 	@Override
 	public void afterPropertiesSet() throws Exception {
 		// 添加消息处理的订阅
@@ -84,6 +101,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
 	@Override
 	public void process(RequestEvent evt) {
 		//  Invite Request消息实现，此消息一般为级联消息，上级给下级发送请求视频指令
+		Long startTimeForInvite = System.currentTimeMillis();
 		try {
 			Request request = evt.getRequest();
 			SipURI sipURI = (SipURI) request.getRequestURI();
@@ -91,6 +109,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
 			String requesterId = null;
 			FromHeader fromHeader = (FromHeader)request.getHeader(FromHeader.NAME);
+			CallIdHeader callIdHeader = (CallIdHeader)request.getHeader(CallIdHeader.NAME);
 			AddressImpl address = (AddressImpl) fromHeader.getAddress();
 			SipUri uri = (SipUri) address.getURI();
 			requesterId = uri.getUser();
@@ -101,7 +120,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
 				return;
 			}
-			// 查询请求方是否上级平台
+			// 查询请求是否来自上级平台\设备
 			ParentPlatform platform = storager.queryParentPlatByServerGBId(requesterId);
 			if (platform != null) {
 				// 查询平台下是否有该通道
@@ -158,7 +177,21 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
 					ssrc = ssrcDefault;
 					sdp = SdpFactory.getInstance().createSessionDescription(contentString);
 				}
-
+				String sessionName = sdp.getSessionName().getValue();
+
+				Long startTime = null;
+				Long stopTime = null;
+				Date start = null;
+				Date end = null;
+				if (sdp.getTimeDescriptions(false) != null && sdp.getTimeDescriptions(false).size() > 0) {
+					TimeDescriptionImpl timeDescription = (TimeDescriptionImpl)(sdp.getTimeDescriptions(false).get(0));
+					TimeField startTimeFiled = (TimeField)timeDescription.getTime();
+					startTime = startTimeFiled.getStartTime();
+					stopTime = startTimeFiled.getStopTime();
+
+					start = new Date(startTime*1000);
+					end = new Date(stopTime*1000);
+				}
 				//  获取支持的格式
 				Vector mediaDescriptions = sdp.getMediaDescriptions(true);
 				// 查看是否支持PS 负载96
@@ -228,23 +261,31 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
 						responseAck(evt, Response.BUSY_HERE);
 						return;
 					}
-
+					sendRtpItem.setCallId(callIdHeader.getCallId());
+					sendRtpItem.setPlay("Play".equals(sessionName));
 					// 写入redis， 超时时回复
 					redisCatchStorage.updateSendRTPSever(sendRtpItem);
-					// 通知下级推流，
-					PlayResult playResult = playService.play(mediaServerItem,device.getDeviceId(), channelId, (mediaServerItemInUSe, responseJSON)->{
-						// 收到推流， 回复200OK, 等待ack
+
+					Device finalDevice = device;
+					MediaServerItem finalMediaServerItem = mediaServerItem;
+					Long finalStartTime = startTime;
+					Long finalStopTime = stopTime;
+					ZLMHttpHookSubscribe.Event hookEvent = (mediaServerItemInUSe, responseJSON)->{
+						logger.info("[上级点播]收到下级开始点播订阅， {}/{}", sendRtpItem.getApp(), sendRtpItem.getStreamId());
 						// if (sendRtpItem == null) return;
 						sendRtpItem.setStatus(1);
 						redisCatchStorage.updateSendRTPSever(sendRtpItem);
-						// TODO 添加对tcp的支持
 						StringBuffer content = new StringBuffer(200);
 						content.append("v=0\r\n");
 						content.append("o="+ channelId +" 0 0 IN IP4 "+mediaServerItemInUSe.getSdpIp()+"\r\n");
-						content.append("s=Play\r\n");
+						content.append("s=" + sessionName+"\r\n");
 						content.append("c=IN IP4 "+mediaServerItemInUSe.getSdpIp()+"\r\n");
-						content.append("t=0 0\r\n");
+						if ("Playback".equals(sessionName)) {
+							content.append("t=" + finalStartTime + " " + finalStopTime + "\r\n");
+						}else {
+							content.append("t=0 0\r\n");
+						}
 						content.append("m=video "+ sendRtpItem.getLocalPort()+" RTP/AVP 96\r\n");
 						content.append("a=sendonly\r\n");
 						content.append("a=rtpmap:96 PS/90000\r\n");
@@ -260,7 +301,11 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
 						} catch (ParseException e) {
 							e.printStackTrace();
 						}
-					} ,((event) -> {
+						if ("Playback".equals(sessionName) && responseJSON != null) {
+							playService.onPublishHandlerForPlayBack(finalMediaServerItem, responseJSON, finalDevice.getDeviceId(), channelId, null);
+						}
+					};
+					SipSubscribe.Event errorEvent = ((event) -> {
 						// 未知错误。直接转发设备点播的错误
 						Response response = null;
 						try {
@@ -271,11 +316,27 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
 						} catch (ParseException | SipException | InvalidArgumentException e) {
 							e.printStackTrace();
 						}
-					}));
-					if (logger.isDebugEnabled()) {
-						logger.debug(playResult.getResult().toString());
+					});
+					if ("Playback".equals(sessionName)) {
+						sendRtpItem.setPlay(false);
+						SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, sendRtpItem.getSsrc(), true);
+						sendRtpItem.setStreamId(ssrc);
+						SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+						commander.playbackStreamCmd(mediaServerItem, ssrcInfo, device, channelId, format.format(start), format.format(end), hookEvent, errorEvent);
+					}else {
+						sendRtpItem.setPlay(true);
+						StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(device.getDeviceId(), channelId);
+						if (streamInfo == null) {
+							if (mediaServerItem.isRtpEnable()) {
+								sendRtpItem.setStreamId(String.format("%s_%s", device.getDeviceId(), channelId));
+							}
+							sendRtpItem.setPlay(false);
+							playService.play(mediaServerItem,device.getDeviceId(), channelId, hookEvent,errorEvent);
+						}else {
+							sendRtpItem.setStreamId(streamInfo.getStreamId());
+							hookEvent.response(mediaServerItem, null);
+						}
 					}
-
 				}else if (gbStream != null) {
 					SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
 							gbStream.getApp(), gbStream.getStream(), channelId,
@@ -295,7 +356,6 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
 					sendRtpItem.setStatus(1);
 					redisCatchStorage.updateSendRTPSever(sendRtpItem);
-					// TODO 添加对tcp的支持
 					StringBuffer content = new StringBuffer(200);
 					content.append("v=0\r\n");
 					content.append("o="+ channelId +" 0 0 IN IP4 "+mediaServerItem.getSdpIp()+"\r\n");
@@ -82,9 +82,6 @@ public class InviteResponseProcessor extends SIPResponseProcessorAbstract {
 				requestURI.setPort(event.getRemotePort());
 				reqAck.setRequestURI(requestURI);
 				logger.info("向 " + event.getRemoteIpAddress() + ":" + event.getRemotePort() + "回复ack");
-				SipURI sipURI = (SipURI)dialog.getRemoteParty().getURI();
-				String deviceId = requestURI.getUser();
-				String channelId = sipURI.getUser();
 				dialog.sendAck(reqAck);
@@ -181,7 +181,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.info("[ ZLM HOOK ]on_publish API调用，参数：" + json.toString());
 		JSONObject ret = new JSONObject();
 		ret.put("code", 0);
 		ret.put("msg", "success");
@@ -77,21 +77,23 @@ public class ZLMHttpHookSubscribe {
         if (eventMap == null) {
             return;
         }
-        Iterator<Map.Entry<JSONObject, Event>> iterator = eventMap.entrySet().iterator();
-        while (iterator.hasNext()){
-            Map.Entry<JSONObject, Event> next = iterator.next();
-            JSONObject key = next.getKey();
-            Boolean result = null;
-            for (String s : key.keySet()) {
-                if (result == null) {
-                    result = key.getString(s).equals(hookResponse.getString(s));
-                }else {
-                    if (key.getString(s) == null) continue;
-                    result = result && key.getString(s).equals(hookResponse.getString(s));
+
+        Set<Map.Entry<JSONObject, Event>> entries = eventMap.entrySet();
+        if (entries.size() > 0) {
+            for (Map.Entry<JSONObject, Event> entry : entries) {
+                JSONObject key = entry.getKey();
+                Boolean result = null;
+                for (String s : key.keySet()) {
+                    if (result == null) {
+                        result = key.getString(s).equals(hookResponse.getString(s));
+                    }else {
+                        if (key.getString(s) == null) continue;
+                        result = result && key.getString(s).equals(hookResponse.getString(s));
+                    }
+                }
+                if (null != result && result){
+                    entries.remove(entry);
                 }
-            }
-            if (null != result && result){
-                iterator.remove();
             }
         }
     }
@@ -72,7 +72,6 @@ public class ZLMRESTfulUtils {
                         ResponseBody responseBody = response.body();
                         if (responseBody != null) {
                             String responseStr = responseBody.string();
-                            System.out.println(responseStr);
                             responseJSON = JSON.parseObject(responseStr);
                         }
                     }else {
@@ -242,9 +242,18 @@ public class ZLMRTPServerFactory {
      */
     public int totalReaderCount(MediaServerItem mediaServerItem, String app, String streamId) {
         JSONObject mediaInfo = zlmresTfulUtils.getMediaInfo(mediaServerItem, app, "rtmp", streamId);
+        Integer code = mediaInfo.getInteger("code");
         if (mediaInfo == null) {
             return 0;
         }
+        if ( code < 0) {
+            logger.warn("查询流({}/{})是否有其它观看者时得到： {}", app, streamId, mediaInfo.getString("msg"));
+            return -1;
+        }
+        if ( code == 0 && ! mediaInfo.getBoolean("online")) {
+            logger.warn("查询流({}/{})是否有其它观看者时得到： {}", app, streamId, mediaInfo.getString("msg"));
+            return -1;
+        }
         return mediaInfo.getInteger("totalReaderCount");
     }
@@ -122,7 +122,6 @@ public class PlayServiceImpl implements IPlayService {
             // 点播结束时调用截图接口
             try {
                 String classPath = ResourceUtils.getURL("classpath:").getPath();
-                // System.out.println(classPath);
                 // 兼容打包为jar的class路径
                 if(classPath.contains("jar")) {
                     classPath = classPath.substring(0, classPath.lastIndexOf("."));
@@ -238,11 +237,11 @@ public class PlayServiceImpl implements IPlayService {
     }
     @Override
-    public void onPublishHandlerForPlay(MediaServerItem mediaServerItem, JSONObject resonse, String deviceId, String channelId, String uuid) {
+    public void onPublishHandlerForPlay(MediaServerItem mediaServerItem, JSONObject response, String deviceId, String channelId, String uuid) {
         RequestMessage msg = new RequestMessage();
         msg.setId(uuid);
         msg.setKey(DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId);
-        StreamInfo streamInfo = onPublishHandler(mediaServerItem, resonse, deviceId, channelId, uuid);
+        StreamInfo streamInfo = onPublishHandler(mediaServerItem, response, deviceId, channelId, uuid);
         if (streamInfo != null) {
             DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
             if (deviceChannel != null) {
@@ -88,7 +88,7 @@ public interface ParentPlatformMapper {
             "</script>"})
     int setDefaultCatalog(String platformId, String catalogId);
-    @Select("select 'channel' as name, count(pgc.platformId) count from platform_gb_channel pgc  where  pgc.platformId=#{platformId} and pgc.channelId =#{gbId} " +
+    @Select("select 'channel' as name, count(pgc.platformId) count from platform_gb_channel pgc left join device_channel dc on dc.id = pgc.deviceChannelId where  pgc.platformId=#{platformId} and dc.channelId =#{gbId} " +
             "union " +
             "select 'stream' as name, count(pgs.platformId) count from platform_gb_stream pgs left join gb_stream gs on pgs.gbStreamId = gs.gbStreamId where  pgs.platformId=#{platformId} and gs.gbId = #{gbId}")
     List<ChannelSourceInfo> getChannelSource(String platformId, String gbId);
@@ -52,8 +52,8 @@ public interface PlatformChannelMapper {
     int cleanChannelForGB(String platformId);
-    @Select("SELECT * FROM device_channel WHERE deviceId = (SELECT deviceId FROM platform_gb_channel WHERE " +
-            "platformId='${platformId}' AND channelId='${channelId}' ) AND channelId='${channelId}'")
+    @Select("SELECT dc.* FROM platform_gb_channel pgc left join device_channel dc on dc.id = pgc.deviceChannelId WHERE " +
+            "pgc.platformId=#{platformId} AND dc.channelId=#{channelId}")
     DeviceChannel queryChannelInParentPlatform(String platformId, String channelId);
@@ -62,7 +62,7 @@ public interface PlatformChannelMapper {
             "where pgc.platformId=#{platformId} and pgc.catalogId=#{catalogId}")
     List<PlatformCatalog> queryChannelInParentPlatformAndCatalog(String platformId, String catalogId);
-    @Select("SELECT * FROM device WHERE deviceId = (SELECT deviceId FROM platform_gb_channel WHERE platformId='${platformId}' AND channelId='${channelId}')")
+    @Select("SELECT * FROM device WHERE deviceId = (SELECT deviceId FROM platform_gb_channel pgc left join device_channel dc on dc.id = pgc.deviceChannelId WHERE pgc.platformId='${platformId}' AND dc.channelId='${channelId}')")
     Device queryVideoDeviceByPlatformIdAndChannelId(String platformId, String channelId);
     @Delete("<script> "+
@@ -71,7 +71,7 @@ public interface PlatformChannelMapper {
     int delByCatalogId(String id);
     @Delete("<script> "+
-           "DELETE FROM platform_gb_channel WHERE catalogId=#{parentId} AND platformId=#{platformId} AND channelId=#{id}"  +
+           "DELETE FROM platform_gb_channel  WHERE catalogId=#{parentId} AND platformId=#{platformId} AND channelId=#{id}"  +
            "</script>")
     int delByCatalogIdAndChannelIdAndPlatformId(PlatformCatalog platformCatalog);
@@ -139,7 +139,6 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
     @Override
     public StreamInfo queryPlayByDevice(String deviceId, String channelId) {
-//		List<Object> playLeys = redis.keys(String.format("%S_*_%s_%s", VideoManagerConstants.PLAYER_PREFIX,
         List<Object> playLeys = redis.scan(String.format("%S_%s_*_%s_%s", VideoManagerConstants.PLAYER_PREFIX,
                 userSetup.getServerId(),
                 deviceId,
@@ -50,14 +50,7 @@ class DeviceAlarmServiceImplTest {
 //        System.out.println(deviceAlarmService.getAllAlarm(0, 10000, "11111111111111111111", null, "1", null,
 //                null, null).getSize());
-        System.out.println(deviceAlarmService.getAllAlarm(0, 10000, "11111111111111111111", null, null, null,
-                "2021-01-01 00:00:00", null).getSize());
-        System.out.println(deviceAlarmService.getAllAlarm(0, 10000, "11111111111111111111", null, null, null,
-                null, "2021-04-01 09:00:00").getSize());
-
-        System.out.println(deviceAlarmService.getAllAlarm(0, 10000, "11111111111111111111", null, null, null,
-                "2021-02-01 01:00:00", "2021-04-01 04:00:00").getSize());
     }