Shikong/wvp-gb28181-project
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

增加流关闭时的处理

Browse Source
This commit is contained in:
648540858 2023-06-28 20:32:48 +08:00
parent def56793ba
commit bfae9780f7
6 changed files with 163 additions and 123 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
src/main/java/com/genersoft/iot/vmp/conf/redis/RedisMsgListenConfig.java
View File

@ -43,6 +43,9 @@ public class RedisMsgListenConfig {
@Autowired @Autowired
private RedisPushStreamResponseListener redisPushStreamResponseListener; private RedisPushStreamResponseListener redisPushStreamResponseListener;
@Autowired
private RedisPushStreamCloseResponseListener redisPushStreamCloseResponseListener;
/** /**
* redis消息监听器容器 可以添加多个监听不同话题的redis监听器，只需要把消息监听器和相应的消息订阅处理器绑定，该消息监听器 * redis消息监听器容器 可以添加多个监听不同话题的redis监听器，只需要把消息监听器和相应的消息订阅处理器绑定，该消息监听器
@ -63,7 +66,7 @@ public class RedisMsgListenConfig {
container.addMessageListener(redisPushStreamStatusMsgListener, new PatternTopic(VideoManagerConstants.VM_MSG_PUSH_STREAM_STATUS_CHANGE)); 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(redisPushStreamListMsgListener, new PatternTopic(VideoManagerConstants.VM_MSG_PUSH_STREAM_LIST_CHANGE));
container.addMessageListener(redisPushStreamResponseListener, new PatternTopic(VideoManagerConstants.VM_MSG_STREAM_PUSH_RESPONSE)); 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; return container;
} }
} }

8
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderPlarformProvider.java
View File

@ -283,7 +283,7 @@ public class SIPRequestHeaderPlarformProvider {
viaHeader.setRPort(); viaHeader.setRPort();
viaHeaders.add(viaHeader); viaHeaders.add(viaHeader);
// from // from
SipURI fromSipURI = SipFactory.getInstance().createAddressFactory().createSipURI(platform.getDeviceGBId(), SipURI fromSipURI = SipFactory.getInstance().createAddressFactory().createSipURI(sendRtpItem.getChannelId(),
platform.getDeviceIp() + ":" + platform.getDevicePort()); platform.getDeviceIp() + ":" + platform.getDevicePort());
Address fromAddress = SipFactory.getInstance().createAddressFactory().createAddress(fromSipURI); Address fromAddress = SipFactory.getInstance().createAddressFactory().createAddress(fromSipURI);
FromHeader fromHeader = SipFactory.getInstance().createHeaderFactory().createFromHeader(fromAddress, sendRtpItem.getToTag()); FromHeader fromHeader = SipFactory.getInstance().createHeaderFactory().createFromHeader(fromAddress, sendRtpItem.getToTag());
@ -296,13 +296,10 @@ public class SIPRequestHeaderPlarformProvider {
MaxForwardsHeader maxForwards = SipFactory.getInstance().createHeaderFactory().createMaxForwardsHeader(70); MaxForwardsHeader maxForwards = SipFactory.getInstance().createHeaderFactory().createMaxForwardsHeader(70);
// ceq // ceq
CSeqHeader cSeqHeader = SipFactory.getInstance().createHeaderFactory().createCSeqHeader(redisCatchStorage.getCSEQ(), Request.BYE); 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()); 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); toHeader, viaHeaders, maxForwards);
request.addHeader(SipUtils.createUserAgentHeader(gitUtil)); request.addHeader(SipUtils.createUserAgentHeader(gitUtil));
@ -310,6 +307,7 @@ public class SIPRequestHeaderPlarformProvider {
String sipAddress = platform.getDeviceIp() + ":" + platform.getDevicePort(); String sipAddress = platform.getDeviceIp() + ":" + platform.getDevicePort();
Address concatAddress = SipFactory.getInstance().createAddressFactory().createAddress(SipFactory.getInstance().createAddressFactory() Address concatAddress = SipFactory.getInstance().createAddressFactory().createAddress(SipFactory.getInstance().createAddressFactory()
.createSipURI(platform.getDeviceGBId(), sipAddress)); .createSipURI(platform.getDeviceGBId(), sipAddress));
request.addHeader(SipFactory.getInstance().createHeaderFactory().createContactHeader(concatAddress)); request.addHeader(SipFactory.getInstance().createHeaderFactory().createContactHeader(concatAddress));
return request; return request;

7
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
View File

@ -347,10 +347,9 @@ public class SIPCommander implements ISIPCommander {
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
errorEvent.response(e); errorEvent.response(e);
}), e -> { }), e -> {
// 这里为例避免一个通道的点播只有一个callID这个参数使用一个固定值
ResponseEvent responseEvent = (ResponseEvent) e.event; ResponseEvent responseEvent = (ResponseEvent) e.event;
SIPResponse response = (SIPResponse) responseEvent.getResponse(); 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); okEvent.response(e);
}); });
} }
@ -452,7 +451,7 @@ public class SIPCommander implements ISIPCommander {
sipSender.transmitRequest(sipLayer.getLocalIp(device.getLocalIp()), request, errorEvent, event -> { sipSender.transmitRequest(sipLayer.getLocalIp(device.getLocalIp()), request, errorEvent, event -> {
ResponseEvent responseEvent = (ResponseEvent) event.event; ResponseEvent responseEvent = (ResponseEvent) event.event;
SIPResponse response = (SIPResponse) responseEvent.getResponse(); 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); okEvent.response(event);
}); });
if (inviteStreamCallback != null) { if (inviteStreamCallback != null) {
@ -580,7 +579,7 @@ public class SIPCommander implements ISIPCommander {
if (ssrcIndex >= 0) { if (ssrcIndex >= 0) {
ssrc = contentString.substring(ssrcIndex + 2, ssrcIndex + 12); 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); okEvent.response(event);
}); });
} }

166
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java
View File

@ -11,6 +11,7 @@ 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.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory; 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.MediaServerItem;
import com.genersoft.iot.vmp.service.IDeviceChannelService;
import com.genersoft.iot.vmp.service.IDeviceService; import com.genersoft.iot.vmp.service.IDeviceService;
import com.genersoft.iot.vmp.service.IMediaServerService; import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IPlatformService; import com.genersoft.iot.vmp.service.IPlatformService;
@ -27,11 +28,7 @@ import org.springframework.stereotype.Component;
import javax.sip.InvalidArgumentException; import javax.sip.InvalidArgumentException;
import javax.sip.RequestEvent; import javax.sip.RequestEvent;
import javax.sip.SipException; import javax.sip.SipException;
import javax.sip.address.SipURI;
import javax.sip.header.CallIdHeader; 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 javax.sip.message.Response;
import java.text.ParseException; import java.text.ParseException;
import java.util.HashMap; import java.util.HashMap;
@ -58,6 +55,9 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
@Autowired @Autowired
private IDeviceService deviceService; private IDeviceService deviceService;
@Autowired
private IDeviceChannelService channelService;
@Autowired @Autowired
private IVideoManagerStorage storager; private IVideoManagerStorage storager;
@ -88,95 +88,91 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
*/ */
@Override @Override
public void process(RequestEvent evt) { public void process(RequestEvent evt) {
SIPRequest request = (SIPRequest) evt.getRequest();
try { try {
responseAck((SIPRequest) evt.getRequest(), Response.OK); responseAck(request, Response.OK);
} catch (SipException | InvalidArgumentException | ParseException e) { } catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[回复BYE信息失败]，{}", e.getMessage()); logger.error("[回复BYE信息失败]，{}", e.getMessage());
} }
CallIdHeader callIdHeader = (CallIdHeader)evt.getRequest().getHeader(CallIdHeader.NAME); 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("[上级平台停止观看] 未找到平台{}的信息，发送redis消息失败", sendRtpItem.getPlatformId());
}
}
int totalReaderCount = zlmrtpServerFactory.totalReaderCount(mediaInfo, sendRtpItem.getApp(), streamId); SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(null, null, null, callIdHeader.getCallId());
if (totalReaderCount <= 0) {
logger.info("[收到bye] {} 无其它观看者，通知设备停止推流", streamId); if (sendRtpItem != null){
if (sendRtpItem.getPlayType().equals(InviteStreamType.PLAY)) { logger.info("[收到bye] 来自平台{}， 停止通道：{}", sendRtpItem.getPlatformId(), sendRtpItem.getChannelId());
Device device = deviceService.getDevice(sendRtpItem.getDeviceId()); String streamId = sendRtpItem.getStreamId();
if (device == null) { Map<String, Object> param = new HashMap<>();
logger.info("[收到bye] {} 通知设备停止推流时未找到设备信息", streamId); param.put("vhost","__defaultVhost__");
} param.put("app",sendRtpItem.getApp());
try { param.put("stream",streamId);
logger.warn("[停止点播] {}/{}", sendRtpItem.getDeviceId(), channelId); param.put("ssrc",sendRtpItem.getSsrc());
cmder.streamByeCmd(device, channelId, streamId, null); logger.info("[收到bye] 停止向上级推流：{}", streamId);
} catch (InvalidArgumentException | ParseException | SipException | MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
SsrcTransactionNotFoundException e) { redisCatchStorage.deleteSendRTPServer(sendRtpItem.getPlatformId(), sendRtpItem.getChannelId(),
logger.error("[收到bye] {} 无其它观看者，通知设备停止推流， 发送BYE失败 {}",streamId, e.getMessage()); callIdHeader.getCallId(), null);
} zlmrtpServerFactory.stopSendRtpStream(mediaInfo, param);
} if (sendRtpItem.getPlayType().equals(InviteStreamType.PUSH)) {
// if (sendRtpItem.getPlayType().equals(InviteStreamType.PUSH)) { ParentPlatform platform = platformService.queryPlatformByServerGBId(sendRtpItem.getPlatformId());
// MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(0, if (platform != null) {
// sendRtpItem.getApp(), sendRtpItem.getStreamId(), sendRtpItem.getChannelId(), MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(0,
// sendRtpItem.getPlatformId(), null, null, sendRtpItem.getMediaServerId()); sendRtpItem.getApp(), sendRtpItem.getStreamId(), sendRtpItem.getChannelId(),
// redisCatchStorage.sendStreamPushRequestedMsg(messageForPushChannel); sendRtpItem.getPlatformId(), platform.getName(), userSetting.getServerId(), sendRtpItem.getMediaServerId());
// } messageForPushChannel.setPlatFormIndex(platform.getId());
} redisCatchStorage.sendPlatformStopPlayMsg(messageForPushChannel);
} }else {
// 可能是设备主动停止 logger.info("[上级平台停止观看] 未找到平台{}的信息，发送redis消息失败", sendRtpItem.getPlatformId());
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());
} }
} }
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] {} 无其它观看者，通知设备停止推流， 发送BYE失败 {}",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());
}
} }
} }

2
src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
View File

@ -143,7 +143,7 @@ public class PlayServiceImpl implements IPlayService {
if (rtpInfo.getBoolean("exist")) { if (rtpInfo.getBoolean("exist")) {
int localPort = rtpInfo.getInteger("local_port"); int localPort = rtpInfo.getInteger("local_port");
if (localPort == 0) { if (localPort == 0) {
logger.warn("[点播]，点播时发现rtpServerC存在，但是尚未开始推流"); logger.warn("[点播]，点播时发现rtpServer存在，但是尚未开始推流");
// 此时说明rtpServer已经创建但是流还没有推上来 // 此时说明rtpServer已经创建但是流还没有推上来
WVPResult wvpResult = new WVPResult(); WVPResult wvpResult = new WVPResult();
wvpResult.setCode(ErrorCode.ERROR100.getCode()); wvpResult.setCode(ErrorCode.ERROR100.getCode());

98
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamCloseResponseListener.java
View File

@ -1,20 +1,34 @@
package com.genersoft.iot.vmp.service.redisMsg; package com.genersoft.iot.vmp.service.redisMsg;
import com.alibaba.fastjson2.JSON; 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.service.bean.MessageForPushChannelResponse;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired; 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.Message;
import org.springframework.data.redis.connection.MessageListener; import org.springframework.data.redis.connection.MessageListener;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component; 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.Map;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
/** /**
* 接收redis发送的结束推流请求 * 接收redis发送的结束推流请求
@ -25,11 +39,26 @@ public class RedisPushStreamCloseResponseListener implements MessageListener {
private final static Logger logger = LoggerFactory.getLogger(RedisPushStreamCloseResponseListener.class); private final static Logger logger = LoggerFactory.getLogger(RedisPushStreamCloseResponseListener.class);
private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
@Qualifier("taskExecutor")
@Autowired @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<>(); private Map<String, PushStreamResponseEvent> responseEvents = new ConcurrentHashMap<>();
@ -40,30 +69,45 @@ public class RedisPushStreamCloseResponseListener implements MessageListener {
@Override @Override
public void onMessage(Message message, byte[] bytes) { public void onMessage(Message message, byte[] bytes) {
logger.info("[REDIS消息-请求推流结果]： {}", new String(message.getBody())); logger.info("[REDIS消息-推流结束]： {}", new String(message.getBody()));
boolean isEmpty = taskQueue.isEmpty(); MessageForPushChannel pushChannel = JSON.parseObject(message.getBody(), MessageForPushChannel.class);
taskQueue.offer(message); StreamPushItem push = streamPushService.getPush(pushChannel.getApp(), pushChannel.getStream());
if (isEmpty) { if (push != null) {
taskExecutor.execute(() -> { if (redisCatchStorage.isChannelSendingRTP(push.getGbId())) {
while (!taskQueue.isEmpty()) { List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByChnnelId(
Message msg = taskQueue.poll(); push.getGbId());
try { if (sendRtpItems.size() > 0) {
MessageForPushChannelResponse response = JSON.parseObject(new String(msg.getBody()), MessageForPushChannelResponse.class); for (SendRtpItem sendRtpItem : sendRtpItems) {
if (response == null || ObjectUtils.isEmpty(response.getApp()) || ObjectUtils.isEmpty(response.getStream())){ ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
logger.info("[REDIS消息-请求推流结果]：参数不全"); // 停止向上级推流
continue; 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("[命令发送失败] 国标级联 发送BYE: {}", e.getMessage());
} }
// 查看正在等待的invite消息 if (InviteStreamType.PUSH == sendRtpItem.getPlayType()) {
if (responseEvents.get(response.getApp() + response.getStream()) != null) { MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(0,
responseEvents.get(response.getApp() + response.getStream()).run(response); 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) { public void addEvent(String app, String stream, PushStreamResponseEvent callback) {