From d41d6b34af2485198ed01e1888db1571e4da1a6a Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期二, 23 四月 2024 20:59:20 +0800
Subject: [PATCH] Merge branch 'refs/heads/2.7.0'
---
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java | 92 ++++++++++++++++++++++------------------------
1 files changed, 44 insertions(+), 48 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
index ff84fc4..f1f277f 100755
--- 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
@@ -8,6 +8,7 @@
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.UserSetting;
+import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager;
import com.genersoft.iot.vmp.gb28181.session.SSRCFactory;
@@ -18,6 +19,18 @@
import com.genersoft.iot.vmp.gb28181.transmit.event.request.ISIPRequestProcessor;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
+import com.genersoft.iot.vmp.media.bean.MediaServer;
+import com.genersoft.iot.vmp.media.event.hook.Hook;
+import com.genersoft.iot.vmp.media.event.hook.HookSubscribe;
+import com.genersoft.iot.vmp.media.event.hook.HookType;
+import com.genersoft.iot.vmp.media.service.IMediaServerService;
+import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager;
+import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
+import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
+import com.genersoft.iot.vmp.service.IInviteStreamService;
+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.media.zlm.SendRtpPortManager;
import com.genersoft.iot.vmp.service.redisMsg.IRedisRpcService;
import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory;
@@ -55,6 +68,9 @@
import javax.sip.message.Response;
import java.text.ParseException;
import java.time.Instant;
+import java.util.Map;
+import java.util.Random;
+import java.util.Vector;
import java.util.*;
/**
@@ -105,13 +121,10 @@
private AudioBroadcastManager audioBroadcastManager;
@Autowired
- private ZLMServerFactory zlmServerFactory;
-
- @Autowired
private IMediaServerService mediaServerService;
@Autowired
- private ZlmHttpHookSubscribe zlmHttpHookSubscribe;
+ private HookSubscribe hookSubscribe;
@Autowired
private SIPProcessorObserver sipProcessorObserver;
@@ -189,7 +202,7 @@
GbStream gbStream = storager.queryStreamInParentPlatform(requesterId, channelId);
PlatformCatalog catalog = storager.getCatalog(requesterId, channelId);
- MediaServerItem mediaServerItem = null;
+ MediaServer mediaServerItem = null;
StreamPushItem streamPushItem = null;
StreamProxyItem proxyByAppAndStream = null;
// 涓嶆槸閫氶亾鍙兘鏄洿鎾祦
@@ -372,8 +385,9 @@
} else {
streamTypeStr = "UDP";
}
- logger.info("[涓婄骇Invite] {}, 骞冲彴锛歿}锛� 閫氶亾锛歿}, 鏀舵祦鍦板潃锛歿}:{}锛屾敹娴佹柟寮忥細{}, ssrc锛歿}", sessionName, username, channelId, addressStr, port, streamTypeStr, ssrc);
- SendRtpItem sendRtpItem = zlmServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
+ logger.info("[涓婄骇Invite] {}, 骞冲彴锛歿}锛� 閫氶亾锛歿}, 鏀舵祦鍦板潃锛歿}:{}锛屾敹娴佹柟寮忥細{}, ssrc锛歿}",
+ sessionName, username, channelId, addressStr, port, streamTypeStr, ssrc);
+ SendRtpItem sendRtpItem = mediaServerService.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
device.getDeviceId(), channelId, mediaTransmissionTCP, platform.isRtcp());
if (tcpActive != null) {
@@ -395,7 +409,7 @@
Long finalStopTime = stopTime;
ErrorCallback<Object> hookEvent = (code, msg, data) -> {
StreamInfo streamInfo = (StreamInfo)data;
- MediaServerItem mediaServerItemInUSe = mediaServerService.getOne(streamInfo.getMediaServerId());
+ MediaServer mediaServerItemInUSe = mediaServerService.getOne(streamInfo.getMediaServerId());
logger.info("[涓婄骇Invite]涓嬬骇宸茬粡寮�濮嬫帹娴併�� 鍥炲200OK(SDP)锛� {}/{}", streamInfo.getApp(), streamInfo.getStream());
// * 0 绛夊緟璁惧鎺ㄦ祦涓婃潵
// * 1 涓嬬骇宸茬粡鎺ㄦ祦锛岀瓑寰呬笂绾у钩鍙板洖澶峚ck
@@ -452,42 +466,21 @@
responseSdpAck(request, content.toString(), platform);
// tcp涓诲姩妯″紡锛屽洖澶峴dp鍚庡紑鍚洃鍚�
if (sendRtpItem.isTcpActive()) {
- MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
- Map<String, Object> param = new HashMap<>(12);
- param.put("vhost","__defaultVhost__");
- param.put("app",sendRtpItem.getApp());
- param.put("stream",sendRtpItem.getStream());
- param.put("ssrc", sendRtpItem.getSsrc());
- if (!sendRtpItem.isTcpActive()) {
- param.put("dst_url",sendRtpItem.getIp());
- param.put("dst_port", sendRtpItem.getPort());
- }
- String is_Udp = sendRtpItem.isTcp() ? "0" : "1";
- param.put("is_udp", is_Udp);
- param.put("src_port", localPort);
- param.put("pt", sendRtpItem.getPt());
- param.put("use_ps", sendRtpItem.isUsePs() ? "1" : "0");
- param.put("only_audio", sendRtpItem.isOnlyAudio() ? "1" : "0");
- if (!sendRtpItem.isTcp()) {
- // 寮�鍚痳tcp淇濇椿
- param.put("udp_rtcp_timeout", sendRtpItem.isRtcp()? "1":"0");
- }
- JSONObject startSendRtpStreamResult = zlmServerFactory.startSendRtpPassive(mediaInfo, param);
- if (startSendRtpStreamResult != null) {
- startSendRtpStreamHand(evt, sendRtpItem, null, startSendRtpStreamResult, param, callIdHeader);
- }
+ MediaServer mediaServer = mediaServerService.getOne(sendRtpItem.getMediaServerId());
+ try {
+ mediaServerService.startSendRtpPassive(mediaServer, platform, sendRtpItem, 5);
+ }catch (ControllerException e) {}
}
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 鍥炲SdpAck", e);
}
};
ErrorCallback<Object> errorEvent = ((statusCode, msg, data) -> {
+ logger.info("[涓婄骇Invite] {}, 澶辫触, 骞冲彴锛歿}锛� 閫氶亾锛歿}, code锛� {}锛� msg锛泏}", sessionName, username, channelId, statusCode, msg);
// 鏈煡閿欒銆傜洿鎺ヨ浆鍙戣澶囩偣鎾殑閿欒
try {
- if (statusCode > 0) {
- Response response = getMessageFactory().createResponse(statusCode, evt.getRequest());
- sipSender.transmitRequest(request.getLocalAddress().getHostAddress(), response);
- }
+ Response response = getMessageFactory().createResponse(statusCode, evt.getRequest());
+ sipSender.transmitRequest(request.getLocalAddress().getHostAddress(), response);
} catch (ParseException | SipException e) {
logger.error("鏈鐞嗙殑寮傚父 ", e);
}
@@ -498,7 +491,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, false, device.getStreamModeForParam());
+ SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, stream, null, device.isSsrcCheck(), true, 0,false,!channel.isHasAudio(), false, device.getStreamModeForParam());
sendRtpItem.setStream(stream);
// 鍐欏叆redis锛� 瓒呮椂鏃跺洖澶�
redisCatchStorage.updateSendRTPSever(sendRtpItem);
@@ -528,7 +521,7 @@
}
sendRtpItem.setPlayType(InviteStreamType.DOWNLOAD);
- SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, null, null, device.isSsrcCheck(), true, 0, false, false, device.getStreamModeForParam());
+ SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, null, null, device.isSsrcCheck(), true, 0, false,!channel.isHasAudio(), false, device.getStreamModeForParam());
sendRtpItem.setStream(ssrcInfo.getStream());
// 鍐欏叆redis锛� 瓒呮椂鏃跺洖澶�
redisCatchStorage.updateSendRTPSever(sendRtpItem);
@@ -601,7 +594,7 @@
sendRtpItem.setPlayType(InviteStreamType.PUSH);
if (streamPushItem != null) {
// 浠巖edis鏌ヨ鏄惁姝ｅ湪鎺ユ敹杩欎釜鎺ㄦ祦
- OnStreamChangedHookParam pushListItem = redisCatchStorage.getPushListItem(gbStream.getApp(), gbStream.getStream());
+ StreamPushItem pushListItem = redisCatchStorage.getPushListItem(gbStream.getApp(), gbStream.getStream());
if (pushListItem != null) {
sendRtpItem.setServerId(pushListItem.getSeverId());
sendRtpItem.setMediaServerId(pushListItem.getMediaServerId());
@@ -667,6 +660,7 @@
private void sendProxyStream(SendRtpItem sendRtpItem, MediaServerItem mediaServerItem, ParentPlatform platform, SIPRequest request) {
Boolean streamReady = zlmServerFactory.isStreamReady(mediaServerItem, sendRtpItem.getApp(), sendRtpItem.getStream());
if (streamReady != null && streamReady) {
+
// 鑷钩鍙板唴瀹�
int localPort = sendRtpPortManager.getNextPort(mediaServerItem);
if (localPort == 0) {
@@ -683,7 +677,7 @@
sendRtpItem.setStatus(1);
sendRtpItem.setLocalIp(mediaServerItem.getSdpIp());
- SIPResponse response = sendStreamAck(request, sendRtpItem, platform);
+ SIPResponse response = sendStreamAck(mediaServer, request, sendRtpItem, platform, evt);
if (response != null) {
sendRtpItem.setToTag(response.getToTag());
}
@@ -771,7 +765,7 @@
redisCatchStorage.sendStreamPushRequestedMsg(messageForPushChannel);
// 璁剧疆瓒呮椂
dynamicTask.startDelay(sendRtpItem.getCallId(), () -> {
- redisRpcService.stopWaitePushStreamOnline(sendRtpItem.getRedisKey(), sendRtpItem);
+ redisRpcService.stopWaitePushStreamOnline(sendRtpItem);
logger.info("[ app={}, stream={} ] 绛夊緟璁惧寮�濮嬫帹娴佽秴鏃�", sendRtpItem.getApp(), sendRtpItem.getStream());
try {
responseAck(request, Response.REQUEST_TIMEOUT); // 瓒呮椂
@@ -780,7 +774,7 @@
}
}, userSetting.getPlatformPlayTimeout());
//
- redisRpcService.waitePushStreamOnline(sendRtpItem, (sendRtpItemKey) -> {
+ long key = redisRpcService.waitePushStreamOnline(sendRtpItem, (sendRtpItemKey) -> {
dynamicTask.stop(sendRtpItem.getCallId());
if (sendRtpItemKey == null) {
logger.warn("[绾ц仈鐐规挱] 绛夊緟鎺ㄦ祦寰楀埌缁撴灉鏈┖锛� {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream());
@@ -831,10 +825,12 @@
}
});
// 娣诲姞鍥炲鐨勬嫆缁濇垨鑰呴敊璇殑閫氱煡
+ // redis娑堟伅渚嬪锛� PUBLISH VM_MSG_STREAM_PUSH_RESPONSE '{"code":1,"msg":"澶辫触","app":"1","stream":"2"}'
redisPushStreamResponseListener.addEvent(sendRtpItem.getApp(), sendRtpItem.getStream(), response -> {
if (response.getCode() != 0) {
dynamicTask.stop(sendRtpItem.getCallId());
- redisRpcService.stopWaitePushStreamOnline(sendRtpItem.getRedisKey(), sendRtpItem);
+ redisRpcService.stopWaitePushStreamOnline(sendRtpItem);
+ redisRpcService.removeCallback(key);
try {
responseAck(request, Response.TEMPORARILY_UNAVAILABLE, response.getMsg());
} catch (SipException | InvalidArgumentException | ParseException e) {
@@ -864,7 +860,7 @@
redisCatchStorage.updateSendRTPSever(sendRtpItem);
}
- public SIPResponse sendStreamAck(SIPRequest request, SendRtpItem sendRtpItem, ParentPlatform platform) {
+ public SIPResponse sendStreamAck(MediaServerItem mediaServerItem, SIPRequest request, SendRtpItem sendRtpItem, ParentPlatform platform, RequestEvent evt) {
String sdpIp = sendRtpItem.getLocalIp();
if (!ObjectUtils.isEmpty(platform.getSendStreamIp())) {
@@ -1010,7 +1006,7 @@
logger.info("璁惧{}璇锋眰璇煶娴侊紝鍦板潃锛歿}:{}锛宻src锛歿}, {}", requesterId, addressStr, port, gb28181Sdp.getSsrc(),
mediaTransmissionTCP ? (tcpActive ? "TCP涓诲姩" : "TCP琚姩") : "UDP");
- MediaServerItem mediaServerItem = broadcastCatch.getMediaServerItem();
+ MediaServer mediaServerItem = broadcastCatch.getMediaServerItem();
if (mediaServerItem == null) {
logger.warn("鏈壘鍒拌闊冲枈璇濅娇鐢ㄧ殑zlm");
try {
@@ -1025,7 +1021,7 @@
mediaTransmissionTCP ? (tcpActive ? "TCP涓诲姩" : "TCP琚姩") : "UDP", sdp.getSessionName().getValue());
CallIdHeader callIdHeader = (CallIdHeader) request.getHeader(CallIdHeader.NAME);
- SendRtpItem sendRtpItem = zlmServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, gb28181Sdp.getSsrc(), requesterId,
+ SendRtpItem sendRtpItem = mediaServerService.createSendRtpItem(mediaServerItem, addressStr, port, gb28181Sdp.getSsrc(), requesterId,
device.getDeviceId(), broadcastCatch.getChannelId(),
mediaTransmissionTCP, false);
@@ -1059,7 +1055,7 @@
redisCatchStorage.updateSendRTPSever(sendRtpItem);
- Boolean streamReady = zlmServerFactory.isStreamReady(mediaServerItem, broadcastCatch.getApp(), broadcastCatch.getStream());
+ Boolean streamReady = mediaServerService.isStreamReady(mediaServerItem, broadcastCatch.getApp(), broadcastCatch.getStream());
if (streamReady) {
sendOk(device, sendRtpItem, sdp, request, mediaServerItem, mediaTransmissionTCP, gb28181Sdp.getSsrc());
} else {
@@ -1087,7 +1083,7 @@
}
}
- SIPResponse sendOk(Device device, SendRtpItem sendRtpItem, SessionDescription sdp, SIPRequest request, MediaServerItem mediaServerItem, boolean mediaTransmissionTCP, String ssrc) {
+ SIPResponse sendOk(Device device, SendRtpItem sendRtpItem, SessionDescription sdp, SIPRequest request, MediaServer mediaServerItem, boolean mediaTransmissionTCP, String ssrc) {
SIPResponse sipResponse = null;
try {
sendRtpItem.setStatus(2);
--
Gitblit v1.8.0