From bfae9780f75db7495f53511f3116bb6c0470a0b0 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期三, 28 六月 2023 20:32:48 +0800
Subject: [PATCH] 增加流关闭时的处理

---
 src/main/java/com/genersoft/iot/vmp/conf/redis/RedisMsgListenConfig.java                         |    5 
 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java                  |    7 
 src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java                            |    2 
 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java |  166 ++++++++++++++++-----------------
 src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamCloseResponseListener.java   |   98 ++++++++++++++-----
 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderPlarformProvider.java   |    8 -
 6 files changed, 163 insertions(+), 123 deletions(-)

diff --git a/src/main/java/com/genersoft/iot/vmp/conf/redis/RedisMsgListenConfig.java b/src/main/java/com/genersoft/iot/vmp/conf/redis/RedisMsgListenConfig.java
index a65491b..762d8e1 100644
--- a/src/main/java/com/genersoft/iot/vmp/conf/redis/RedisMsgListenConfig.java
+++ b/src/main/java/com/genersoft/iot/vmp/conf/redis/RedisMsgListenConfig.java
@@ -43,6 +43,9 @@
 	@Autowired
 	private RedisPushStreamResponseListener redisPushStreamResponseListener;
 
+	@Autowired
+	private RedisPushStreamCloseResponseListener redisPushStreamCloseResponseListener;
+
 
 	/**
 	 * redis娑堟伅鐩戝惉鍣ㄥ鍣� 鍙互娣诲姞澶氫釜鐩戝惉涓嶅悓璇濋鐨剅edis鐩戝惉鍣紝鍙渶瑕佹妸娑堟伅鐩戝惉鍣ㄥ拰鐩稿簲鐨勬秷鎭闃呭鐞嗗櫒缁戝畾锛岃娑堟伅鐩戝惉鍣�
@@ -63,7 +66,7 @@
 		container.addMessageListener(redisPushStreamStatusMsgListener, new PatternTopic(VideoManagerConstants.VM_MSG_PUSH_STREAM_STATUS_CHANGE));
 		container.addMessageListener(redisPushStreamListMsgListener, new PatternTopic(VideoManagerConstants.VM_MSG_PUSH_STREAM_LIST_CHANGE));
 		container.addMessageListener(redisPushStreamResponseListener, new PatternTopic(VideoManagerConstants.VM_MSG_STREAM_PUSH_RESPONSE));
-//		container.addMessageListener(, new PatternTopic(VideoManagerConstants.VM_MSG_STREAM_PUSH_CLOSE_REQUESTED));
+		container.addMessageListener(redisPushStreamCloseResponseListener, new PatternTopic(VideoManagerConstants.VM_MSG_STREAM_PUSH_CLOSE_REQUESTED));
         return container;
     }
 }
diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderPlarformProvider.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderPlarformProvider.java
index 831897a..e19886b 100644
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderPlarformProvider.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderPlarformProvider.java
@@ -283,7 +283,7 @@
 		viaHeader.setRPort();
 		viaHeaders.add(viaHeader);
 		// from
-		SipURI fromSipURI = SipFactory.getInstance().createAddressFactory().createSipURI(platform.getDeviceGBId(),
+		SipURI fromSipURI = SipFactory.getInstance().createAddressFactory().createSipURI(sendRtpItem.getChannelId(),
 				platform.getDeviceIp() + ":" + platform.getDevicePort());
 		Address fromAddress = SipFactory.getInstance().createAddressFactory().createAddress(fromSipURI);
 		FromHeader fromHeader = SipFactory.getInstance().createHeaderFactory().createFromHeader(fromAddress, sendRtpItem.getToTag());
@@ -296,13 +296,10 @@
 		MaxForwardsHeader maxForwards = SipFactory.getInstance().createHeaderFactory().createMaxForwardsHeader(70);
 		// ceq
 		CSeqHeader cSeqHeader = SipFactory.getInstance().createHeaderFactory().createCSeqHeader(redisCatchStorage.getCSEQ(), Request.BYE);
-		MessageFactoryImpl messageFactory = (MessageFactoryImpl) SipFactory.getInstance().createMessageFactory();
-		// 璁剧疆缂栫爜锛� 闃叉涓枃涔辩爜
-		messageFactory.setDefaultContentEncodingCharset("gb2312");
 
 		CallIdHeader callIdHeader = SipFactory.getInstance().createHeaderFactory().createCallIdHeader(sendRtpItem.getCallId());
 
-		request = (SIPRequest) messageFactory.createRequest(requestURI, Request.BYE, callIdHeader, cSeqHeader, fromHeader,
+		request = (SIPRequest) SipFactory.getInstance().createMessageFactory().createRequest(requestURI, Request.BYE, callIdHeader, cSeqHeader, fromHeader,
 				toHeader, viaHeaders, maxForwards);
 
 		request.addHeader(SipUtils.createUserAgentHeader(gitUtil));
@@ -310,6 +307,7 @@
 		String sipAddress = platform.getDeviceIp() + ":" + platform.getDevicePort();
 		Address concatAddress = SipFactory.getInstance().createAddressFactory().createAddress(SipFactory.getInstance().createAddressFactory()
 				.createSipURI(platform.getDeviceGBId(), sipAddress));
+
 		request.addHeader(SipFactory.getInstance().createHeaderFactory().createContactHeader(concatAddress));
 
 		return request;
diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
index beec9e9..df48e8e 100644
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
@@ -347,10 +347,9 @@
             mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
             errorEvent.response(e);
         }), e -> {
-            // 杩欓噷涓轰緥閬垮厤涓�涓�氶亾鐨勭偣鎾彧鏈変竴涓猚allID杩欎釜鍙傛暟浣跨敤涓�涓浐瀹氬��
             ResponseEvent responseEvent = (ResponseEvent) e.event;
             SIPResponse response = (SIPResponse) responseEvent.getResponse();
-            streamSession.put(device.getDeviceId(), channelId, "play", stream, ssrcInfo.getSsrc(), mediaServerItem.getId(), response, VideoStreamSessionManager.SessionType.play);
+            streamSession.put(device.getDeviceId(), channelId, e.callId, stream, ssrcInfo.getSsrc(), mediaServerItem.getId(), response, VideoStreamSessionManager.SessionType.play);
             okEvent.response(e);
         });
     }
@@ -452,7 +451,7 @@
         sipSender.transmitRequest(sipLayer.getLocalIp(device.getLocalIp()), request, errorEvent, event -> {
             ResponseEvent responseEvent = (ResponseEvent) event.event;
             SIPResponse response = (SIPResponse) responseEvent.getResponse();
-            streamSession.put(device.getDeviceId(), channelId,sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()),device.getTransport()).getCallId(), ssrcInfo.getStream(), ssrcInfo.getSsrc(), mediaServerItem.getId(), response, VideoStreamSessionManager.SessionType.playback);
+            streamSession.put(device.getDeviceId(), channelId, event.callId, ssrcInfo.getStream(), ssrcInfo.getSsrc(), mediaServerItem.getId(), response, VideoStreamSessionManager.SessionType.playback);
             okEvent.response(event);
         });
         if (inviteStreamCallback != null) {
@@ -580,7 +579,7 @@
             if (ssrcIndex >= 0) {
                 ssrc = contentString.substring(ssrcIndex + 2, ssrcIndex + 12);
             }
-            streamSession.put(device.getDeviceId(), channelId, response.getCallIdHeader().getCallId(), ssrcInfo.getStream(), ssrc, mediaServerItem.getId(), response, VideoStreamSessionManager.SessionType.download);
+            streamSession.put(device.getDeviceId(), channelId, newCallIdHeader.getCallId(), ssrcInfo.getStream(), ssrc, mediaServerItem.getId(), response, VideoStreamSessionManager.SessionType.download);
             okEvent.response(event);
         });
     }
diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java
index 78e2956..8c93ae3 100644
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java
@@ -11,6 +11,7 @@
 import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
 import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
 import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
+import com.genersoft.iot.vmp.service.IDeviceChannelService;
 import com.genersoft.iot.vmp.service.IDeviceService;
 import com.genersoft.iot.vmp.service.IMediaServerService;
 import com.genersoft.iot.vmp.service.IPlatformService;
@@ -27,11 +28,7 @@
 import javax.sip.InvalidArgumentException;
 import javax.sip.RequestEvent;
 import javax.sip.SipException;
-import javax.sip.address.SipURI;
 import javax.sip.header.CallIdHeader;
-import javax.sip.header.FromHeader;
-import javax.sip.header.HeaderAddress;
-import javax.sip.header.ToHeader;
 import javax.sip.message.Response;
 import java.text.ParseException;
 import java.util.HashMap;
@@ -57,6 +54,9 @@
 
 	@Autowired
 	private IDeviceService deviceService;
+
+	@Autowired
+	private IDeviceChannelService channelService;
 
 	@Autowired
 	private IVideoManagerStorage storager;
@@ -88,95 +88,91 @@
 	 */
 	@Override
 	public void process(RequestEvent evt) {
-
+		SIPRequest request = (SIPRequest) evt.getRequest();
 		try {
-			responseAck((SIPRequest) evt.getRequest(), Response.OK);
+			responseAck(request, Response.OK);
 		} catch (SipException | InvalidArgumentException | ParseException e) {
 			logger.error("[鍥炲BYE淇℃伅澶辫触]锛寋}", e.getMessage());
 		}
 		CallIdHeader callIdHeader = (CallIdHeader)evt.getRequest().getHeader(CallIdHeader.NAME);
-			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, null, callIdHeader.getCallId());
-			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("[鏀跺埌bye] 鍋滄鍚戜笂绾ф帹娴侊細{}", streamId);
-				MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
-				redisCatchStorage.deleteSendRTPServer(platformGbId, channelId, callIdHeader.getCallId(), null);
-				zlmrtpServerFactory.stopSendRtpStream(mediaInfo, param);
-				if (sendRtpItem.getPlayType().equals(InviteStreamType.PUSH)) {
-					ParentPlatform platform = platformService.queryPlatformByServerGBId(sendRtpItem.getPlatformId());
-					if (platform != null) {
-						MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(0,
-								sendRtpItem.getApp(), sendRtpItem.getStreamId(), sendRtpItem.getChannelId(),
-								sendRtpItem.getPlatformId(), platform.getName(), userSetting.getServerId(), sendRtpItem.getMediaServerId());
-						messageForPushChannel.setPlatFormIndex(platform.getId());
-						redisCatchStorage.sendPlatformStopPlayMsg(messageForPushChannel);
-					}else {
-						logger.info("[涓婄骇骞冲彴鍋滄瑙傜湅] 鏈壘鍒板钩鍙皗}鐨勪俊鎭紝鍙戦�乺edis娑堟伅澶辫触", sendRtpItem.getPlatformId());
-					}
-				}
 
-				int totalReaderCount = zlmrtpServerFactory.totalReaderCount(mediaInfo, sendRtpItem.getApp(), streamId);
-				if (totalReaderCount <= 0) {
-					logger.info("[鏀跺埌bye] {} 鏃犲叾瀹冭鐪嬭�咃紝閫氱煡璁惧鍋滄鎺ㄦ祦", streamId);
-					if (sendRtpItem.getPlayType().equals(InviteStreamType.PLAY)) {
-						Device device = deviceService.getDevice(sendRtpItem.getDeviceId());
-						if (device == null) {
-							logger.info("[鏀跺埌bye] {} 閫氱煡璁惧鍋滄鎺ㄦ祦鏃舵湭鎵惧埌璁惧淇℃伅", streamId);
-						}
-						try {
-							logger.warn("[鍋滄鐐规挱] {}/{}", sendRtpItem.getDeviceId(), channelId);
-							cmder.streamByeCmd(device, channelId, streamId, null);
-						} catch (InvalidArgumentException | ParseException | SipException |
-								 SsrcTransactionNotFoundException e) {
-							logger.error("[鏀跺埌bye] {} 鏃犲叾瀹冭鐪嬭�咃紝閫氱煡璁惧鍋滄鎺ㄦ祦锛� 鍙戦�丅YE澶辫触 {}",streamId, e.getMessage());
-						}
-					}
-//					if (sendRtpItem.getPlayType().equals(InviteStreamType.PUSH)) {
-//						MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(0,
-//								sendRtpItem.getApp(), sendRtpItem.getStreamId(), sendRtpItem.getChannelId(),
-//								sendRtpItem.getPlatformId(), null, null, sendRtpItem.getMediaServerId());
-//						redisCatchStorage.sendStreamPushRequestedMsg(messageForPushChannel);
-//					}
-				}
-			}
-			// 鍙兘鏄澶囦富鍔ㄥ仠姝�
-			Device device = storager.queryVideoDeviceByChannelId(platformGbId);
-			if (device != null) {
-				storager.stopPlay(device.getDeviceId(), channelId);
-				StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(device.getDeviceId(), channelId);
-				if (streamInfo != null) {
-					redisCatchStorage.stopPlay(streamInfo);
-					mediaServerService.closeRTPServer(streamInfo.getMediaServerId(), streamInfo.getStream());
-				}
-				SsrcTransaction ssrcTransactionForPlay = streamSession.getSsrcTransaction(device.getDeviceId(), channelId, "play", null);
-				if (ssrcTransactionForPlay != null){
-					if (ssrcTransactionForPlay.getCallId().equals(callIdHeader.getCallId())){
-						// 閲婃斁ssrc
-						MediaServerItem mediaServerItem = mediaServerService.getOne(ssrcTransactionForPlay.getMediaServerId());
-						if (mediaServerItem != null) {
-							mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcTransactionForPlay.getSsrc());
-						}
-						streamSession.remove(device.getDeviceId(), channelId, ssrcTransactionForPlay.getStream());
-					}
-				}
-				SsrcTransaction ssrcTransactionForPlayBack = streamSession.getSsrcTransaction(device.getDeviceId(), channelId, callIdHeader.getCallId(), null);
-				if (ssrcTransactionForPlayBack != null) {
-					// 閲婃斁ssrc
-					MediaServerItem mediaServerItem = mediaServerService.getOne(ssrcTransactionForPlayBack.getMediaServerId());
-					if (mediaServerItem != null) {
-						mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcTransactionForPlayBack.getSsrc());
-					}
-					streamSession.remove(device.getDeviceId(), channelId, ssrcTransactionForPlayBack.getStream());
+		SendRtpItem sendRtpItem =  redisCatchStorage.querySendRTPServer(null, null, null, callIdHeader.getCallId());
+
+		if (sendRtpItem != null){
+			logger.info("[鏀跺埌bye] 鏉ヨ嚜骞冲彴{}锛� 鍋滄閫氶亾锛歿}", sendRtpItem.getPlatformId(), sendRtpItem.getChannelId());
+			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("[鏀跺埌bye] 鍋滄鍚戜笂绾ф帹娴侊細{}", streamId);
+			MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
+			redisCatchStorage.deleteSendRTPServer(sendRtpItem.getPlatformId(), sendRtpItem.getChannelId(),
+					callIdHeader.getCallId(), null);
+			zlmrtpServerFactory.stopSendRtpStream(mediaInfo, param);
+			if (sendRtpItem.getPlayType().equals(InviteStreamType.PUSH)) {
+				ParentPlatform platform = platformService.queryPlatformByServerGBId(sendRtpItem.getPlatformId());
+				if (platform != null) {
+					MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(0,
+							sendRtpItem.getApp(), sendRtpItem.getStreamId(), sendRtpItem.getChannelId(),
+							sendRtpItem.getPlatformId(), platform.getName(), userSetting.getServerId(), sendRtpItem.getMediaServerId());
+					messageForPushChannel.setPlatFormIndex(platform.getId());
+					redisCatchStorage.sendPlatformStopPlayMsg(messageForPushChannel);
+				}else {
+					logger.info("[涓婄骇骞冲彴鍋滄瑙傜湅] 鏈壘鍒板钩鍙皗}鐨勪俊鎭紝鍙戦�乺edis娑堟伅澶辫触", sendRtpItem.getPlatformId());
 				}
 			}
 
+			int totalReaderCount = zlmrtpServerFactory.totalReaderCount(mediaInfo, sendRtpItem.getApp(), streamId);
+			if (totalReaderCount <= 0) {
+				logger.info("[鏀跺埌bye] {} 鏃犲叾瀹冭鐪嬭�咃紝閫氱煡璁惧鍋滄鎺ㄦ祦", streamId);
+				if (sendRtpItem.getPlayType().equals(InviteStreamType.PLAY)) {
+					Device device = deviceService.getDevice(sendRtpItem.getDeviceId());
+					if (device == null) {
+						logger.info("[鏀跺埌bye] {} 閫氱煡璁惧鍋滄鎺ㄦ祦鏃舵湭鎵惧埌璁惧淇℃伅", streamId);
+					}
+					try {
+						logger.warn("[鍋滄鐐规挱] {}/{}", sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
+						cmder.streamByeCmd(device, sendRtpItem.getChannelId(), streamId, null);
+					} catch (InvalidArgumentException | ParseException | SipException |
+							 SsrcTransactionNotFoundException e) {
+						logger.error("[鏀跺埌bye] {} 鏃犲叾瀹冭鐪嬭�咃紝閫氱煡璁惧鍋滄鎺ㄦ祦锛� 鍙戦�丅YE澶辫触 {}",streamId, e.getMessage());
+					}
+				}
+			}
+		}else {
+			// 鍙兘鏄澶囧彂閫佺殑鍋滄
+			SsrcTransaction ssrcTransaction = streamSession.getSsrcTransaction(null, null, callIdHeader.getCallId(), null);
+			if (ssrcTransaction == null) {
+				logger.info("[鏀跺埌bye] 浣嗘槸鏃犳硶鑾峰彇鎺ㄦ祦淇℃伅鍜屽彂娴佷俊鎭紝蹇界暐姝よ姹�");
+				logger.info(request.toString());
+				return;
+			}
+			logger.info("[鏀跺埌bye] 鏉ヨ嚜璁惧锛歿}, 閫氶亾宸插仠姝㈡帹娴�: {}", ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId());
+
+			Device device = deviceService.getDevice(ssrcTransaction.getDeviceId());
+			if (device == null) {
+				logger.info("[鏀跺埌bye] 鏈壘鍒拌澶囷細{} ", ssrcTransaction.getDeviceId());
+				return;
+			}
+			DeviceChannel channel = channelService.getOne(ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId());
+			if (channel == null) {
+				logger.info("[鏀跺埌bye] 鏈壘鍒伴�氶亾锛岃澶囷細{}锛� 閫氶亾锛歿}", ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId());
+				return;
+			}
+			storager.stopPlay(device.getDeviceId(), channel.getChannelId());
+			StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(device.getDeviceId(), channel.getChannelId());
+			if (streamInfo != null) {
+				redisCatchStorage.stopPlay(streamInfo);
+				mediaServerService.closeRTPServer(streamInfo.getMediaServerId(), streamInfo.getStream());
+			}
+			// 閲婃斁ssrc
+			MediaServerItem mediaServerItem = mediaServerService.getOne(ssrcTransaction.getMediaServerId());
+			if (mediaServerItem != null) {
+				mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcTransaction.getSsrc());
+			}
+			streamSession.remove(device.getDeviceId(), channel.getChannelId(), ssrcTransaction.getStream());
+		}
 	}
 }
diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
index 2fbe6bb..03e00c2 100644
--- a/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
@@ -143,7 +143,7 @@
                 if (rtpInfo.getBoolean("exist")) {
                     int localPort = rtpInfo.getInteger("local_port");
                     if (localPort == 0) {
-                        logger.warn("[鐐规挱]锛岀偣鎾椂鍙戠幇rtpServerC瀛樺湪锛屼絾鏄皻鏈紑濮嬫帹娴�");
+                        logger.warn("[鐐规挱]锛岀偣鎾椂鍙戠幇rtpServer瀛樺湪锛屼絾鏄皻鏈紑濮嬫帹娴�");
                         // 姝ゆ椂璇存槑rtpServer宸茬粡鍒涘缓浣嗘槸娴佽繕娌℃湁鎺ㄤ笂鏉�
                         WVPResult wvpResult = new WVPResult();
                         wvpResult.setCode(ErrorCode.ERROR100.getCode());
diff --git a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamCloseResponseListener.java b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamCloseResponseListener.java
index 4e73578..f78b692 100644
--- a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamCloseResponseListener.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamCloseResponseListener.java
@@ -1,20 +1,34 @@
 package com.genersoft.iot.vmp.service.redisMsg;
 
 import com.alibaba.fastjson2.JSON;
+import com.genersoft.iot.vmp.conf.UserSetting;
+import com.genersoft.iot.vmp.gb28181.bean.InviteStreamType;
+import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
+import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
+import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
+import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
+import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
+import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
+import com.genersoft.iot.vmp.service.IMediaServerService;
+import com.genersoft.iot.vmp.service.IStreamPushService;
+import com.genersoft.iot.vmp.service.bean.MessageForPushChannel;
 import com.genersoft.iot.vmp.service.bean.MessageForPushChannelResponse;
+import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
+import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
 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.data.redis.connection.Message;
 import org.springframework.data.redis.connection.MessageListener;
-import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.stereotype.Component;
-import org.springframework.util.ObjectUtils;
 
+import javax.sip.InvalidArgumentException;
+import javax.sip.SipException;
+import java.text.ParseException;
+import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentLinkedQueue;
 
 /**
  * 鎺ユ敹redis鍙戦�佺殑缁撴潫鎺ㄦ祦璇锋眰
@@ -25,11 +39,26 @@
 
     private final static Logger logger = LoggerFactory.getLogger(RedisPushStreamCloseResponseListener.class);
 
-    private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
-
-    @Qualifier("taskExecutor")
     @Autowired
-    private ThreadPoolTaskExecutor taskExecutor;
+    private IStreamPushService streamPushService;
+
+    @Autowired
+    private IRedisCatchStorage redisCatchStorage;
+
+    @Autowired
+    private IVideoManagerStorage storager;
+
+    @Autowired
+    private ISIPCommanderForPlatform commanderFroPlatform;
+
+    @Autowired
+    private UserSetting userSetting;
+
+    @Autowired
+    private IMediaServerService mediaServerService;
+
+    @Autowired
+    private ZLMRTPServerFactory zlmrtpServerFactory;
 
 
     private Map<String, PushStreamResponseEvent> responseEvents = new ConcurrentHashMap<>();
@@ -40,30 +69,45 @@
 
     @Override
     public void onMessage(Message message, byte[] bytes) {
-        logger.info("[REDIS娑堟伅-璇锋眰鎺ㄦ祦缁撴灉]锛� {}", new String(message.getBody()));
-        boolean isEmpty = taskQueue.isEmpty();
-        taskQueue.offer(message);
-        if (isEmpty) {
-            taskExecutor.execute(() -> {
-                while (!taskQueue.isEmpty()) {
-                    Message msg = taskQueue.poll();
-                    try {
-                        MessageForPushChannelResponse response = JSON.parseObject(new String(msg.getBody()), MessageForPushChannelResponse.class);
-                        if (response == null || ObjectUtils.isEmpty(response.getApp()) || ObjectUtils.isEmpty(response.getStream())){
-                            logger.info("[REDIS娑堟伅-璇锋眰鎺ㄦ祦缁撴灉]锛氬弬鏁颁笉鍏�");
-                            continue;
+        logger.info("[REDIS娑堟伅-鎺ㄦ祦缁撴潫]锛� {}", new String(message.getBody()));
+        MessageForPushChannel pushChannel = JSON.parseObject(message.getBody(), MessageForPushChannel.class);
+        StreamPushItem push = streamPushService.getPush(pushChannel.getApp(), pushChannel.getStream());
+        if (push != null) {
+            if (redisCatchStorage.isChannelSendingRTP(push.getGbId())) {
+                List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByChnnelId(
+                        push.getGbId());
+                if (sendRtpItems.size() > 0) {
+                    for (SendRtpItem sendRtpItem : sendRtpItems) {
+                        ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
+                        // 鍋滄鍚戜笂绾ф帹娴�
+                        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("[REDIS娑堟伅-鎺ㄦ祦缁撴潫] 鍋滄鍚戜笂绾ф帹娴侊細{}", streamId);
+                        MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
+                        redisCatchStorage.deleteSendRTPServer(sendRtpItem.getPlatformId(), sendRtpItem.getChannelId(), sendRtpItem.getCallId(), sendRtpItem.getStreamId());
+                        zlmrtpServerFactory.stopSendRtpStream(mediaInfo, param);
+
+                        try {
+                            commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem);
+                        } catch (SipException | InvalidArgumentException | ParseException e) {
+                            logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 鍙戦�丅YE: {}", e.getMessage());
                         }
-                        // 鏌ョ湅姝e湪绛夊緟鐨刬nvite娑堟伅
-                        if (responseEvents.get(response.getApp() + response.getStream()) != null) {
-                            responseEvents.get(response.getApp() + response.getStream()).run(response);
+                        if (InviteStreamType.PUSH == sendRtpItem.getPlayType()) {
+                            MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(0,
+                                    sendRtpItem.getApp(), sendRtpItem.getStreamId(), sendRtpItem.getChannelId(),
+                                    sendRtpItem.getPlatformId(), parentPlatform.getName(), userSetting.getServerId(), sendRtpItem.getMediaServerId());
+                            messageForPushChannel.setPlatFormIndex(parentPlatform.getId());
+                            redisCatchStorage.sendPlatformStopPlayMsg(messageForPushChannel);
                         }
-                    }catch (Exception e) {
-                        logger.warn("[REDIS娑堟伅-璇锋眰鎺ㄦ祦缁撴灉] 鍙戠幇鏈鐞嗙殑寮傚父, \r\n{}", JSON.toJSONString(message));
-                        logger.error("[REDIS娑堟伅-璇锋眰鎺ㄦ祦缁撴灉] 寮傚父鍐呭锛� ", e);
                     }
                 }
-            });
+            }
         }
+
     }
 
     public void addEvent(String app, String stream, PushStreamResponseEvent callback) {

--
Gitblit v1.8.0