From cb5afce55eec2badda5090c27564e43f201ac9af Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期二, 12 三月 2024 23:55:00 +0800
Subject: [PATCH] 优化语音对讲定时任务的删除
---
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java | 97 ++++++++++++++++++++++++++++++------------------
1 files changed, 60 insertions(+), 37 deletions(-)
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
old mode 100644
new mode 100755
index ff855de..8084435
--- 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
@@ -2,6 +2,7 @@
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
+import com.genersoft.iot.vmp.common.InviteSessionType;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.conf.DynamicTask;
@@ -10,6 +11,7 @@
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager;
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.ISIPCommanderForPlatform;
@@ -21,10 +23,7 @@
import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.dto.*;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
-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.*;
import com.genersoft.iot.vmp.service.bean.ErrorCallback;
import com.genersoft.iot.vmp.service.bean.InviteErrorCode;
import com.genersoft.iot.vmp.service.bean.MessageForPushChannel;
@@ -36,8 +35,10 @@
import com.genersoft.iot.vmp.utils.DateUtil;
import gov.nist.javax.sdp.TimeDescriptionImpl;
import gov.nist.javax.sdp.fields.TimeField;
+import gov.nist.javax.sdp.fields.URIField;
import gov.nist.javax.sip.message.SIPRequest;
import gov.nist.javax.sip.message.SIPResponse;
+import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
@@ -84,6 +85,9 @@
private IRedisCatchStorage redisCatchStorage;
@Autowired
+ private IInviteStreamService inviteStreamService;
+
+ @Autowired
private SSRCFactory ssrcFactory;
@Autowired
@@ -126,6 +130,9 @@
@Autowired
private RedisGbPlayMsgListener redisGbPlayMsgListener;
+ @Autowired
+ private VideoStreamSessionManager streamSession;
+
@Override
public void afterPropertiesSet() throws Exception {
@@ -142,12 +149,25 @@
public void process(RequestEvent evt) {
// Invite Request娑堟伅瀹炵幇锛屾娑堟伅涓�鑸负绾ц仈娑堟伅锛屼笂绾х粰涓嬬骇鍙戦�佽姹傝棰戞寚浠�
try {
- SIPRequest request = (SIPRequest) evt.getRequest();
- String channelId = SipUtils.getChannelIdFromRequest(request);
+ SIPRequest request = (SIPRequest)evt.getRequest();
+ String channelIdFromSub = SipUtils.getChannelIdFromRequest(request);
+
+ // 瑙ｆ瀽sdp娑堟伅, 浣跨敤jainsip 鑷甫鐨剆dp瑙ｆ瀽鏂瑰紡
+ String contentString = new String(request.getRawContent());
+ Gb28181Sdp gb28181Sdp = SipUtils.parseSDP(contentString);
+ SessionDescription sdp = gb28181Sdp.getBaseSdb();
+ String sessionName = sdp.getSessionName().getValue();
+ String channelIdFromSdp = null;
+ if(StringUtils.equalsIgnoreCase("Playback", sessionName)){
+ URIField uriField = (URIField)sdp.getURI();
+ channelIdFromSdp = uriField.getURI().split(":")[0];
+ }
+ final String channelId = StringUtils.isNotBlank(channelIdFromSdp) ? channelIdFromSdp : channelIdFromSub;
+
String requesterId = SipUtils.getUserIdFromFromHeader(request);
CallIdHeader callIdHeader = (CallIdHeader) request.getHeader(CallIdHeader.NAME);
if (requesterId == null || channelId == null) {
- logger.info("鏃犳硶浠嶧romHeader鐨凙ddress涓幏鍙栧埌骞冲彴id锛岃繑鍥�400");
+ logger.info("鏃犳硶浠庤姹備腑鑾峰彇鍒板钩鍙癷d锛岃繑鍥�400");
// 鍙傛暟涓嶅叏锛� 鍙�400锛岃姹傞敊璇�
try {
responseAck(request, Response.BAD_REQUEST);
@@ -157,6 +177,8 @@
return;
}
+ logger.info("[INVITE] requesterId: {}, callId: {}, 鏉ヨ嚜锛歿}锛歿}",
+ requesterId, callIdHeader.getCallId(), request.getRemoteAddress(), request.getRemotePort());
// 鏌ヨ璇锋眰鏄惁鏉ヨ嚜涓婄骇骞冲彴\璁惧
ParentPlatform platform = storager.queryParentPlatByServerGBId(requesterId);
@@ -250,12 +272,6 @@
}
return;
}
- // 瑙ｆ瀽sdp娑堟伅, 浣跨敤jainsip 鑷甫鐨剆dp瑙ｆ瀽鏂瑰紡
- String contentString = new String(request.getRawContent());
-
- Gb28181Sdp gb28181Sdp = SipUtils.parseSDP(contentString);
- SessionDescription sdp = gb28181Sdp.getBaseSdb();
- String sessionName = sdp.getSessionName().getValue();
Long startTime = null;
Long stopTime = null;
@@ -404,7 +420,16 @@
// 闈炰弗鏍兼ā寮忕鍙ｄ笉缁熶竴, 澧炲姞鍏煎鎬э紝淇敼涓轰竴涓笉涓�0鐨勭鍙�
localPort = new Random().nextInt(65535) + 1;
}
- content.append("m=video " + localPort + " RTP/AVP 96\r\n");
+ if (sendRtpItem.isTcp()) {
+ content.append("m=video " + localPort + " TCP/RTP/AVP 96\r\n");
+ if (!sendRtpItem.isTcpActive()) {
+ content.append("a=setup:active\r\n");
+ } else {
+ content.append("a=setup:passive\r\n");
+ }
+ }else {
+ content.append("m=video " + localPort + " RTP/AVP 96\r\n");
+ }
content.append("a=sendonly\r\n");
content.append("a=rtpmap:96 PS/90000\r\n");
content.append("y=" + sendRtpItem.getSsrc() + "\r\n");
@@ -472,7 +497,7 @@
String startTimeStr = DateUtil.urlFormatter.format(start);
String endTimeStr = DateUtil.urlFormatter.format(end);
String stream = device.getDeviceId() + "_" + channelId + "_" + startTimeStr + "_" + endTimeStr;
- SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, stream, null, device.isSsrcCheck(), true, 0, false, device.getStreamModeForParam());
+ SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, stream, null, device.isSsrcCheck(), true, 0,false, false, device.getStreamModeForParam());
// 鍐欏叆redis锛� 瓒呮椂鏃跺洖澶�
redisCatchStorage.updateSendRTPSever(sendRtpItem);
playService.playBack(mediaServerItem, ssrcInfo, device.getDeviceId(), channelId, DateUtil.formatter.format(start),
@@ -518,10 +543,13 @@
errorEvent.run(code, msg, data);
}
});
- }else {
-
+ } else {
+ sendRtpItem.setPlayType(InviteStreamType.PLAY);
+ String streamId = String.format("%s_%s", device.getDeviceId(), channelId);
+ sendRtpItem.setStream(streamId);
+ redisCatchStorage.updateSendRTPSever(sendRtpItem);
SSRCInfo ssrcInfo = playService.play(mediaServerItem, device.getDeviceId(), channelId, ssrc, ((code, msg, data) -> {
- if (code == InviteErrorCode.SUCCESS.getCode()){
+ 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);
@@ -531,9 +559,6 @@
errorEvent.run(code, msg, data);
}
}));
- sendRtpItem.setPlayType(InviteStreamType.PLAY);
- String streamId = String.format("%s_%s", device.getDeviceId(), channelId);
- sendRtpItem.setStream(streamId);
sendRtpItem.setSsrc(ssrcInfo.getSsrc());
redisCatchStorage.updateSendRTPSever(sendRtpItem);
@@ -716,8 +741,6 @@
zlmHttpHookSubscribe.removeSubscribe(hookSubscribe);
dynamicTask.stop(callIdHeader.getCallId());
}
-
-
} else if ("push".equals(gbStream.getStreamType())) {
if (!platform.isStartOfflinePush()) {
// 骞冲彴璁剧疆涓叧闂簡鎷夎捣绂荤嚎鐨勬帹娴佸垯鐩存帴鍥炲
@@ -740,13 +763,10 @@
dynamicTask.startDelay(callIdHeader.getCallId(), () -> {
logger.info("[ app={}, stream={} ] 绛夊緟璁惧寮�濮嬫帹娴佽秴鏃�", gbStream.getApp(), gbStream.getStream());
try {
+ redisPushStreamResponseListener.removeEvent(gbStream.getApp(), gbStream.getStream());
mediaListManager.removedChannelOnlineEventLister(gbStream.getApp(), gbStream.getStream());
responseAck(request, Response.REQUEST_TIMEOUT); // 瓒呮椂
- } catch (SipException e) {
- logger.error("鏈鐞嗙殑寮傚父 ", e);
- } catch (InvalidArgumentException e) {
- logger.error("鏈鐞嗙殑寮傚父 ", e);
- } catch (ParseException e) {
+ } catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("鏈鐞嗙殑寮傚父 ", e);
}
}, userSetting.getPlatformPlayTimeout());
@@ -757,6 +777,7 @@
// 娣诲姞鍦ㄦ湰鏈轰笂绾跨殑閫氱煡
mediaListManager.addChannelOnlineEventLister(gbStream.getApp(), gbStream.getStream(), (app, stream, serverId) -> {
dynamicTask.stop(callIdHeader.getCallId());
+ redisPushStreamResponseListener.removeEvent(gbStream.getApp(), gbStream.getStream());
if (serverId.equals(userSetting.getServerId())) {
SendRtpItem sendRtpItem = zlmServerFactory.createSendRtpItem(mediaServerItem, addressStr, finalPort, ssrc, requesterId,
app, stream, channelId, mediaTransmissionTCP, platform.isRtcp());
@@ -821,7 +842,7 @@
// 鍙戦�乺edis娑堟伅
redisGbPlayMsgListener.sendMsg(streamPushItem.getServerId(), streamPushItem.getMediaServerId(),
streamPushItem.getApp(), streamPushItem.getStream(), addressStr, port, ssrc, requesterId,
- channelId, mediaTransmissionTCP, platform.isRtcp(), null, responseSendItemMsg -> {
+ channelId, mediaTransmissionTCP, platform.isRtcp(),platform.getName(), responseSendItemMsg -> {
SendRtpItem sendRtpItem = responseSendItemMsg.getSendRtpItem();
if (sendRtpItem == null || responseSendItemMsg.getMediaServerItem() == null) {
logger.warn("鏈嶅姟鍣ㄧ鍙ｈ祫婧愪笉瓒�");
@@ -956,7 +977,10 @@
}
if (device != null) {
logger.info("鏀跺埌璁惧" + requesterId + "鐨勮闊冲箍鎾璉nvite璇锋眰");
- String key = VideoManagerConstants.BROADCAST_WAITE_INVITE + device.getDeviceId() + broadcastCatch.getChannelId();
+ String key = VideoManagerConstants.BROADCAST_WAITE_INVITE + device.getDeviceId();
+ if (!SipUtils.isFrontEnd(device.getDeviceId())) {
+ key += broadcastCatch.getChannelId();
+ }
dynamicTask.stop(key);
try {
responseAck(request, Response.TRYING);
@@ -966,8 +990,6 @@
return;
}
String contentString = new String(request.getRawContent());
- // jainSip涓嶆敮鎸亂=瀛楁锛� 绉婚櫎绉婚櫎浠ヨВ鏋愩��
- String ssrc = "0000000404";
try {
Gb28181Sdp gb28181Sdp = SipUtils.parseSDP(contentString);
@@ -1015,7 +1037,7 @@
return;
}
String addressStr = sdp.getOrigin().getAddress();
- logger.info("璁惧{}璇锋眰璇煶娴侊紝鍦板潃锛歿}:{}锛宻src锛歿}, {}", requesterId, addressStr, port, ssrc,
+ logger.info("璁惧{}璇锋眰璇煶娴侊紝鍦板潃锛歿}:{}锛宻src锛歿}, {}", requesterId, addressStr, port, gb28181Sdp.getSsrc(),
mediaTransmissionTCP ? (tcpActive ? "TCP涓诲姩" : "TCP琚姩") : "UDP");
MediaServerItem mediaServerItem = broadcastCatch.getMediaServerItem();
@@ -1029,11 +1051,11 @@
}
return;
}
- logger.info("璁惧{}璇锋眰璇煶娴侊紝 鏀舵祦鍦板潃锛歿}:{}锛宻src锛歿}, {}, 瀵硅鏂瑰紡锛歿}", requesterId, addressStr, port, ssrc,
+ logger.info("璁惧{}璇锋眰璇煶娴侊紝 鏀舵祦鍦板潃锛歿}:{}锛宻src锛歿}, {}, 瀵硅鏂瑰紡锛歿}", requesterId, addressStr, port, gb28181Sdp.getSsrc(),
mediaTransmissionTCP ? (tcpActive ? "TCP涓诲姩" : "TCP琚姩") : "UDP", sdp.getSessionName().getValue());
CallIdHeader callIdHeader = (CallIdHeader) request.getHeader(CallIdHeader.NAME);
- SendRtpItem sendRtpItem = zlmServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
+ SendRtpItem sendRtpItem = zlmServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, gb28181Sdp.getSsrc(), requesterId,
device.getDeviceId(), broadcastCatch.getChannelId(),
mediaTransmissionTCP, false);
@@ -1069,7 +1091,7 @@
Boolean streamReady = zlmServerFactory.isStreamReady(mediaServerItem, broadcastCatch.getApp(), broadcastCatch.getStream());
if (streamReady) {
- sendOk(device, sendRtpItem, sdp, request, mediaServerItem, mediaTransmissionTCP, ssrc);
+ sendOk(device, sendRtpItem, sdp, request, mediaServerItem, mediaTransmissionTCP, gb28181Sdp.getSsrc());
} else {
logger.warn("[璇煶閫氳瘽]锛� 鏈彂鐜板緟鎺ㄩ�佺殑娴�,app={},stream={}", broadcastCatch.getApp(), broadcastCatch.getStream());
try {
@@ -1139,9 +1161,10 @@
audioBroadcastCatch.setStatus(AudioBroadcastCatchStatus.Ok);
audioBroadcastCatch.setSipTransactionInfoByRequset(sipResponse);
audioBroadcastManager.update(audioBroadcastCatch);
-
+ streamSession.put(device.getDeviceId(), sendRtpItem.getChannelId(), request.getCallIdHeader().getCallId(), sendRtpItem.getStream(), sendRtpItem.getSsrc(), sendRtpItem.getMediaServerId(), sipResponse, InviteSessionType.BROADCAST);
// 寮�鍚彂娴侊紝澶у崕鍦ㄦ敹鍒�200OK鍚庡氨浼氬紑濮嬪缓绔嬭繛鎺�
if (!device.isBroadcastPushAfterAck()) {
+ logger.info("[璇煶鍠婅瘽] 鍥炲200OK鍚庡彂鐜� BroadcastPushAfterAck涓篎alse锛岀幇鍦ㄥ紑濮嬫帹娴�");
playService.startPushStream(sendRtpItem, sipResponse, parentPlatform, request.getCallIdHeader());
}
--
Gitblit v1.8.0