From 764d04b497356ba6bcbb75fd42b51eca750f7223 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期三, 29 五月 2024 15:02:51 +0800
Subject: [PATCH] 调整上级观看消息的发送
---
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java | 130 ++++++++++++++++++++----------------------
1 files changed, 62 insertions(+), 68 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 090de96..de1a929 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,13 +19,19 @@
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.MediaInfo;
+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.dto.StreamProxyItem;
+import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
+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;
-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.*;
import com.genersoft.iot.vmp.service.bean.ErrorCallback;
import com.genersoft.iot.vmp.service.bean.InviteErrorCode;
import com.genersoft.iot.vmp.service.bean.MessageForPushChannel;
@@ -55,6 +62,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 +115,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 +196,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;
// 涓嶆槸閫氶亾鍙兘鏄洿鎾祦
@@ -328,8 +335,12 @@
return;
}
String username = sdp.getOrigin().getUsername();
- String addressStr = sdp.getConnection().getAddress();
-
+ String addressStr;
+ if(StringUtils.isEmpty(platform.getSendStreamIp())){
+ addressStr = sdp.getConnection().getAddress();
+ }else {
+ addressStr = platform.getSendStreamIp();
+ }
Device device = null;
// 閫氳繃 channel 鍜� gbStream 鏄惁涓簄ull 鍊煎垽鏂潵婧愭槸鐩存挱娴佸悎閫傚浗鏍�
@@ -372,8 +383,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 +407,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 +464,22 @@
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, sendRtpItem, 5);
+ redisCatchStorage.sendPlatformStartPlayMsg(sendRtpItem, platform);
+ }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 +490,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.getHasAudio(), false, device.getStreamModeForParam());
sendRtpItem.setStream(stream);
// 鍐欏叆redis锛� 瓒呮椂鏃跺洖澶�
redisCatchStorage.updateSendRTPSever(sendRtpItem);
@@ -528,7 +520,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.getHasAudio(), false, device.getStreamModeForParam());
sendRtpItem.setStream(ssrcInfo.getStream());
// 鍐欏叆redis锛� 瓒呮椂鏃跺洖澶�
redisCatchStorage.updateSendRTPSever(sendRtpItem);
@@ -601,13 +593,12 @@
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.setServerId(pushListItem.getServerId());
sendRtpItem.setMediaServerId(pushListItem.getMediaServerId());
- StreamPushItem transform = streamPushService.transform(pushListItem);
- transform.setSelf(userSetting.getServerId().equals(pushListItem.getSeverId()));
+ pushListItem.setSelf(userSetting.getServerId().equals(pushListItem.getServerId()));
redisCatchStorage.updateSendRTPSever(sendRtpItem);
// 寮�濮嬫帹娴�
sendPushStream(sendRtpItem, mediaServerItem, platform, request);
@@ -664,9 +655,11 @@
/**
* 瀹夋帓鎺ㄦ祦
*/
- private void sendProxyStream(SendRtpItem sendRtpItem, MediaServerItem mediaServerItem, ParentPlatform platform, SIPRequest request) {
- Boolean streamReady = zlmServerFactory.isStreamReady(mediaServerItem, sendRtpItem.getApp(), sendRtpItem.getStream());
- if (streamReady != null && streamReady) {
+ private void sendProxyStream(SendRtpItem sendRtpItem, MediaServer mediaServerItem, ParentPlatform platform, SIPRequest request) {
+ MediaInfo mediaInfo = mediaServerService.getMediaInfo(mediaServerItem, sendRtpItem.getApp(), sendRtpItem.getStream());
+
+ if (mediaInfo != null) {
+
// 鑷钩鍙板唴瀹�
int localPort = sendRtpPortManager.getNextPort(mediaServerItem);
if (localPort == 0) {
@@ -691,11 +684,11 @@
}
}
- private void sendPushStream(SendRtpItem sendRtpItem, MediaServerItem mediaServerItem, ParentPlatform platform, SIPRequest request) {
+ private void sendPushStream(SendRtpItem sendRtpItem, MediaServer mediaServerItem, ParentPlatform platform, SIPRequest request) {
// 鎺ㄦ祦
if (sendRtpItem.getServerId().equals(userSetting.getServerId())) {
- Boolean streamReady = zlmServerFactory.isStreamReady(mediaServerItem, sendRtpItem.getApp(), sendRtpItem.getStream());
- if (streamReady != null && streamReady) {
+ MediaInfo mediaInfo = mediaServerService.getMediaInfo(mediaServerItem, sendRtpItem.getApp(), sendRtpItem.getStream());
+ if (mediaInfo != null ) {
// 鑷钩鍙板唴瀹�
int localPort = sendRtpPortManager.getNextPort(mediaServerItem);
if (localPort == 0) {
@@ -732,20 +725,19 @@
/**
* 閫氱煡娴佷笂绾�
*/
- private void notifyProxyStreamOnline(SendRtpItem sendRtpItem, MediaServerItem mediaServerItem, ParentPlatform platform, SIPRequest request) {
+ private void notifyProxyStreamOnline(SendRtpItem sendRtpItem, MediaServer mediaServerItem, ParentPlatform platform, SIPRequest request) {
// TODO 鎺у埗鍚敤浠ヤ娇璁惧涓婄嚎
logger.info("[ app={}, stream={} ]閫氶亾鏈帹娴侊紝鍚敤娴佸悗寮�濮嬫帹娴�", sendRtpItem.getApp(), sendRtpItem.getStream());
// 鐩戝惉娴佷笂绾�
- HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed(sendRtpItem.getApp(), sendRtpItem.getStream(), true, "rtsp", mediaServerItem.getId());
- zlmHttpHookSubscribe.addSubscribe(hookSubscribe, (mediaServerItemInUSe, hookParam) -> {
- OnStreamChangedHookParam streamChangedHookParam = (OnStreamChangedHookParam)hookParam;
- logger.info("[涓婄骇鐐规挱]鎷夋祦浠ｇ悊宸茬粡灏辩华锛� {}/{}", streamChangedHookParam.getApp(), streamChangedHookParam.getStream());
+ Hook hook = Hook.getInstance(HookType.on_media_arrival, sendRtpItem.getApp(), sendRtpItem.getStream(), mediaServerItem.getId());
+ hookSubscribe.addSubscribe(hook, (hookData)->{
+ logger.info("[涓婄骇鐐规挱]鎷夋祦浠ｇ悊宸茬粡灏辩华锛� {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream());
dynamicTask.stop(sendRtpItem.getCallId());
sendProxyStream(sendRtpItem, mediaServerItem, platform, request);
});
dynamicTask.startDelay(sendRtpItem.getCallId(), () -> {
logger.info("[ app={}, stream={} ] 绛夊緟鎷夋祦浠ｇ悊娴佽秴鏃�", sendRtpItem.getApp(), sendRtpItem.getStream());
- zlmHttpHookSubscribe.removeSubscribe(hookSubscribe);
+ hookSubscribe.removeSubscribe(hook);
}, userSetting.getPlatformPlayTimeout());
boolean start = streamProxyService.start(sendRtpItem.getApp(), sendRtpItem.getStream());
if (!start) {
@@ -754,7 +746,7 @@
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[鍛戒护鍙戦�佸け璐 invite 閫氶亾鏈帹娴�: {}", e.getMessage());
}
- zlmHttpHookSubscribe.removeSubscribe(hookSubscribe);
+ hookSubscribe.removeSubscribe(hook);
dynamicTask.stop(sendRtpItem.getCallId());
}
}
@@ -762,7 +754,7 @@
/**
* 閫氱煡娴佷笂绾�
*/
- private void notifyPushStreamOnline(SendRtpItem sendRtpItem, MediaServerItem mediaServerItem, ParentPlatform platform, SIPRequest request) {
+ private void notifyPushStreamOnline(SendRtpItem sendRtpItem, MediaServer mediaServerItem, ParentPlatform platform, SIPRequest request) {
// 鍙戦�乺edis娑堟伅浠ヤ娇璁惧涓婄嚎锛屾祦涓婄嚎鍚庤
logger.info("[ app={}, stream={} ]閫氶亾鏈帹娴侊紝鍙戦�乺edis淇℃伅鎺у埗璁惧寮�濮嬫帹娴�", sendRtpItem.getApp(), sendRtpItem.getStream());
MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(1,
@@ -780,7 +772,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 +823,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);
+ redisRpcService.removeCallback(key);
try {
responseAck(request, Response.TEMPORARILY_UNAVAILABLE, response.getMsg());
} catch (SipException | InvalidArgumentException | ParseException e) {
@@ -1010,7 +1004,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 +1019,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 +1053,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 +1081,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