diff --git a/src/main/java/com/genersoft/iot/vmp/common/InviteInfo.java b/src/main/java/com/genersoft/iot/vmp/common/InviteInfo.java new file mode 100644 index 00000000..9fe43f74 --- /dev/null +++ b/src/main/java/com/genersoft/iot/vmp/common/InviteInfo.java @@ -0,0 +1,126 @@ +package com.genersoft.iot.vmp.common; + +import com.genersoft.iot.vmp.service.bean.SSRCInfo; + +/** + * 记录每次发送invite消息的状态 + */ +public class InviteInfo { + + private String deviceId; + + private String channelId; + + private String stream; + + private SSRCInfo ssrcInfo; + + private String receiveIp; + + private Integer receivePort; + + private String streamMode; + + private InviteSessionType type; + + private InviteSessionStatus status; + + private StreamInfo streamInfo; + + + public static InviteInfo getinviteInfo(String deviceId, String channelId, String stream, SSRCInfo ssrcInfo, + String receiveIp, Integer receivePort, String streamMode, + InviteSessionType type, InviteSessionStatus status) { + InviteInfo inviteInfo = new InviteInfo(); + inviteInfo.setDeviceId(deviceId); + inviteInfo.setChannelId(channelId); + inviteInfo.setStream(stream); + inviteInfo.setSsrcInfo(ssrcInfo); + inviteInfo.setReceiveIp(receiveIp); + inviteInfo.setReceivePort(receivePort); + inviteInfo.setStreamMode(streamMode); + inviteInfo.setType(type); + inviteInfo.setStatus(status); + return inviteInfo; + } + + public String getDeviceId() { + return deviceId; + } + + public void setDeviceId(String deviceId) { + this.deviceId = deviceId; + } + + public String getChannelId() { + return channelId; + } + + public void setChannelId(String channelId) { + this.channelId = channelId; + } + + public InviteSessionType getType() { + return type; + } + + public void setType(InviteSessionType type) { + this.type = type; + } + + public InviteSessionStatus getStatus() { + return status; + } + + public void setStatus(InviteSessionStatus status) { + this.status = status; + } + + public StreamInfo getStreamInfo() { + return streamInfo; + } + + public void setStreamInfo(StreamInfo streamInfo) { + this.streamInfo = streamInfo; + } + + public String getStream() { + return stream; + } + + public void setStream(String stream) { + this.stream = stream; + } + + public SSRCInfo getSsrcInfo() { + return ssrcInfo; + } + + public void setSsrcInfo(SSRCInfo ssrcInfo) { + this.ssrcInfo = ssrcInfo; + } + + public String getReceiveIp() { + return receiveIp; + } + + public void setReceiveIp(String receiveIp) { + this.receiveIp = receiveIp; + } + + public Integer getReceivePort() { + return receivePort; + } + + public void setReceivePort(Integer receivePort) { + this.receivePort = receivePort; + } + + public String getStreamMode() { + return streamMode; + } + + public void setStreamMode(String streamMode) { + this.streamMode = streamMode; + } +} diff --git a/src/main/java/com/genersoft/iot/vmp/common/InviteSessionStatus.java b/src/main/java/com/genersoft/iot/vmp/common/InviteSessionStatus.java new file mode 100644 index 00000000..04cc7c91 --- /dev/null +++ b/src/main/java/com/genersoft/iot/vmp/common/InviteSessionStatus.java @@ -0,0 +1,11 @@ +package com.genersoft.iot.vmp.common; + +/** + * 标识invite消息发出后的各个状态， + * 收到ok钱停止invite发送cancel， + * 收到200ok后发送BYE停止invite + */ +public enum InviteSessionStatus { + ready, + ok, +} diff --git a/src/main/java/com/genersoft/iot/vmp/common/InviteSessionType.java b/src/main/java/com/genersoft/iot/vmp/common/InviteSessionType.java new file mode 100644 index 00000000..5a6eb85b --- /dev/null +++ b/src/main/java/com/genersoft/iot/vmp/common/InviteSessionType.java @@ -0,0 +1,7 @@ +package com.genersoft.iot.vmp.common; + +public enum InviteSessionType { + PLAY, + PLAYBACK, + DOWNLOAD +} diff --git a/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java b/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java index ccfe77ec..51c2ab10 100644 --- a/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java +++ b/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java @@ -16,8 +16,6 @@ public class VideoManagerConstants { public static final String MEDIA_SERVERS_ONLINE_PREFIX = "VMP_MEDIA_ONLINE_SERVERS_"; - public static final String MEDIA_STREAM_PREFIX = "VMP_MEDIA_STREAM"; - public static final String DEVICE_PREFIX = "VMP_DEVICE_"; // 设备同步完成 @@ -28,9 +26,10 @@ public class VideoManagerConstants { public static final String KEEPLIVEKEY_PREFIX = "VMP_KEEPALIVE_"; // TODO 此处多了一个_，暂不修改 - public static final String PLAYER_PREFIX = "VMP_PLAYER_"; - public static final String PLAY_BLACK_PREFIX = "VMP_PLAYBACK_"; - public static final String DOWNLOAD_PREFIX = "VMP_DOWNLOAD_"; + public static final String INVITE_PREFIX = "VMP_INVITE"; + public static final String PLAYER_PREFIX = "VMP_INVITE_PLAY_"; + public static final String PLAY_BLACK_PREFIX = "VMP_INVITE_PLAYBACK_"; + public static final String DOWNLOAD_PREFIX = "VMP_INVITE_DOWNLOAD_"; public static final String PLATFORM_KEEPALIVE_PREFIX = "VMP_PLATFORM_KEEPALIVE_"; diff --git a/src/main/java/com/genersoft/iot/vmp/conf/ProxyServletConfig.java b/src/main/java/com/genersoft/iot/vmp/conf/ProxyServletConfig.java index 6cc3b415..569b5e19 100644 --- a/src/main/java/com/genersoft/iot/vmp/conf/ProxyServletConfig.java +++ b/src/main/java/com/genersoft/iot/vmp/conf/ProxyServletConfig.java @@ -2,7 +2,6 @@ package com.genersoft.iot.vmp.conf; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; import com.genersoft.iot.vmp.service.IMediaServerService; -import org.apache.catalina.connector.ClientAbortException; import org.apache.http.HttpHost; import org.apache.http.HttpRequest; import org.apache.http.HttpResponse; @@ -194,11 +193,11 @@ public class ProxyServletConfig { } catch (IOException ioException) { if (ioException instanceof ConnectException) { logger.error("录像服务 连接失败"); - }else if (ioException instanceof ClientAbortException) { - /** - * TODO 使用这个代理库实现代理在遇到代理视频文件时，如果是206结果，会遇到报错蛋市目前功能正常， - * TODO 暂时去除异常处理。后续使用其他代理框架修改测试 - */ +// }else if (ioException instanceof ClientAbortException) { +// /** +// * TODO 使用这个代理库实现代理在遇到代理视频文件时，如果是206结果，会遇到报错蛋市目前功能正常， +// * TODO 暂时去除异常处理。后续使用其他代理框架修改测试 +// */ }else { logger.error("录像服务 代理失败： ", e); diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SsrcTransaction.java b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SsrcTransaction.java index d27ce262..6ed8d144 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SsrcTransaction.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SsrcTransaction.java @@ -1,6 +1,6 @@ package com.genersoft.iot.vmp.gb28181.bean; -import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager; +import com.genersoft.iot.vmp.common.InviteSessionType; public class SsrcTransaction { @@ -13,7 +13,7 @@ public class SsrcTransaction { private SipTransactionInfo sipTransactionInfo; - private VideoStreamSessionManager.SessionType type; + private InviteSessionType type; public String getDeviceId() { return deviceId; @@ -63,11 +63,11 @@ public class SsrcTransaction { this.ssrc = ssrc; } - public VideoStreamSessionManager.SessionType getType() { + public InviteSessionType getType() { return type; } - public void setType(VideoStreamSessionManager.SessionType type) { + public void setType(InviteSessionType type) { this.type = type; } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/session/VideoStreamSessionManager.java b/src/main/java/com/genersoft/iot/vmp/gb28181/session/VideoStreamSessionManager.java index dabfdff4..a5da0186 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/session/VideoStreamSessionManager.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/session/VideoStreamSessionManager.java @@ -1,5 +1,6 @@ package com.genersoft.iot.vmp.gb28181.session; +import com.genersoft.iot.vmp.common.InviteSessionType; import com.genersoft.iot.vmp.common.VideoManagerConstants; import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.gb28181.bean.SipTransactionInfo; @@ -27,12 +28,6 @@ public class VideoStreamSessionManager { @Autowired private RedisTemplate redisTemplate; - public enum SessionType { - play, - playback, - download - } - /** * 添加一个点播/回放的事务信息 * 后续可以通过流Id/callID @@ -43,7 +38,7 @@ public class VideoStreamSessionManager { * @param mediaServerId 所使用的流媒体ID * @param response 回复 */ - public void put(String deviceId, String channelId, String callId, String stream, String ssrc, String mediaServerId, SIPResponse response, SessionType type){ + public void put(String deviceId, String channelId, String callId, String stream, String ssrc, String mediaServerId, SIPResponse response, InviteSessionType type){ SsrcTransaction ssrcTransaction = new SsrcTransaction(); ssrcTransaction.setDeviceId(deviceId); ssrcTransaction.setChannelId(channelId); 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 4f0dc11a..e26d6ed8 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 @@ -1,6 +1,7 @@ package com.genersoft.iot.vmp.gb28181.transmit.cmd.impl; import com.alibaba.fastjson2.JSONObject; +import com.genersoft.iot.vmp.common.InviteSessionType; import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.conf.SipConfig; import com.genersoft.iot.vmp.conf.UserSetting; @@ -350,7 +351,7 @@ public class SIPCommander implements ISIPCommander { // 这里为例避免一个通道的点播只有一个callID这个参数使用一个固定值 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, "play", stream, ssrcInfo.getSsrc(), mediaServerItem.getId(), response, InviteSessionType.PLAY); okEvent.response(e); }); } @@ -452,7 +453,7 @@ public class SIPCommander implements ISIPCommander { 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,sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()),device.getTransport()).getCallId(), ssrcInfo.getStream(), ssrcInfo.getSsrc(), mediaServerItem.getId(), response, InviteSessionType.PLAYBACK); okEvent.response(event); }); if (inviteStreamCallback != null) { @@ -580,7 +581,7 @@ public class SIPCommander implements ISIPCommander { 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, response.getCallIdHeader().getCallId(), ssrcInfo.getStream(), ssrc, mediaServerItem.getId(), response, InviteSessionType.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 cc3051fb..66a1ce05 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 @@ -1,6 +1,7 @@ package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl; -import com.genersoft.iot.vmp.common.StreamInfo; +import com.genersoft.iot.vmp.common.InviteInfo; +import com.genersoft.iot.vmp.common.InviteSessionType; import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException; import com.genersoft.iot.vmp.gb28181.bean.Device; import com.genersoft.iot.vmp.gb28181.bean.InviteStreamType; @@ -15,6 +16,7 @@ import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorP import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; import com.genersoft.iot.vmp.service.IDeviceService; +import com.genersoft.iot.vmp.service.IInviteStreamService; import com.genersoft.iot.vmp.service.IMediaServerService; import com.genersoft.iot.vmp.service.bean.MessageForPushChannel; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; @@ -26,7 +28,9 @@ import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; -import javax.sip.*; +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; @@ -52,6 +56,9 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In @Autowired private IRedisCatchStorage redisCatchStorage; + @Autowired + private IInviteStreamService inviteStreamService; + @Autowired private IDeviceService deviceService; @@ -136,11 +143,6 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In 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())){ @@ -151,6 +153,14 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In } streamSession.remove(device.getDeviceId(), channelId, ssrcTransactionForPlay.getStream()); } + InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channelId); + + if (inviteInfo != null) { + inviteStreamService.removeInviteInfo(inviteInfo); + if (inviteInfo.getStreamInfo() != null) { + mediaServerService.closeRTPServer(inviteInfo.getStreamInfo().getMediaServerId(), inviteInfo.getStream()); + } + } } SsrcTransaction ssrcTransactionForPlayBack = streamSession.getSsrcTransaction(device.getDeviceId(), channelId, callIdHeader.getCallId(), null); if (ssrcTransactionForPlayBack != null) { @@ -160,6 +170,14 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcTransactionForPlayBack.getSsrc()); } streamSession.remove(device.getDeviceId(), channelId, ssrcTransactionForPlayBack.getStream()); + InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAYBACK, device.getDeviceId(), channelId); + + if (inviteInfo != null) { + inviteStreamService.removeInviteInfo(inviteInfo); + if (inviteInfo.getStreamInfo() != null) { + mediaServerService.closeRTPServer(inviteInfo.getStreamInfo().getMediaServerId(), inviteInfo.getStream()); + } + } } } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java index 7034a0e5..2acf402c 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java @@ -1,12 +1,10 @@ package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl; -import com.alibaba.fastjson2.JSONObject; +import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.conf.DynamicTask; import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.gb28181.bean.*; -import com.genersoft.iot.vmp.gb28181.event.SipSubscribe; import com.genersoft.iot.vmp.gb28181.session.SSRCFactory; -import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager; import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver; import com.genersoft.iot.vmp.gb28181.transmit.SIPSender; import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform; @@ -21,6 +19,8 @@ import com.genersoft.iot.vmp.service.IMediaServerService; import com.genersoft.iot.vmp.service.IPlayService; import com.genersoft.iot.vmp.service.IStreamProxyService; import com.genersoft.iot.vmp.service.IStreamPushService; +import com.genersoft.iot.vmp.service.bean.InviteErrorCallback; +import com.genersoft.iot.vmp.service.bean.InviteErrorCode; import com.genersoft.iot.vmp.service.bean.MessageForPushChannel; import com.genersoft.iot.vmp.service.bean.SSRCInfo; import com.genersoft.iot.vmp.service.redisMsg.RedisGbPlayMsgListener; @@ -101,9 +101,6 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements @Autowired private SIPProcessorObserver sipProcessorObserver; - @Autowired - private VideoStreamSessionManager sessionManager; - @Autowired private UserSetting userSetting; @@ -380,10 +377,10 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements Long finalStartTime = startTime; Long finalStopTime = stopTime; - ZlmHttpHookSubscribe.Event hookEvent = (mediaServerItemInUSe, responseJSON) -> { - String app = responseJSON.getString("app"); - String stream = responseJSON.getString("stream"); - logger.info("[上级点播]下级已经开始推流。 回复200OK(SDP)， {}/{}", app, stream); + InviteErrorCallback hookEvent = (code, msg, data) -> { + StreamInfo streamInfo = (StreamInfo)data; + MediaServerItem mediaServerItemInUSe = mediaServerService.getOne(streamInfo.getMediaServerId()); + logger.info("[上级点播]下级已经开始推流。 回复200OK(SDP)， {}/{}", streamInfo.getApp(), streamInfo.getStream()); // * 0 等待设备推流上来 // * 1 下级已经推流，等待上级平台回复ack // * 2 推流中 @@ -429,10 +426,10 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements logger.error("[命令发送失败] 国标级联 回复SdpAck", e); } }; - SipSubscribe.Event errorEvent = ((event) -> { + InviteErrorCallback errorEvent = ((statusCode, msg, data) -> { // 未知错误。直接转发设备点播的错误 try { - Response response = getMessageFactory().createResponse(event.statusCode, evt.getRequest()); + Response response = getMessageFactory().createResponse(statusCode, evt.getRequest()); sipSender.transmitRequest(request.getLocalAddress().getHostAddress(), response); } catch (ParseException | SipException e) { logger.error("未处理的异常 ", e); @@ -450,7 +447,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements if (result.getCode() != 0) { logger.warn("录像回放失败"); if (result.getEvent() != null) { - errorEvent.response(result.getEvent()); +// errorEvent.response(result.getEvent()); } redisCatchStorage.deleteSendRTPServer(platform.getServerGBId(), channelId, callIdHeader.getCallId(), null); try { @@ -460,53 +457,31 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements } } else { if (result.getMediaServerItem() != null) { - hookEvent.response(result.getMediaServerItem(), result.getResponse()); +// hookEvent.response(result.getMediaServerItem(), result.getResponse()); } } }); } else { sendRtpItem.setPlayType(InviteStreamType.PLAY); - SsrcTransaction playTransaction = sessionManager.getSsrcTransaction(device.getDeviceId(), channelId, "play", null); - if (playTransaction != null) { - Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, "rtp", playTransaction.getStream()); - if (!streamReady) { - boolean hasRtpServer = mediaServerService.checkRtpServer(mediaServerItem, "rtp", playTransaction.getStream()); - if (hasRtpServer) { - logger.info("[上级点播]已经开启rtpServer但是尚未收到流，开启监听流的到来"); - HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", playTransaction.getStream(), true, "rtsp", mediaServerItem.getId()); - zlmHttpHookSubscribe.addSubscribe(hookSubscribe, hookEvent); - }else { - playTransaction = null; - } - } + String streamId = null; + if (mediaServerItem.isRtpEnable()) { + streamId = String.format("%s_%s", device.getDeviceId(), channelId); + }else { + streamId = String.format("%08x", Integer.parseInt(ssrc)).toUpperCase(); } - if (playTransaction == null) { - // 被点播的通道目前未被点播，则开始点播 - String streamId = null; - if (mediaServerItem.isRtpEnable()) { - streamId = String.format("%s_%s", device.getDeviceId(), channelId); - } - SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, ssrc, device.isSsrcCheck(), false, 0, false, device.getStreamModeForParam()); - logger.info(JSONObject.toJSONString(ssrcInfo)); - sendRtpItem.setStreamId(ssrcInfo.getStream()); -// sendRtpItem.setSsrc(ssrcInfo.getSsrc()); - - // 写入redis， 超时时回复 - redisCatchStorage.updateSendRTPSever(sendRtpItem); - playService.play(mediaServerItem, ssrcInfo, device, channelId, hookEvent, errorEvent, (code, msg) -> { + sendRtpItem.setStreamId(streamId); + redisCatchStorage.updateSendRTPSever(sendRtpItem); + playService.play(mediaServerItem, device.getDeviceId(), channelId, ((code, msg, data) -> { + if (code == InviteErrorCode.SUCCESS.getCode()){ + hookEvent.run(code, msg, data); + }else if (code == InviteErrorCode.ERROR_FOR_SIGNALLING_TIMEOUT.getCode() || code == InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getCode()){ logger.info("[上级点播]超时, 用户：{}， 通道：{}", username, channelId); redisCatchStorage.deleteSendRTPServer(platform.getServerGBId(), channelId, callIdHeader.getCallId(), null); - }); - } else { + }else { + errorEvent.run(code, msg, data); + } + })); - sendRtpItem.setStreamId(playTransaction.getStream()); - // 写入redis， 超时时回复 - redisCatchStorage.updateSendRTPSever(sendRtpItem); - JSONObject jsonObject = new JSONObject(); - jsonObject.put("app", sendRtpItem.getApp()); - jsonObject.put("stream", sendRtpItem.getStreamId()); - hookEvent.response(mediaServerItem, jsonObject); - } } } else if (gbStream != null) { @@ -559,7 +534,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements int port, Boolean tcpActive, boolean mediaTransmissionTCP, String channelId, String addressStr, String ssrc, String requesterId) { Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, gbStream.getApp(), gbStream.getStream()); - if (streamReady) { + if (streamReady != null && streamReady) { // 自平台内容 SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId, gbStream.getApp(), gbStream.getStream(), channelId, mediaTransmissionTCP, platform.isRtcp()); @@ -598,7 +573,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements // 推流 if (streamPushItem.isSelf()) { Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, gbStream.getApp(), gbStream.getStream()); - if (streamReady) { + if (streamReady != null && streamReady) { // 自平台内容 SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId, gbStream.getApp(), gbStream.getStream(), channelId, mediaTransmissionTCP, platform.isRtcp()); diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java index 4e9b57d2..16818356 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java @@ -2,6 +2,8 @@ package com.genersoft.iot.vmp.media.zlm; import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSONObject; +import com.genersoft.iot.vmp.common.InviteInfo; +import com.genersoft.iot.vmp.common.InviteSessionType; import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException; @@ -22,10 +24,8 @@ import com.genersoft.iot.vmp.media.zlm.dto.hook.*; import com.genersoft.iot.vmp.service.*; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage; -import com.genersoft.iot.vmp.vmanager.bean.DeferredResultEx; import com.genersoft.iot.vmp.vmanager.bean.ErrorCode; import com.genersoft.iot.vmp.vmanager.bean.StreamContent; -import com.genersoft.iot.vmp.vmanager.bean.WVPResult; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; @@ -70,6 +70,9 @@ public class ZLMHttpHookListener { @Autowired private IRedisCatchStorage redisCatchStorage; + @Autowired + private IInviteStreamService inviteStreamService; + @Autowired private IDeviceService deviceService; @@ -252,7 +255,7 @@ public class ZLMHttpHookListener { result.setEnable_audio(deviceChannel.isHasAudio()); } // 如果是录像下载就设置视频间隔十秒 - if (ssrcTransactionForAll.get(0).getType() == VideoStreamSessionManager.SessionType.download) { + if (ssrcTransactionForAll.get(0).getType() == InviteSessionType.DOWNLOAD) { result.setMp4_max_second(10); result.setEnable_audio(true); result.setEnable_mp4(true); @@ -342,17 +345,10 @@ public class ZLMHttpHookListener { } if ("rtp".equals(param.getApp()) && !param.isRegist()) { - StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(param.getStream()); - if (streamInfo != null) { - redisCatchStorage.stopPlay(streamInfo); - storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId()); - } else { - streamInfo = redisCatchStorage.queryPlayback(null, null, - param.getStream(), null); - if (streamInfo != null) { - redisCatchStorage.stopPlayback(streamInfo.getDeviceID(), streamInfo.getChannelId(), - streamInfo.getStream(), null); - } + InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(null, param.getStream()); + if (inviteInfo != null && (inviteInfo.getType() == InviteSessionType.PLAY || inviteInfo.getType() == InviteSessionType.PLAYBACK)) { + inviteStreamService.removeInviteInfo(inviteInfo); + storager.stopPlay(inviteInfo.getDeviceId(), inviteInfo.getChannelId()); } } else { if (!"rtp".equals(param.getApp())) { @@ -450,13 +446,15 @@ public class ZLMHttpHookListener { if ("rtp".equals(param.getApp())) { ret.put("close", userSetting.getStreamOnDemand()); // 国标流， 点播/录像回放/录像下载 - StreamInfo streamInfoForPlayCatch = redisCatchStorage.queryPlayByStreamId(param.getStream()); +// StreamInfo streamInfoForPlayCatch = redisCatchStorage.queryPlayByStreamId(param.getStream()); + + InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(null, param.getStream()); // 点播 - if (streamInfoForPlayCatch != null) { + if (inviteInfo != null) { // 收到无人观看说明流也没有在往上级推送 - if (redisCatchStorage.isChannelSendingRTP(streamInfoForPlayCatch.getChannelId())) { + if (redisCatchStorage.isChannelSendingRTP(inviteInfo.getChannelId())) { List sendRtpItems = redisCatchStorage.querySendRTPServerByChnnelId( - streamInfoForPlayCatch.getChannelId()); + inviteInfo.getChannelId()); if (sendRtpItems.size() > 0) { for (SendRtpItem sendRtpItem : sendRtpItems) { ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId()); @@ -470,19 +468,22 @@ public class ZLMHttpHookListener { } } } - Device device = deviceService.getDevice(streamInfoForPlayCatch.getDeviceID()); + Device device = deviceService.getDevice(inviteInfo.getDeviceId()); if (device != null) { try { - cmder.streamByeCmd(device, streamInfoForPlayCatch.getChannelId(), - streamInfoForPlayCatch.getStream(), null); + if (inviteStreamService.getInviteInfo(inviteInfo.getType(), inviteInfo.getDeviceId(), inviteInfo.getChannelId(), inviteInfo.getStream()) != null) { + cmder.streamByeCmd(device, inviteInfo.getChannelId(), + inviteInfo.getStream(), null); + } } catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) { logger.error("[无人观看]点播， 发送BYE失败 {}", e.getMessage()); } } - redisCatchStorage.stopPlay(streamInfoForPlayCatch); - storager.stopPlay(streamInfoForPlayCatch.getDeviceID(), streamInfoForPlayCatch.getChannelId()); + inviteStreamService.removeInviteInfo(inviteInfo.getType(), inviteInfo.getDeviceId(), + inviteInfo.getChannelId(), inviteInfo.getStream()); + storager.stopPlay(inviteInfo.getDeviceId(), inviteInfo.getChannelId()); return ret; } // 录像回放 @@ -582,6 +583,7 @@ public class ZLMHttpHookListener { return defaultResult; } logger.info("[ZLM HOOK] 流未找到, 发起自动点播：{}->{}->{}/{}", param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream()); + RequestMessage msg = new RequestMessage(); String key = DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId; boolean exist = resultHolder.exist(key, null); @@ -589,31 +591,22 @@ public class ZLMHttpHookListener { String uuid = UUID.randomUUID().toString(); msg.setId(uuid); DeferredResult result = new DeferredResult<>(userSetting.getPlayTimeout().longValue()); - DeferredResultEx deferredResultEx = new DeferredResultEx<>(result); result.onTimeout(() -> { - logger.info("点播接口等待超时"); + logger.info("[ZLM HOOK] 自动点播, 等待超时"); // 释放rtpserver msg.setData(new HookResult(ErrorCode.ERROR100.getCode(), "点播超时")); resultHolder.invokeResult(msg); }); - // TODO 在点播未成功的情况下在此调用接口点播会导致返回的流地址ip错误 - deferredResultEx.setFilter(result1 -> { - WVPResult wvpResult1 = (WVPResult) result1; - HookResult resultForEnd = new HookResult(); - resultForEnd.setCode(wvpResult1.getCode()); - resultForEnd.setMsg(wvpResult1.getMsg()); - return resultForEnd; - }); // 录像查询以channelId作为deviceId查询 - resultHolder.put(key, uuid, deferredResultEx); + resultHolder.put(key, uuid, result); if (!exist) { - playService.play(mediaInfo, deviceId, channelId, null, eventResult -> { - msg.setData(new HookResult(eventResult.statusCode, eventResult.msg)); + playService.play(mediaInfo, deviceId, channelId, (code, message, data) -> { + msg.setData(new HookResult(code, message)); resultHolder.invokeResult(msg); - }, null); + }); } return result; } else { diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMMediaListManager.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMMediaListManager.java index db2beb0c..8e9b3d0b 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMMediaListManager.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMMediaListManager.java @@ -97,7 +97,8 @@ public class ZLMMediaListManager { public void sendStreamEvent(String app, String stream, String mediaServerId) { MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId); // 查看推流状态 - if (zlmrtpServerFactory.isStreamReady(mediaServerItem, app, stream)) { + Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, app, stream); + if (streamReady != null && streamReady) { ChannelOnlineEvent channelOnlineEventLister = getChannelOnlineEventLister(app, stream); if (channelOnlineEventLister != null) { try { diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java index f3e4d440..423777f6 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java @@ -330,6 +330,9 @@ public class ZLMRTPServerFactory { */ public Boolean isRtpReady(MediaServerItem mediaServerItem, String streamId) { JSONObject mediaInfo = zlmresTfulUtils.getMediaInfo(mediaServerItem,"rtp", "rtsp", streamId); + if (mediaInfo.getInteger("code") == -2) { + return null; + } return (mediaInfo.getInteger("code") == 0 && mediaInfo.getBoolean("online")); } @@ -338,8 +341,10 @@ public class ZLMRTPServerFactory { */ public Boolean isStreamReady(MediaServerItem mediaServerItem, String app, String streamId) { JSONObject mediaInfo = zlmresTfulUtils.getMediaList(mediaServerItem, app, streamId); - return mediaInfo != null && (mediaInfo.getInteger("code") == 0 - + if (mediaInfo == null || (mediaInfo.getInteger("code") == -2)) { + return null; + } + return (mediaInfo.getInteger("code") == 0 && mediaInfo.getJSONArray("data") != null && mediaInfo.getJSONArray("data").size() > 0); } diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZlmHttpHookSubscribe.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZlmHttpHookSubscribe.java index cf33bb24..4d0069b1 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZlmHttpHookSubscribe.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZlmHttpHookSubscribe.java @@ -134,9 +134,10 @@ public class ZlmHttpHookSubscribe { /** * 对订阅数据进行过期清理 */ - @Scheduled(cron="0 0/5 * * * ?") //每5分钟执行一次 +// @Scheduled(cron="0 0/5 * * * ?") //每5分钟执行一次 + @Scheduled(fixedRate = 2 * 1000) public void execute(){ - + System.out.println(allSubscribes.size()); Instant instant = Instant.now().minusMillis(TimeUnit.MINUTES.toMillis(5)); int total = 0; for (HookType hookType : allSubscribes.keySet()) { diff --git a/src/main/java/com/genersoft/iot/vmp/service/IInviteStreamService.java b/src/main/java/com/genersoft/iot/vmp/service/IInviteStreamService.java new file mode 100644 index 00000000..439cdded --- /dev/null +++ b/src/main/java/com/genersoft/iot/vmp/service/IInviteStreamService.java @@ -0,0 +1,63 @@ +package com.genersoft.iot.vmp.service; + +import com.genersoft.iot.vmp.common.InviteInfo; +import com.genersoft.iot.vmp.common.InviteSessionType; +import com.genersoft.iot.vmp.service.bean.InviteErrorCallback; + +/** + * 记录国标点播的状态，包括实时预览，下载，录像回放 + */ +public interface IInviteStreamService { + + /** + * 更新点播的状态信息 + */ + void updateInviteInfo(InviteInfo inviteInfo); + + /** + * 获取点播的状态信息 + */ + InviteInfo getInviteInfo(InviteSessionType type, + String deviceId, + String channelId, + String stream); + + /** + * 移除点播的状态信息 + */ + void removeInviteInfo(InviteSessionType type, + String deviceId, + String channelId, + String stream); + /** + * 移除点播的状态信息 + */ + void removeInviteInfo(InviteInfo inviteInfo); + /** + * 移除点播的状态信息 + */ + void removeInviteInfoByDeviceAndChannel(InviteSessionType inviteSessionType, String deviceId, String channelId); + + /** + * 获取点播的状态信息 + */ + InviteInfo getInviteInfoByDeviceAndChannel(InviteSessionType type, + String deviceId, + String channelId); + + /** + * 获取点播的状态信息 + */ + InviteInfo getInviteInfoByStream(InviteSessionType type, String stream); + + + /** + * 添加一个invite回调 + */ + void once(InviteSessionType type, String deviceId, String channelId, String stream, InviteErrorCallback callback); + + /** + * 调用一个invite回调 + */ + void call(InviteSessionType type, String deviceId, String channelId, String stream, int code, String msg, Object data); +} diff --git a/src/main/java/com/genersoft/iot/vmp/service/IPlayService.java b/src/main/java/com/genersoft/iot/vmp/service/IPlayService.java index ad59cb6e..b2b03089 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/IPlayService.java +++ b/src/main/java/com/genersoft/iot/vmp/service/IPlayService.java @@ -1,15 +1,11 @@ package com.genersoft.iot.vmp.service; -import com.alibaba.fastjson2.JSONObject; import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.conf.exception.ServiceException; import com.genersoft.iot.vmp.gb28181.bean.Device; import com.genersoft.iot.vmp.gb28181.bean.InviteStreamCallback; -import com.genersoft.iot.vmp.gb28181.bean.InviteStreamInfo; -import com.genersoft.iot.vmp.gb28181.event.SipSubscribe; -import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; -import com.genersoft.iot.vmp.service.bean.InviteTimeOutCallback; +import com.genersoft.iot.vmp.service.bean.InviteErrorCallback; import com.genersoft.iot.vmp.service.bean.PlayBackCallback; import com.genersoft.iot.vmp.service.bean.SSRCInfo; @@ -22,12 +18,9 @@ import java.text.ParseException; */ public interface IPlayService { - void onPublishHandlerForPlay(MediaServerItem mediaServerItem, JSONObject resonse, String deviceId, String channelId); - void play(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId, - ZlmHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent, - InviteTimeOutCallback timeoutCallback); - void play(MediaServerItem mediaServerItem, String deviceId, String channelId, ZlmHttpHookSubscribe.Event event, SipSubscribe.Event errorEvent, Runnable timeoutCallback); + InviteErrorCallback callback); + SSRCInfo play(MediaServerItem mediaServerItem, String deviceId, String channelId, InviteErrorCallback callback); MediaServerItem getNewMediaServerItem(Device device); @@ -36,8 +29,6 @@ public interface IPlayService { */ MediaServerItem getNewMediaServerItemHasAssist(Device device); - void onPublishHandlerForDownload(InviteStreamInfo inviteStreamInfo, String deviceId, String channelId, String toString); - void playBack(String deviceId, String channelId, String startTime, String endTime, InviteStreamCallback infoCallBack, PlayBackCallback playBackCallback); void playBack(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, String deviceId, String channelId, String startTime, String endTime, InviteStreamCallback infoCallBack, PlayBackCallback hookCallBack); diff --git a/src/main/java/com/genersoft/iot/vmp/service/bean/InviteErrorCallback.java b/src/main/java/com/genersoft/iot/vmp/service/bean/InviteErrorCallback.java new file mode 100644 index 00000000..974057e5 --- /dev/null +++ b/src/main/java/com/genersoft/iot/vmp/service/bean/InviteErrorCallback.java @@ -0,0 +1,6 @@ +package com.genersoft.iot.vmp.service.bean; + +public interface InviteErrorCallback { + + void run(int code, String msg, T data); +} diff --git a/src/main/java/com/genersoft/iot/vmp/service/bean/InviteErrorCode.java b/src/main/java/com/genersoft/iot/vmp/service/bean/InviteErrorCode.java new file mode 100644 index 00000000..3f3c76b3 --- /dev/null +++ b/src/main/java/com/genersoft/iot/vmp/service/bean/InviteErrorCode.java @@ -0,0 +1,34 @@ +package com.genersoft.iot.vmp.service.bean; + +/** + * 全局错误码 + */ +public enum InviteErrorCode { + SUCCESS(0, "成功"), + ERROR_FOR_SIGNALLING_TIMEOUT(-1, "点播超时"), + ERROR_FOR_STREAM_TIMEOUT(-2, "收流超时"), + ERROR_FOR_RESOURCE_EXHAUSTION(-3, "资源耗尽"), + ERROR_FOR_CATCH_DATA(-4, "缓存数据异常"), + ERROR_FOR_SIGNALLING_ERROR(-5, "收到信令错误"), + ERROR_FOR_STREAM_PARSING_EXCEPTIONS(-6, "流地址解析错误"), + ERROR_FOR_SDP_PARSING_EXCEPTIONS(-7, "SDP信息解析失败"), + ERROR_FOR_SSRC_UNAVAILABLE(-8, "SSRC不可用"), + ERROR_FOR_RESET_SSRC(-9, "重新设置收流信息失败"), + ERROR_FOR_SIP_SENDING_FAILED(-10, "命令发送失败"); + + private final int code; + private final String msg; + + InviteErrorCode(int code, String msg) { + this.code = code; + this.msg = msg; + } + + public int getCode() { + return code; + } + + public String getMsg() { + return msg; + } +} diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceChannelServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceChannelServiceImpl.java index 229bc0d2..73adf2e4 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceChannelServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceChannelServiceImpl.java @@ -1,10 +1,12 @@ package com.genersoft.iot.vmp.service.impl; -import com.genersoft.iot.vmp.common.StreamInfo; +import com.genersoft.iot.vmp.common.InviteInfo; +import com.genersoft.iot.vmp.common.InviteSessionType; import com.genersoft.iot.vmp.gb28181.bean.Device; import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel; import com.genersoft.iot.vmp.gb28181.utils.Coordtransform; import com.genersoft.iot.vmp.service.IDeviceChannelService; +import com.genersoft.iot.vmp.service.IInviteStreamService; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.dao.DeviceChannelMapper; import com.genersoft.iot.vmp.storager.dao.DeviceMapper; @@ -32,6 +34,9 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService { @Autowired private IRedisCatchStorage redisCatchStorage; + @Autowired + private IInviteStreamService inviteStreamService; + @Autowired private DeviceChannelMapper channelMapper; @@ -78,9 +83,10 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService { public void updateChannel(String deviceId, DeviceChannel channel) { String channelId = channel.getChannelId(); channel.setDeviceId(deviceId); - StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId); - if (streamInfo != null) { - channel.setStreamId(streamInfo.getStream()); +// StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId); + InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId); + if (inviteInfo != null && inviteInfo.getStreamInfo() != null) { + channel.setStreamId(inviteInfo.getStreamInfo().getStream()); } String now = DateUtil.getNow(); channel.setUpdateTime(now); @@ -106,9 +112,9 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService { if (channelList.size() == 0) { for (DeviceChannel channel : channels) { channel.setDeviceId(deviceId); - StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channel.getChannelId()); - if (streamInfo != null) { - channel.setStreamId(streamInfo.getStream()); + InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channel.getChannelId()); + if (inviteInfo != null && inviteInfo.getStreamInfo() != null) { + channel.setStreamId(inviteInfo.getStreamInfo().getStream()); } String now = DateUtil.getNow(); channel.setUpdateTime(now); @@ -122,9 +128,9 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService { } for (DeviceChannel channel : channels) { channel.setDeviceId(deviceId); - StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channel.getChannelId()); - if (streamInfo != null) { - channel.setStreamId(streamInfo.getStream()); + InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channel.getChannelId()); + if (inviteInfo != null && inviteInfo.getStreamInfo() != null) { + channel.setStreamId(inviteInfo.getStreamInfo().getStream()); } String now = DateUtil.getNow(); channel.setUpdateTime(now); diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/InviteStreamServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/InviteStreamServiceImpl.java new file mode 100644 index 00000000..8b8c839e --- /dev/null +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/InviteStreamServiceImpl.java @@ -0,0 +1,178 @@ +package com.genersoft.iot.vmp.service.impl; + +import com.alibaba.fastjson2.JSON; +import com.genersoft.iot.vmp.common.InviteInfo; +import com.genersoft.iot.vmp.common.InviteSessionStatus; +import com.genersoft.iot.vmp.common.InviteSessionType; +import com.genersoft.iot.vmp.common.VideoManagerConstants; +import com.genersoft.iot.vmp.service.IInviteStreamService; +import com.genersoft.iot.vmp.service.bean.InviteErrorCallback; +import com.genersoft.iot.vmp.utils.redis.RedisUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.stereotype.Service; + +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; + +@Service +public class InviteStreamServiceImpl implements IInviteStreamService { + + private final Logger logger = LoggerFactory.getLogger(InviteStreamServiceImpl.class); + + private final Map>> inviteErrorCallbackMap = new ConcurrentHashMap<>(); + + @Autowired + private RedisTemplate redisTemplate; + + @Override + public void updateInviteInfo(InviteInfo inviteInfo) { + if (inviteInfo == null || (inviteInfo.getDeviceId() == null || inviteInfo.getChannelId() == null)) { + logger.warn("[更新Invite信息]，参数不全： {}", JSON.toJSON(inviteInfo)); + return; + } + InviteInfo inviteInfoForUpdate = null; + + if (InviteSessionStatus.ready == inviteInfo.getStatus()) { + if (inviteInfo.getDeviceId() == null + || inviteInfo.getChannelId() == null + || inviteInfo.getType() == null + || inviteInfo.getStream() == null + ) { + return; + } + inviteInfoForUpdate = inviteInfo; + } else { + InviteInfo inviteInfoInRedis = getInviteInfo(inviteInfo.getType(), inviteInfo.getDeviceId(), + inviteInfo.getChannelId(), inviteInfo.getStream()); + if (inviteInfoInRedis == null) { + logger.warn("[更新Invite信息]，未从缓存中读取到Invite信息： deviceId: {}, channel: {}, stream: {}", + inviteInfo.getDeviceId(), inviteInfo.getChannelId(), inviteInfo.getStream()); + return; + } + if (inviteInfo.getStreamInfo() != null) { + inviteInfoInRedis.setStreamInfo(inviteInfo.getStreamInfo()); + } + if (inviteInfo.getSsrcInfo() != null) { + inviteInfoInRedis.setSsrcInfo(inviteInfo.getSsrcInfo()); + } + if (inviteInfo.getStreamMode() != null) { + inviteInfoInRedis.setStreamMode(inviteInfo.getStreamMode()); + } + if (inviteInfo.getReceiveIp() != null) { + inviteInfoInRedis.setReceiveIp(inviteInfo.getReceiveIp()); + } + if (inviteInfo.getReceivePort() != null) { + inviteInfoInRedis.setReceivePort(inviteInfo.getReceivePort()); + } + if (inviteInfo.getStatus() != null) { + inviteInfoInRedis.setStatus(inviteInfo.getStatus()); + } + + inviteInfoForUpdate = inviteInfoInRedis; + + } + String key = VideoManagerConstants.INVITE_PREFIX + + "_" + inviteInfoForUpdate.getType() + + "_" + inviteInfoForUpdate.getDeviceId() + + "_" + inviteInfoForUpdate.getChannelId() + + "_" + inviteInfoForUpdate.getStream(); + redisTemplate.opsForValue().set(key, inviteInfoForUpdate); + } + + @Override + public InviteInfo getInviteInfo(InviteSessionType type, String deviceId, String channelId, String stream) { + String key = VideoManagerConstants.INVITE_PREFIX + + "_" + (type != null ? type : "*") + + "_" + (deviceId != null ? deviceId : "*") + + "_" + (channelId != null ? channelId : "*") + + "_" + (stream != null ? stream : "*"); + List scanResult = RedisUtil.scan(redisTemplate, key); + if (scanResult.size() != 1) { + return null; + } + + return (InviteInfo) redisTemplate.opsForValue().get(scanResult.get(0)); + } + + @Override + public InviteInfo getInviteInfoByDeviceAndChannel(InviteSessionType type, String deviceId, String channelId) { + return getInviteInfo(type, deviceId, channelId, null); + } + + @Override + public InviteInfo getInviteInfoByStream(InviteSessionType type, String stream) { + return getInviteInfo(type, null, null, stream); + } + + @Override + public void removeInviteInfo(InviteSessionType type, String deviceId, String channelId, String stream) { + String scanKey = VideoManagerConstants.INVITE_PREFIX + + "_" + (type != null ? type : "*") + + "_" + (deviceId != null ? deviceId : "*") + + "_" + (channelId != null ? channelId : "*") + + "_" + (stream != null ? stream : "*"); + List scanResult = RedisUtil.scan(redisTemplate, scanKey); + if (scanResult.size() > 0) { + for (Object keyObj : scanResult) { + String key = (String) keyObj; + InviteInfo inviteInfo = (InviteInfo) redisTemplate.opsForValue().get(key); + if (inviteInfo == null) { + continue; + } + redisTemplate.delete(key); + inviteErrorCallbackMap.remove(buildKey(type, deviceId, channelId, inviteInfo.getStream())); + } + } + } + + @Override + public void removeInviteInfoByDeviceAndChannel(InviteSessionType inviteSessionType, String deviceId, String channelId) { + removeInviteInfo(inviteSessionType, deviceId, channelId, null); + } + + @Override + public void removeInviteInfo(InviteInfo inviteInfo) { + removeInviteInfo(inviteInfo.getType(), inviteInfo.getDeviceId(), inviteInfo.getChannelId(), inviteInfo.getStream()); + } + + @Override + public void once(InviteSessionType type, String deviceId, String channelId, String stream, InviteErrorCallback callback) { + String key = buildKey(type, deviceId, channelId, stream); + List> callbacks = inviteErrorCallbackMap.get(key); + if (callbacks == null) { + callbacks = new CopyOnWriteArrayList<>(); + inviteErrorCallbackMap.put(key, callbacks); + } + callbacks.add(callback); + + } + + @Override + public void call(InviteSessionType type, String deviceId, String channelId, String stream, int code, String msg, Object data) { + String key = buildKey(type, deviceId, channelId, stream); + List> callbacks = inviteErrorCallbackMap.get(key); + if (callbacks == null) { + return; + } + for (InviteErrorCallback callback : callbacks) { + callback.run(code, msg, data); + } + inviteErrorCallbackMap.remove(key); + } + + private String buildKey(InviteSessionType type, String deviceId, String channelId, String stream) { + String key = type + "_" + deviceId + "_" + channelId; + // 如果ssrc未null那么可以实现一个通道只能一次操作，ssrc不为null则可以支持一个通道多次invite + if (stream != null) { + key += ("_" + stream); + } + return key; + } + + +} 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 96e4098a..1fcac383 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 @@ -3,6 +3,9 @@ package com.genersoft.iot.vmp.service.impl; import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSONArray; import com.alibaba.fastjson2.JSONObject; +import com.genersoft.iot.vmp.common.InviteInfo; +import com.genersoft.iot.vmp.common.InviteSessionStatus; +import com.genersoft.iot.vmp.common.InviteSessionType; import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.conf.DynamicTask; import com.genersoft.iot.vmp.conf.UserSetting; @@ -19,18 +22,13 @@ 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.media.zlm.AssistRESTfulUtils; import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils; +import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory; import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe; import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory; import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; -import com.genersoft.iot.vmp.service.IDeviceService; -import com.genersoft.iot.vmp.service.IMediaServerService; -import com.genersoft.iot.vmp.service.IMediaService; -import com.genersoft.iot.vmp.service.IPlayService; -import com.genersoft.iot.vmp.service.bean.InviteTimeOutCallback; -import com.genersoft.iot.vmp.service.bean.PlayBackCallback; -import com.genersoft.iot.vmp.service.bean.PlayBackResult; -import com.genersoft.iot.vmp.service.bean.SSRCInfo; +import com.genersoft.iot.vmp.service.*; +import com.genersoft.iot.vmp.service.bean.*; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage; import com.genersoft.iot.vmp.utils.DateUtil; @@ -72,12 +70,18 @@ public class PlayServiceImpl implements IPlayService { @Autowired private IRedisCatchStorage redisCatchStorage; + @Autowired + private IInviteStreamService inviteStreamService; + @Autowired private DeferredResultHolder resultHolder; @Autowired private ZLMRESTfulUtils zlmresTfulUtils; + @Autowired + private ZLMRTPServerFactory zlmrtpServerFactory; + @Autowired private AssistRESTfulUtils assistRESTfulUtils; @@ -111,137 +115,122 @@ public class PlayServiceImpl implements IPlayService { @Override - public void play(MediaServerItem mediaServerItem, String deviceId, String channelId, - ZlmHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent, - Runnable timeoutCallback) { + public SSRCInfo play(MediaServerItem mediaServerItem, String deviceId, String channelId, InviteErrorCallback callback) { if (mediaServerItem == null) { throw new ControllerException(ErrorCode.ERROR100.getCode(), "未找到可用的zlm"); } - String key = DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId; - - RequestMessage msg = new RequestMessage(); - msg.setKey(key); Device device = redisCatchStorage.getDevice(deviceId); - StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId); + InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId); - if (streamInfo != null) { - String streamId = streamInfo.getStream(); - if (streamId == null) { - WVPResult wvpResult = new WVPResult(); - wvpResult.setCode(ErrorCode.ERROR100.getCode()); - wvpResult.setMsg("点播失败， redis缓存streamId等于null"); - msg.setData(wvpResult); - resultHolder.invokeAllResult(msg); - return; - } - String mediaServerId = streamInfo.getMediaServerId(); - MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId); + if (inviteInfo != null ) { + System.out.println("inviteInfo 已存在"); + if (inviteInfo.getStreamInfo() == null) { + System.out.println("inviteInfo 已存在， StreamInfo 不存在，添加回调等待"); + // 点播发起了但是尚未成功, 仅注册回调等待结果即可 + inviteStreamService.once(InviteSessionType.PLAY, deviceId, channelId, null, callback); + return inviteInfo.getSsrcInfo(); + }else { + StreamInfo streamInfo = inviteInfo.getStreamInfo(); + String streamId = streamInfo.getStream(); + if (streamId == null) { + callback.run(InviteErrorCode.ERROR_FOR_CATCH_DATA.getCode(), "点播失败， redis缓存streamId等于null", null); + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, + InviteErrorCode.ERROR_FOR_CATCH_DATA.getCode(), + "点播失败， redis缓存streamId等于null", + null); + return inviteInfo.getSsrcInfo(); + } + String mediaServerId = streamInfo.getMediaServerId(); + MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId); - JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(mediaInfo, streamId); - if (rtpInfo.getInteger("code") == 0) { - if (rtpInfo.getBoolean("exist")) { - int localPort = rtpInfo.getInteger("local_port"); - if (localPort == 0) { - logger.warn("[点播]，点播时发现rtpServer存在，但是尚未开始推流"); - // 此时说明rtpServer已经创建但是流还没有推上来 - WVPResult wvpResult = new WVPResult(); - wvpResult.setCode(ErrorCode.ERROR100.getCode()); - wvpResult.setMsg("点播已经在进行中，请稍候重试"); - msg.setData(wvpResult); - - resultHolder.invokeAllResult(msg); - return; - } else { - WVPResult wvpResult = new WVPResult(); - wvpResult.setCode(ErrorCode.SUCCESS.getCode()); - wvpResult.setMsg(ErrorCode.SUCCESS.getMsg()); - wvpResult.setData(streamInfo); - msg.setData(wvpResult); - resultHolder.invokeAllResult(msg); - if (hookEvent != null) { - hookEvent.response(mediaServerItem, JSON.parseObject(JSON.toJSONString(streamInfo))); - } - } - - } else { - redisCatchStorage.stopPlay(streamInfo); + Boolean ready = zlmrtpServerFactory.isStreamReady(mediaInfo, "rtp", streamId); + if (ready != null && ready) { + callback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), streamInfo); + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, + InviteErrorCode.SUCCESS.getCode(), + InviteErrorCode.SUCCESS.getMsg(), + streamInfo); + return inviteInfo.getSsrcInfo(); + }else { + // 点播发起了但是尚未成功, 仅注册回调等待结果即可 + inviteStreamService.once(InviteSessionType.PLAY, deviceId, channelId, null, callback); storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId()); - streamInfo = null; + inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId); } - } else { - //zlm连接失败 - redisCatchStorage.stopPlay(streamInfo); - storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId()); - streamInfo = null; - } } - if (streamInfo == null) { - String streamId = null; - if (mediaServerItem.isRtpEnable()) { - streamId = String.format("%s_%s", device.getDeviceId(), channelId); - } - SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, null, device.isSsrcCheck(), false, 0, false, device.getStreamModeForParam()); - if (ssrcInfo == null) { - WVPResult wvpResult = new WVPResult(); - wvpResult.setCode(ErrorCode.ERROR100.getCode()); - wvpResult.setMsg("开启收流失败"); - msg.setData(wvpResult); - resultHolder.invokeAllResult(msg); - return; - } - play(mediaServerItem, ssrcInfo, device, channelId, (mediaServerItemInUse, response) -> { - if (hookEvent != null) { - hookEvent.response(mediaServerItem, response); - } - }, event -> { - // sip error错误 - WVPResult wvpResult = new WVPResult(); - wvpResult.setCode(ErrorCode.ERROR100.getCode()); - wvpResult.setMsg(String.format("点播失败， 错误码： %s, %s", event.statusCode, event.msg)); - msg.setData(wvpResult); - resultHolder.invokeAllResult(msg); - if (errorEvent != null) { - errorEvent.response(event); - } - }, (code, msgStr) -> { - // invite点播超时 - WVPResult wvpResult = new WVPResult(); - wvpResult.setCode(ErrorCode.ERROR100.getCode()); - if (code == 0) { - wvpResult.setMsg("点播超时，请稍候重试"); - } else if (code == 1) { - wvpResult.setMsg("收流超时，请稍候重试"); - } - msg.setData(wvpResult); - // 回复之前所有的点播请求 - resultHolder.invokeAllResult(msg); - }); + String streamId = null; + if (mediaServerItem.isRtpEnable()) { + streamId = String.format("%s_%s", device.getDeviceId(), channelId); } + SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, null, device.isSsrcCheck(), false, 0, false, device.getStreamModeForParam()); + if (ssrcInfo == null) { + callback.run(InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getCode(), InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getMsg(), null); + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, + InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getCode(), + InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getMsg(), + null); + return null; + } + // TODO 记录点播的状态 + play(mediaServerItem, ssrcInfo, device, channelId, callback); + return ssrcInfo; } @Override public void play(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId, - ZlmHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent, - InviteTimeOutCallback timeoutCallback) { + InviteErrorCallback callback) { logger.info("[点播开始] deviceId: {}, channelId: {},收流端口：{}, 收流模式：{}, SSRC: {}, SSRC校验：{}", device.getDeviceId(), channelId, ssrcInfo.getPort(), device.getStreamMode(), ssrcInfo.getSsrc(), device.isSsrcCheck()); + + //端口获取失败的ssrcInfo 没有必要发送点播指令 + if (ssrcInfo.getPort() <= 0) { + logger.info("[点播端口分配异常]，deviceId={},channelId={},ssrcInfo={}", device.getDeviceId(), channelId, ssrcInfo); + // 释放ssrc + mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); + streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); + + callback.run(InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getCode(), "点播端口分配异常", null); + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, + InviteErrorCode.ERROR_FOR_RESOURCE_EXHAUSTION.getCode(), "点播端口分配异常", null); + return; + } + + // 初始化redis中的invite消息状态 + InviteInfo inviteInfo = InviteInfo.getinviteInfo(device.getDeviceId(), channelId, ssrcInfo.getStream(), ssrcInfo, + mediaServerItem.getSdpIp(), ssrcInfo.getPort(), device.getStreamMode(), InviteSessionType.PLAY, + InviteSessionStatus.ready); + inviteStreamService.updateInviteInfo(inviteInfo); // 超时处理 String timeOutTaskKey = UUID.randomUUID().toString(); dynamicTask.startDelay(timeOutTaskKey, () -> { // 执行超时任务时查询是否已经成功，成功了则不执行超时任务，防止超时任务取消失败的情况 - if (redisCatchStorage.queryPlayByDevice(device.getDeviceId(), channelId) == null) { + InviteInfo inviteInfoForTimeOut = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channelId); + if (inviteInfoForTimeOut == null || inviteInfoForTimeOut.getStreamInfo() == null) { logger.info("[点播超时] 收流超时 deviceId: {}, channelId: {}，端口：{}, SSRC: {}", device.getDeviceId(), channelId, ssrcInfo.getPort(), ssrcInfo.getSsrc()); // 点播超时回复BYE 同时释放ssrc以及此次点播的资源 +// InviteInfo inviteInfoForTimeout = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.play, device.getDeviceId(), channelId); +// if (inviteInfoForTimeout == null) { +// return; +// } +// if (InviteSessionStatus.ok == inviteInfoForTimeout.getStatus() ) { +// // TODO 发送bye +// }else { +// // TODO 发送cancel +// } + callback.run(InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getCode(), InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getMsg(), null); + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, + InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getCode(), InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getMsg(), null); + + inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channelId); try { cmder.streamByeCmd(device, channelId, ssrcInfo.getStream(), null); } catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) { logger.error("[点播超时]， 发送BYE失败 {}", e.getMessage()); } finally { - timeoutCallback.run(1, "收流超时"); mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream()); streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); @@ -252,28 +241,26 @@ public class PlayServiceImpl implements IPlayService { } } }, userSetting.getPlayTimeout()); - //端口获取失败的ssrcInfo 没有必要发送点播指令 - if (ssrcInfo.getPort() <= 0) { - logger.info("[点播端口分配异常]，deviceId={},channelId={},ssrcInfo={}", device.getDeviceId(), channelId, ssrcInfo); - dynamicTask.stop(timeOutTaskKey); - // 释放ssrc - mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); - streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); - RequestMessage msg = new RequestMessage(); - msg.setKey(DeferredResultHolder.CALLBACK_CMD_PLAY + device.getDeviceId() + channelId); - msg.setData(WVPResult.fail(ErrorCode.ERROR100.getCode(), "点播端口分配异常")); - resultHolder.invokeAllResult(msg); - return; - } try { cmder.playStreamCmd(mediaServerItem, ssrcInfo, device, channelId, (MediaServerItem mediaServerItemInuse, JSONObject response) -> { logger.info("收到订阅消息： " + response.toJSONString()); dynamicTask.stop(timeOutTaskKey); - // hook响应 - onPublishHandlerForPlay(mediaServerItemInuse, response, device.getDeviceId(), channelId); - hookEvent.response(mediaServerItemInuse, response); + StreamInfo streamInfo = onPublishHandlerForPlay(mediaServerItemInuse, response, device.getDeviceId(), channelId); + if (streamInfo == null){ + callback.run(InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getCode(), + InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getMsg(), null); + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, + InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getCode(), + InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getMsg(), null); + return; + } + callback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), streamInfo); + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, + InviteErrorCode.SUCCESS.getCode(), + InviteErrorCode.SUCCESS.getMsg(), + streamInfo); logger.info("[点播成功] deviceId: {}, channelId: {}", device.getDeviceId(), channelId); String streamUrl; if (mediaServerItemInuse.getRtspPort() != 0) { @@ -288,6 +275,8 @@ public class PlayServiceImpl implements IPlayService { zlmresTfulUtils.getSnap(mediaServerItemInuse, streamUrl, 15, 1, path, fileName); }, (event) -> { + inviteInfo.setStatus(InviteSessionStatus.ok); + ResponseEvent responseEvent = (ResponseEvent) event.event; String contentString = new String(responseEvent.getResponse().getRawContent()); // 获取ssrc @@ -319,6 +308,18 @@ public class PlayServiceImpl implements IPlayService { logger.info("[点播-TCP主动连接对方] 结果： {}", jsonObject); } catch (SdpException e) { logger.error("[点播-TCP主动连接对方] deviceId: {}, channelId: {}, 解析200OK的SDP信息失败", device.getDeviceId(), channelId, e); + dynamicTask.stop(timeOutTaskKey); + mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream()); + // 释放ssrc + mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); + + streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); + + callback.run(InviteErrorCode.ERROR_FOR_SDP_PARSING_EXCEPTIONS.getCode(), + InviteErrorCode.ERROR_FOR_SDP_PARSING_EXCEPTIONS.getMsg(), null); + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, + InviteErrorCode.ERROR_FOR_SDP_PARSING_EXCEPTIONS.getCode(), + InviteErrorCode.ERROR_FOR_SDP_PARSING_EXCEPTIONS.getMsg(), null); } } return; @@ -332,9 +333,13 @@ public class PlayServiceImpl implements IPlayService { // 释放ssrc ssrcFactory.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); - event.msg = "下级自定义了ssrc,但是此ssrc不可用"; - event.statusCode = 400; - errorEvent.response(event); + + callback.run(InviteErrorCode.ERROR_FOR_SSRC_UNAVAILABLE.getCode(), + InviteErrorCode.ERROR_FOR_SSRC_UNAVAILABLE.getMsg(), null); + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, + InviteErrorCode.ERROR_FOR_SSRC_UNAVAILABLE.getCode(), + InviteErrorCode.ERROR_FOR_SSRC_UNAVAILABLE.getMsg(), null); + return; } // 单端口模式streamId也有变化，重新设置监听即可 @@ -342,18 +347,32 @@ public class PlayServiceImpl implements IPlayService { // 添加订阅 HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId()); subscribe.removeSubscribe(hookSubscribe); - hookSubscribe.getContent().put("stream", String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase()); + String stream = String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase(); + hookSubscribe.getContent().put("stream", stream); + inviteInfo.setStream(stream); subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject response) -> { logger.info("[ZLM HOOK] ssrc修正后收到订阅消息： " + response.toJSONString()); dynamicTask.stop(timeOutTaskKey); // hook响应 - onPublishHandlerForPlay(mediaServerItemInUse, response, device.getDeviceId(), channelId); - hookEvent.response(mediaServerItemInUse, response); + StreamInfo streamInfo = onPublishHandlerForPlay(mediaServerItemInUse, response, device.getDeviceId(), channelId); + if (streamInfo == null){ + callback.run(InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getCode(), + InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getMsg(), null); + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, + InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getCode(), + InviteErrorCode.ERROR_FOR_STREAM_PARSING_EXCEPTIONS.getMsg(), null); + return; + } + callback.run(InviteErrorCode.SUCCESS.getCode(), + InviteErrorCode.SUCCESS.getMsg(), null); + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, + InviteErrorCode.SUCCESS.getCode(), + InviteErrorCode.SUCCESS.getMsg(), + streamInfo); }); return; } - // 更新ssrc Boolean result = mediaServerService.updateRtpServerSSRC(mediaServerItem, ssrcInfo.getStream(), ssrcInResponse); if (!result) { @@ -370,14 +389,23 @@ public class PlayServiceImpl implements IPlayService { mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); - event.msg = "下级自定义了ssrc,重新设置收流信息失败"; - event.statusCode = 500; - errorEvent.response(event); + + callback.run(InviteErrorCode.ERROR_FOR_RESET_SSRC.getCode(), + "下级自定义了ssrc,重新设置收流信息失败", null); + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, + InviteErrorCode.ERROR_FOR_RESET_SSRC.getCode(), + "下级自定义了ssrc,重新设置收流信息失败", null); + + }else { + ssrcInfo.setSsrc(ssrcInResponse); + inviteInfo.setSsrcInfo(ssrcInfo); + inviteInfo.setStream(ssrcInfo.getStream()); } }else { logger.info("[点播消息] 收到invite 200, 下级自定义了ssrc, 但是当前模式无需修正"); } } + inviteStreamService.updateInviteInfo(inviteInfo); }, (event) -> { dynamicTask.stop(timeOutTaskKey); mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream()); @@ -385,7 +413,14 @@ public class PlayServiceImpl implements IPlayService { mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); - errorEvent.response(event); + + callback.run(InviteErrorCode.ERROR_FOR_SIGNALLING_ERROR.getCode(), + String.format("点播失败， 错误码： %s, %s", event.statusCode, event.msg), null); + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, + InviteErrorCode.ERROR_FOR_RESET_SSRC.getCode(), + String.format("点播失败， 错误码： %s, %s", event.statusCode, event.msg), null); + + inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channelId); }); } catch (InvalidArgumentException | SipException | ParseException e) { @@ -396,40 +431,34 @@ public class PlayServiceImpl implements IPlayService { mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc()); streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream()); - SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(); - eventResult.type = SipSubscribe.EventResultType.cmdSendFailEvent; - eventResult.statusCode = -1; - eventResult.msg = "命令发送失败"; - errorEvent.response(eventResult); + + callback.run(InviteErrorCode.ERROR_FOR_SIP_SENDING_FAILED.getCode(), + InviteErrorCode.ERROR_FOR_SIP_SENDING_FAILED.getMsg(), null); + inviteStreamService.call(InviteSessionType.PLAY, device.getDeviceId(), channelId, null, + InviteErrorCode.ERROR_FOR_SIP_SENDING_FAILED.getCode(), + InviteErrorCode.ERROR_FOR_SIP_SENDING_FAILED.getMsg(), null); + + inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channelId); } } - @Override - public void onPublishHandlerForPlay(MediaServerItem mediaServerItem, JSONObject response, String deviceId, String channelId) { + private StreamInfo onPublishHandlerForPlay(MediaServerItem mediaServerItem, JSONObject response, String deviceId, String channelId) { StreamInfo streamInfo = onPublishHandler(mediaServerItem, response, deviceId, channelId); - RequestMessage msg = new RequestMessage(); - msg.setKey(DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId); if (streamInfo != null) { DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId); if (deviceChannel != null) { deviceChannel.setStreamId(streamInfo.getStream()); storager.startPlay(deviceId, channelId, streamInfo.getStream()); } - redisCatchStorage.startPlay(streamInfo); - - WVPResult wvpResult = new WVPResult(); - wvpResult.setCode(ErrorCode.SUCCESS.getCode()); - wvpResult.setMsg(ErrorCode.SUCCESS.getMsg()); - wvpResult.setData(streamInfo); - - msg.setData(wvpResult); - resultHolder.invokeAllResult(msg); - - } else { - logger.warn("设备预览API调用失败！"); - msg.setData(WVPResult.fail(ErrorCode.ERROR100.getCode(), "设备预览API调用失败！")); - resultHolder.invokeAllResult(msg); + InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId); + if (inviteInfo != null) { + inviteInfo.setStatus(InviteSessionStatus.ok); + inviteInfo.setStreamInfo(streamInfo); + inviteStreamService.updateInviteInfo(inviteInfo); + } } + return streamInfo; + } private void onPublishHandlerForPlayback(MediaServerItem mediaServerItem, JSONObject response, String deviceId, String channelId, PlayBackCallback playBackCallback) { @@ -442,8 +471,12 @@ public class PlayServiceImpl implements IPlayService { deviceChannel.setStreamId(streamInfo.getStream()); storager.startPlay(deviceId, channelId, streamInfo.getStream()); } - redisCatchStorage.startPlay(streamInfo); - + InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAYBACK, deviceId, channelId); + if (inviteInfo != null) { + inviteInfo.setStatus(InviteSessionStatus.ok); + inviteInfo.setStreamInfo(streamInfo); + inviteStreamService.updateInviteInfo(inviteInfo); + } playBackResult.setCode(ErrorCode.SUCCESS.getCode()); playBackResult.setMsg(ErrorCode.SUCCESS.getMsg()); @@ -560,6 +593,7 @@ public class PlayServiceImpl implements IPlayService { return; } redisCatchStorage.startPlayback(streamInfo, inviteStreamInfo.getCallId()); + playBackResult.setCode(ErrorCode.SUCCESS.getCode()); playBackResult.setMsg(ErrorCode.SUCCESS.getMsg()); playBackResult.setData(streamInfo); @@ -858,8 +892,7 @@ public class PlayServiceImpl implements IPlayService { return streamInfo; } - @Override - public void onPublishHandlerForDownload(InviteStreamInfo inviteStreamInfo, String deviceId, String channelId, String uuid) { + private void onPublishHandlerForDownload(InviteStreamInfo inviteStreamInfo, String deviceId, String channelId, String uuid) { RequestMessage msg = new RequestMessage(); msg.setKey(DeferredResultHolder.CALLBACK_CMD_DOWNLOAD + deviceId + channelId); msg.setId(uuid); diff --git a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGbPlayMsgListener.java b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGbPlayMsgListener.java index 868e861d..7399b2a7 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGbPlayMsgListener.java +++ b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGbPlayMsgListener.java @@ -264,8 +264,8 @@ public class RedisGbPlayMsgListener implements MessageListener { return; } // 确定流是否在线 - boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, content.getApp(), content.getStream()); - if (streamReady) { + Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, content.getApp(), content.getStream()); + if (streamReady != null && streamReady) { logger.info("[回复推流信息] {}/{}", content.getApp(), content.getStream()); responseSendItem(mediaServerItem, content, toId, serial); }else { @@ -301,9 +301,6 @@ public class RedisGbPlayMsgListener implements MessageListener { String key = VideoManagerConstants.VM_MSG_STREAM_PUSH_REQUESTED; logger.info("[redis发送通知] 推流被请求 {}: {}/{}", key, messageForPushChannel.getApp(), messageForPushChannel.getStream()); redisTemplate.convertAndSend(key, JSON.toJSON(messageForPushChannel)); - -// redisCatchStorage.sendStreamPushRequestedMsg(messageForPushChannel); - } } diff --git a/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java b/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java index 42708f7f..b6cfab4f 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java @@ -23,34 +23,6 @@ public interface IRedisCatchStorage { */ Long getCSEQ(); - /** - * 开始播放时将流存入 - * - * @param stream 流信息 - * @return - */ - boolean startPlay(StreamInfo stream); - - - /** - * 停止播放时删除 - * - * @return - */ - boolean stopPlay(StreamInfo streamInfo); - - /** - * 查询播放列表 - * @return - */ - StreamInfo queryPlay(StreamInfo streamInfo); - - StreamInfo queryPlayByStreamId(String steamId); - - StreamInfo queryPlayByDevice(String deviceId, String channelId); - - Map queryPlayByDeviceId(String deviceId); - boolean startPlayback(StreamInfo stream, String callId); boolean stopPlayback(String deviceId, String channelId, String stream, String callId); diff --git a/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java b/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java index aed98119..dabe9f82 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java @@ -92,87 +92,6 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage { } } - /** - * 开始播放时将流存入redis - */ - @Override - public boolean startPlay(StreamInfo stream) { - - redisTemplate.opsForValue().set(String.format("%S_%s_%s_%s_%s_%s", VideoManagerConstants.PLAYER_PREFIX, userSetting.getServerId(), - stream.getMediaServerId(), stream.getStream(), stream.getDeviceID(), stream.getChannelId()), - stream); - return true; - } - - /** - * 停止播放时从redis删除 - */ - @Override - public boolean stopPlay(StreamInfo streamInfo) { - if (streamInfo == null) { - return false; - } - Boolean result = redisTemplate.delete(String.format("%S_%s_%s_%s_%s_%s", VideoManagerConstants.PLAYER_PREFIX, - userSetting.getServerId(), - streamInfo.getMediaServerId(), - streamInfo.getStream(), - streamInfo.getDeviceID(), - streamInfo.getChannelId())); - return result != null && result; - } - - /** - * 查询播放列表 - */ - @Override - public StreamInfo queryPlay(StreamInfo streamInfo) { - return (StreamInfo)redisTemplate.opsForValue().get(String.format("%S_%s_%s_%s_%s_%s", - VideoManagerConstants.PLAYER_PREFIX, - userSetting.getServerId(), - streamInfo.getMediaServerId(), - streamInfo.getStream(), - streamInfo.getDeviceID(), - streamInfo.getChannelId())); - } - @Override - public StreamInfo queryPlayByStreamId(String streamId) { - List playLeys = RedisUtil.scan(redisTemplate, String.format("%S_%s_*_%s_*", VideoManagerConstants.PLAYER_PREFIX, userSetting.getServerId(), streamId)); - if (playLeys.size() == 0) { - return null; - } - return (StreamInfo)redisTemplate.opsForValue().get(playLeys.get(0).toString()); - } - - @Override - public StreamInfo queryPlayByDevice(String deviceId, String channelId) { - List playLeys = RedisUtil.scan(redisTemplate, String.format("%S_%s_*_*_%s_%s", VideoManagerConstants.PLAYER_PREFIX, - userSetting.getServerId(), - deviceId, - channelId)); - if (playLeys.size() == 0) { - return null; - } - return (StreamInfo)redisTemplate.opsForValue().get(playLeys.get(0).toString()); - } - - @Override - public Map queryPlayByDeviceId(String deviceId) { - Map streamInfos = new HashMap<>(); - List players = RedisUtil.scan(redisTemplate, String.format("%S_%s_*_*_%s_*", VideoManagerConstants.PLAYER_PREFIX, userSetting.getServerId(),deviceId)); - if (players.size() == 0) { - return streamInfos; - } - for (Object player : players) { - String key = (String) player; - StreamInfo streamInfo = JsonUtil.redisJsonToObject(redisTemplate, key, StreamInfo.class); - if (Objects.isNull(streamInfo)) { - continue; - } - streamInfos.put(streamInfo.getDeviceID() + "_" + streamInfo.getChannelId(), streamInfo); - } - return streamInfos; - } - @Override public boolean startPlayback(StreamInfo stream, String callId) { diff --git a/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java b/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java index 958cc689..eb828a89 100644 --- a/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java +++ b/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java @@ -1,7 +1,11 @@ package com.genersoft.iot.vmp.vmanager.gb28181.play; +import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSONArray; import com.alibaba.fastjson2.JSONObject; +import com.genersoft.iot.vmp.common.InviteInfo; +import com.genersoft.iot.vmp.common.InviteSessionStatus; +import com.genersoft.iot.vmp.common.InviteSessionType; import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.exception.ControllerException; @@ -14,12 +18,13 @@ import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage; import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander; import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; +import com.genersoft.iot.vmp.service.IInviteStreamService; import com.genersoft.iot.vmp.service.IMediaServerService; import com.genersoft.iot.vmp.service.IMediaService; import com.genersoft.iot.vmp.service.IPlayService; +import com.genersoft.iot.vmp.service.bean.InviteErrorCode; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage; -import com.genersoft.iot.vmp.vmanager.bean.DeferredResultEx; import com.genersoft.iot.vmp.vmanager.bean.ErrorCode; import com.genersoft.iot.vmp.vmanager.bean.StreamContent; import com.genersoft.iot.vmp.vmanager.bean.WVPResult; @@ -59,6 +64,9 @@ public class PlayController { @Autowired private IRedisCatchStorage redisCatchStorage; + @Autowired + private IInviteStreamService inviteStreamService; + @Autowired private ZLMRESTfulUtils zlmresTfulUtils; @@ -88,14 +96,12 @@ public class PlayController { Device device = storager.queryVideoDevice(deviceId); MediaServerItem newMediaServerItem = playService.getNewMediaServerItem(device); - RequestMessage msg = new RequestMessage(); + RequestMessage requestMessage = new RequestMessage(); String key = DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId; - boolean exist = resultHolder.exist(key, null); - msg.setKey(key); + requestMessage.setKey(key); String uuid = UUID.randomUUID().toString(); - msg.setId(uuid); + requestMessage.setId(uuid); DeferredResult> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue()); - DeferredResultEx> deferredResultEx = new DeferredResultEx<>(result); result.onTimeout(()->{ logger.info("点播接口等待超时"); @@ -103,32 +109,36 @@ public class PlayController { WVPResult wvpResult = new WVPResult<>(); wvpResult.setCode(ErrorCode.ERROR100.getCode()); wvpResult.setMsg("点播超时"); - msg.setData(wvpResult); - resultHolder.invokeResult(msg); + requestMessage.setData(wvpResult); + resultHolder.invokeResult(requestMessage); }); - // TODO 在点播未成功的情况下在此调用接口点播会导致返回的流地址ip错误 - deferredResultEx.setFilter(result1 -> { - WVPResult wvpResult1 = (WVPResult)result1; - WVPResult resultStream = new WVPResult<>(); - resultStream.setCode(wvpResult1.getCode()); - resultStream.setMsg(wvpResult1.getMsg()); - if (wvpResult1.getCode() == ErrorCode.SUCCESS.getCode()) { - StreamInfo data = wvpResult1.getData().clone(); - if (userSetting.getUseSourceIpAsStreamIp()) { - data.channgeStreamIp(request.getLocalName()); - } - resultStream.setData(new StreamContent(wvpResult1.getData())); - } - return resultStream; - }); - // 录像查询以channelId作为deviceId查询 - resultHolder.put(key, uuid, deferredResultEx); + resultHolder.put(key, uuid, result); - if (!exist) { - playService.play(newMediaServerItem, deviceId, channelId, null, null, null); - } + playService.play(newMediaServerItem, deviceId, channelId, ((code, msg, data) -> { + System.out.println("controller收到回调"); + System.out.println(JSON.toJSONString(data)); + WVPResult wvpResult = new WVPResult<>(); + if (code == InviteErrorCode.SUCCESS.getCode()) { + wvpResult.setCode(ErrorCode.SUCCESS.getCode()); + wvpResult.setMsg(ErrorCode.SUCCESS.getMsg()); + + if (data != null) { + StreamInfo streamInfo = (StreamInfo)data; + if (userSetting.getUseSourceIpAsStreamIp()) { + streamInfo.channgeStreamIp(request.getLocalName()); + } + wvpResult.setData(new StreamContent(streamInfo)); + } + }else { + wvpResult.setCode(code); + wvpResult.setMsg(msg); + } + System.out.println(JSON.toJSONString(wvpResult)); + requestMessage.setData(wvpResult); + resultHolder.invokeResult(requestMessage); + })); return result; } @@ -149,21 +159,22 @@ public class PlayController { throw new ControllerException(ErrorCode.ERROR100.getCode(), "设备[" + deviceId + "]不存在"); } - StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId); - if (streamInfo == null) { + InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId); + if (inviteInfo == null) { throw new ControllerException(ErrorCode.ERROR100.getCode(), "点播未找到"); } - - try { - logger.warn("[停止点播] {}/{}", device.getDeviceId(), channelId); - cmder.streamByeCmd(device, channelId, streamInfo.getStream(), null, null); - } catch (InvalidArgumentException | SipException | ParseException | SsrcTransactionNotFoundException e) { - logger.error("[命令发送失败] 停止点播， 发送BYE: {}", e.getMessage()); - throw new ControllerException(ErrorCode.ERROR100.getCode(), "命令发送失败: " + e.getMessage()); + if (InviteSessionStatus.ok == inviteInfo.getStatus()) { + try { + logger.warn("[停止点播] {}/{}", device.getDeviceId(), channelId); + cmder.streamByeCmd(device, channelId, inviteInfo.getStream(), null, null); + } catch (InvalidArgumentException | SipException | ParseException | SsrcTransactionNotFoundException e) { + logger.error("[命令发送失败] 停止点播， 发送BYE: {}", e.getMessage()); + throw new ControllerException(ErrorCode.ERROR100.getCode(), "命令发送失败: " + e.getMessage()); + } } - redisCatchStorage.stopPlay(streamInfo); + inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId); - storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId()); + storager.stopPlay(deviceId, channelId); JSONObject json = new JSONObject(); json.put("deviceId", deviceId); json.put("channelId", channelId); @@ -178,15 +189,14 @@ public class PlayController { @Parameter(name = "streamId", description = "视频流ID", required = true) @PostMapping("/convert/{streamId}") public JSONObject playConvert(@PathVariable String streamId) { - StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(streamId); - if (streamInfo == null) { - streamInfo = redisCatchStorage.queryPlayback(null, null, streamId, null); - } - if (streamInfo == null) { +// StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(streamId); + + InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(null, streamId); + if (inviteInfo == null || inviteInfo.getStreamInfo() == null) { logger.warn("视频转码API调用失败！, 视频流已经停止!"); throw new ControllerException(ErrorCode.ERROR100.getCode(), "未找到视频流信息, 视频流可能已经停止"); } - MediaServerItem mediaInfo = mediaServerService.getOne(streamInfo.getMediaServerId()); + MediaServerItem mediaInfo = mediaServerService.getOne(inviteInfo.getStreamInfo().getMediaServerId()); JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(mediaInfo, streamId); if (!rtpInfo.getBoolean("exist")) { logger.warn("视频转码API调用失败！, 视频流已停止推流!"); diff --git a/src/main/java/com/genersoft/iot/vmp/web/gb28181/ApiStreamController.java b/src/main/java/com/genersoft/iot/vmp/web/gb28181/ApiStreamController.java index 72a1b5da..5d04f107 100644 --- a/src/main/java/com/genersoft/iot/vmp/web/gb28181/ApiStreamController.java +++ b/src/main/java/com/genersoft/iot/vmp/web/gb28181/ApiStreamController.java @@ -1,7 +1,8 @@ package com.genersoft.iot.vmp.web.gb28181; import com.alibaba.fastjson2.JSONObject; -import com.genersoft.iot.vmp.common.StreamInfo; +import com.genersoft.iot.vmp.common.InviteInfo; +import com.genersoft.iot.vmp.common.InviteSessionType; import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException; import com.genersoft.iot.vmp.gb28181.bean.Device; @@ -9,13 +10,18 @@ import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel; import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; import com.genersoft.iot.vmp.service.IDeviceService; +import com.genersoft.iot.vmp.service.IInviteStreamService; import com.genersoft.iot.vmp.service.IPlayService; +import com.genersoft.iot.vmp.service.bean.InviteErrorCode; 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.web.bind.annotation.*; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestParam; +import org.springframework.web.bind.annotation.ResponseBody; +import org.springframework.web.bind.annotation.RestController; import org.springframework.web.context.request.async.DeferredResult; import javax.sip.InvalidArgumentException; @@ -45,6 +51,9 @@ public class ApiStreamController { @Autowired private IRedisCatchStorage redisCatchStorage; + @Autowired + private IInviteStreamService inviteStreamService; + @Autowired private IDeviceService deviceService; @@ -111,46 +120,96 @@ public class ApiStreamController { return resultDeferredResult; } MediaServerItem newMediaServerItem = playService.getNewMediaServerItem(device); - playService.play(newMediaServerItem, serial, code, (mediaServerItem, response)->{ - StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(serial, code); - JSONObject result = new JSONObject(); - result.put("StreamID", streamInfo.getStream()); - result.put("DeviceID", device.getDeviceId()); - result.put("ChannelID", code); - result.put("ChannelName", deviceChannel.getName()); - result.put("ChannelCustomName", ""); - result.put("FLV", streamInfo.getFlv().getUrl()); - result.put("WS_FLV", streamInfo.getWs_flv().getUrl()); - result.put("RTMP", streamInfo.getRtmp().getUrl()); - result.put("HLS", streamInfo.getHls().getUrl()); - result.put("RTSP", streamInfo.getRtsp().getUrl()); - result.put("WEBRTC", streamInfo.getRtc().getUrl()); - result.put("CDN", ""); - result.put("SnapURL", ""); - result.put("Transport", device.getTransport()); - result.put("StartAt", ""); - result.put("Duration", ""); - result.put("SourceVideoCodecName", ""); - result.put("SourceVideoWidth", ""); - result.put("SourceVideoHeight", ""); - result.put("SourceVideoFrameRate", ""); - result.put("SourceAudioCodecName", ""); - result.put("SourceAudioSampleRate", ""); - result.put("AudioEnable", ""); - result.put("Ondemand", ""); - result.put("InBytes", ""); - result.put("InBitRate", ""); - result.put("OutBytes", ""); - result.put("NumOutputs", ""); - result.put("CascadeSize", ""); - result.put("RelaySize", ""); - result.put("ChannelPTZType", "0"); - resultDeferredResult.setResult(result); - }, (eventResult) -> { - JSONObject result = new JSONObject(); - result.put("error", "channel[ " + code + " ] " + eventResult.msg); - resultDeferredResult.setResult(result); - }, null); +// playService.play(newMediaServerItem, serial, code, (mediaServerItem, response)->{ +// InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, serial, code); +// if (inviteInfo != null && inviteInfo.getStreamInfo() != null) { +// JSONObject result = new JSONObject(); +// result.put("StreamID", inviteInfo.getStreamInfo().getStream()); +// result.put("DeviceID", device.getDeviceId()); +// result.put("ChannelID", code); +// result.put("ChannelName", deviceChannel.getName()); +// result.put("ChannelCustomName", ""); +// result.put("FLV", inviteInfo.getStreamInfo().getFlv().getUrl()); +// result.put("WS_FLV", inviteInfo.getStreamInfo().getWs_flv().getUrl()); +// result.put("RTMP", inviteInfo.getStreamInfo().getRtmp().getUrl()); +// result.put("HLS", inviteInfo.getStreamInfo().getHls().getUrl()); +// result.put("RTSP", inviteInfo.getStreamInfo().getRtsp().getUrl()); +// result.put("WEBRTC", inviteInfo.getStreamInfo().getRtc().getUrl()); +// result.put("CDN", ""); +// result.put("SnapURL", ""); +// result.put("Transport", device.getTransport()); +// result.put("StartAt", ""); +// result.put("Duration", ""); +// result.put("SourceVideoCodecName", ""); +// result.put("SourceVideoWidth", ""); +// result.put("SourceVideoHeight", ""); +// result.put("SourceVideoFrameRate", ""); +// result.put("SourceAudioCodecName", ""); +// result.put("SourceAudioSampleRate", ""); +// result.put("AudioEnable", ""); +// result.put("Ondemand", ""); +// result.put("InBytes", ""); +// result.put("InBitRate", ""); +// result.put("OutBytes", ""); +// result.put("NumOutputs", ""); +// result.put("CascadeSize", ""); +// result.put("RelaySize", ""); +// result.put("ChannelPTZType", "0"); +// resultDeferredResult.setResult(result); +// } +// +// }, (eventResult) -> { +// JSONObject result = new JSONObject(); +// result.put("error", "channel[ " + code + " ] " + eventResult.msg); +// resultDeferredResult.setResult(result); +// }, null); + + + playService.play(newMediaServerItem, serial, code, (errorCode, msg, data) -> { + if (errorCode == InviteErrorCode.SUCCESS.getCode()) { + InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, serial, code); + if (inviteInfo != null && inviteInfo.getStreamInfo() != null) { + JSONObject result = new JSONObject(); + result.put("StreamID", inviteInfo.getStreamInfo().getStream()); + result.put("DeviceID", device.getDeviceId()); + result.put("ChannelID", code); + result.put("ChannelName", deviceChannel.getName()); + result.put("ChannelCustomName", ""); + result.put("FLV", inviteInfo.getStreamInfo().getFlv().getUrl()); + result.put("WS_FLV", inviteInfo.getStreamInfo().getWs_flv().getUrl()); + result.put("RTMP", inviteInfo.getStreamInfo().getRtmp().getUrl()); + result.put("HLS", inviteInfo.getStreamInfo().getHls().getUrl()); + result.put("RTSP", inviteInfo.getStreamInfo().getRtsp().getUrl()); + result.put("WEBRTC", inviteInfo.getStreamInfo().getRtc().getUrl()); + result.put("CDN", ""); + result.put("SnapURL", ""); + result.put("Transport", device.getTransport()); + result.put("StartAt", ""); + result.put("Duration", ""); + result.put("SourceVideoCodecName", ""); + result.put("SourceVideoWidth", ""); + result.put("SourceVideoHeight", ""); + result.put("SourceVideoFrameRate", ""); + result.put("SourceAudioCodecName", ""); + result.put("SourceAudioSampleRate", ""); + result.put("AudioEnable", ""); + result.put("Ondemand", ""); + result.put("InBytes", ""); + result.put("InBitRate", ""); + result.put("OutBytes", ""); + result.put("NumOutputs", ""); + result.put("CascadeSize", ""); + result.put("RelaySize", ""); + result.put("ChannelPTZType", "0"); + resultDeferredResult.setResult(result); + } + }else { + JSONObject result = new JSONObject(); + result.put("error", "channel[ " + code + " ] " + msg); + resultDeferredResult.setResult(result); + } + }); + return resultDeferredResult; } @@ -171,8 +230,8 @@ public class ApiStreamController { ){ - StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(serial, code); - if (streamInfo == null) { + InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, serial, code); + if (inviteInfo == null) { JSONObject result = new JSONObject(); result.put("error","未找到流信息"); return result; @@ -184,14 +243,14 @@ public class ApiStreamController { return result; } try { - cmder.streamByeCmd(device, code, streamInfo.getStream(), null); + cmder.streamByeCmd(device, code, inviteInfo.getStream(), null); } catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) { JSONObject result = new JSONObject(); result.put("error","发送BYE失败：" + e.getMessage()); return result; } - redisCatchStorage.stopPlay(streamInfo); - storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId()); + inviteStreamService.removeInviteInfo(inviteInfo); + storager.stopPlay(inviteInfo.getDeviceId(), inviteInfo.getChannelId()); return null; }