XW-AIOT/wvp-platform
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复国标视频点播三种点播方式（自动点播，上级点播，接口点播）并发情况下失败的问题

Browse Source
This commit is contained in:
648540858
2023-05-06 17:40:57 +08:00
parent 9ccce01692
commit e2f9ee8f7b
26 changed files with 916 additions and 517 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
src/main/java/com/genersoft/iot/vmp/gb28181/bean/SsrcTransaction.java
View File

@@ -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;
}

9
src/main/java/com/genersoft/iot/vmp/gb28181/session/VideoStreamSessionManager.java
View File

@@ -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<Object, Object> 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);

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

@@ -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);
});
}

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

@@ -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());
}
}
}
}

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

@@ -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<Object> 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<Object> 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());