From bea63f67e75ea6c38d946c2ee463260fcf815f87 Mon Sep 17 00:00:00 2001
From: Fang <costa11@qq.com>
Date: 星期一, 07 三月 2022 14:21:29 +0800
Subject: [PATCH] Merge branch '648540858:wvp-28181-2.0' into wvp-28181-2.0
---
src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java | 156 ++++++++++++++++++++++++++++++++++++++++------------
1 files changed, 120 insertions(+), 36 deletions(-)
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 778bd67..2df78b7 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
@@ -5,13 +5,13 @@
import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.UserSetup;
-import com.genersoft.iot.vmp.gb28181.bean.Device;
-import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
+import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
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.ZLMHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
@@ -37,8 +37,7 @@
import org.springframework.web.context.request.async.DeferredResult;
import java.io.FileNotFoundException;
-import java.util.Objects;
-import java.util.UUID;
+import java.util.*;
@SuppressWarnings(value = {"rawtypes", "unchecked"})
@Service
@@ -51,6 +50,9 @@
@Autowired
private SIPCommander cmder;
+
+ @Autowired
+ private SIPCommanderFroPlatform sipCommanderFroPlatform;
@Autowired
private IRedisCatchStorage redisCatchStorage;
@@ -78,7 +80,9 @@
@Override
- public PlayResult play(MediaServerItem mediaServerItem, String deviceId, String channelId, ZLMHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent) {
+ public PlayResult play(MediaServerItem mediaServerItem, String deviceId, String channelId,
+ ZLMHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent,
+ Runnable timeoutCallback) {
PlayResult playResult = new PlayResult();
RequestMessage msg = new RequestMessage();
String key = DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId;
@@ -101,29 +105,10 @@
Device device = redisCatchStorage.getDevice(deviceId);
StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId);
playResult.setDevice(device);
- // 瓒呮椂澶勭悊
- result.onTimeout(()->{
- logger.warn(String.format("璁惧鐐规挱瓒呮椂锛宒eviceId锛�%s 锛宑hannelId锛�%s", deviceId, channelId));
- WVPResult wvpResult = new WVPResult();
- wvpResult.setCode(-1);
- SIPDialog dialog = streamSession.getDialogByStream(deviceId, channelId, streamInfo.getStream());
- if (dialog != null) {
- wvpResult.setMsg("鏀舵祦瓒呮椂锛岃绋嶅�欓噸璇�");
- }else {
- wvpResult.setMsg("鐐规挱瓒呮椂锛岃绋嶅�欓噸璇�");
- }
- msg.setData(wvpResult);
- // 鐐规挱瓒呮椂鍥炲BYE
- cmder.streamByeCmd(device.getDeviceId(), channelId, streamInfo.getStream());
- // 閲婃斁rtpserver
- mediaServerService.closeRTPServer(playResult.getDevice().getDeviceId(), channelId, streamInfo.getStream());
- // 鍥炲涔嬪墠鎵�鏈夌殑鐐规挱璇锋眰
- resultHolder.invokeAllResult(msg);
- // TODO 閲婃斁ssrc
- });
result.onCompletion(()->{
// 鐐规挱缁撴潫鏃惰皟鐢ㄦ埅鍥炬帴鍙�
+ // TODO 搴旇鍦ㄤ笂娴佹椂璋冪敤鏇村ソ锛岀粨鏉熶篃鍙兘鏄敊璇粨鏉�
try {
String classPath = ResourceUtils.getURL("classpath:").getPath();
// 鍏煎鎵撳寘涓簀ar鐨刢lass璺緞
@@ -161,31 +146,60 @@
if (mediaServerItem.isRtpEnable()) {
streamId = String.format("%s_%s", device.getDeviceId(), channelId);
}
-
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId);
+ // 瓒呮椂澶勭悊
+ Timer timer = new Timer();
+ timer.schedule(new TimerTask() {
+ @Override
+ public void run() {
+ logger.warn(String.format("璁惧鐐规挱瓒呮椂锛宒eviceId锛�%s 锛宑hannelId锛�%s", deviceId, channelId));
+ if (timeoutCallback != null) {
+ timeoutCallback.run();
+ }
+ WVPResult wvpResult = new WVPResult();
+ wvpResult.setCode(-1);
+ SIPDialog dialog = streamSession.getDialogByStream(deviceId, channelId, ssrcInfo.getStream());
+ if (dialog != null) {
+ wvpResult.setMsg("鏀舵祦瓒呮椂锛岃绋嶅�欓噸璇�");
+ // 鐐规挱瓒呮椂鍥炲BYE 鍚屾椂閲婃斁ssrc浠ュ強姝ゆ鐐规挱鐨勮祫婧�
+ cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream());
+ }else {
+ wvpResult.setMsg("鐐规挱瓒呮椂锛岃绋嶅�欓噸璇�");
+ mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
+ mediaServerService.closeRTPServer(deviceId, channelId, ssrcInfo.getStream());
+ streamSession.remove(deviceId, channelId, ssrcInfo.getStream());
+ }
+
+ msg.setData(wvpResult);
+
+ // 鍥炲涔嬪墠鎵�鏈夌殑鐐规挱璇锋眰
+ resultHolder.invokeAllResult(msg);
+ }
+ }, userSetup.getPlayTimeout());
// 鍙戦�佺偣鎾秷鎭�
cmder.playStreamCmd(mediaServerItem, ssrcInfo, device, channelId, (MediaServerItem mediaServerItemInUse, JSONObject response) -> {
logger.info("鏀跺埌璁㈤槄娑堟伅锛� " + response.toJSONString());
+ timer.cancel();
onPublishHandlerForPlay(mediaServerItemInUse, response, deviceId, channelId, uuid);
if (hookEvent != null) {
hookEvent.response(mediaServerItem, response);
}
}, (event) -> {
+ timer.cancel();
WVPResult wvpResult = new WVPResult();
wvpResult.setCode(-1);
// 鐐规挱杩斿洖sip閿欒
mediaServerService.closeRTPServer(playResult.getDevice().getDeviceId(), channelId, ssrcInfo.getStream());
// 閲婃斁ssrc
- mediaServerService.releaseSsrc(mediaServerItem, ssrcInfo.getSsrc());
+ mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
streamSession.remove(deviceId, channelId, ssrcInfo.getStream());
+
wvpResult.setMsg(String.format("鐐规挱澶辫触锛� 閿欒鐮侊細 %s, %s", event.statusCode, event.msg));
msg.setData(wvpResult);
resultHolder.invokeAllResult(msg);
if (errorEvent != null) {
errorEvent.response(event);
}
-
-
});
} else {
String streamId = streamInfo.getStream();
@@ -222,13 +236,41 @@
streamId2 = String.format("%s_%s", device.getDeviceId(), channelId);
}
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId2);
+ // 瓒呮椂澶勭悊
+ Timer timer = new Timer();
+ timer.schedule(new TimerTask() {
+ @Override
+ public void run() {
+ logger.warn(String.format("璁惧鐐规挱瓒呮椂锛宒eviceId锛�%s 锛宑hannelId锛�%s", deviceId, channelId));
+ if (timeoutCallback != null) {
+ timeoutCallback.run();
+ }
+ WVPResult wvpResult = new WVPResult();
+ wvpResult.setCode(-1);
+ SIPDialog dialog = streamSession.getDialogByStream(deviceId, channelId, ssrcInfo.getStream());
+ if (dialog != null) {
+ wvpResult.setMsg("鏀舵祦瓒呮椂锛岃绋嶅�欓噸璇�");
+ // 鐐规挱瓒呮椂鍥炲BYE 鍚屾椂閲婃斁ssrc浠ュ強姝ゆ鐐规挱鐨勮祫婧�
+ cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream());
+ }else {
+ wvpResult.setMsg("鐐规挱瓒呮椂锛岃绋嶅�欓噸璇�");
+ mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
+ mediaServerService.closeRTPServer(deviceId, channelId, ssrcInfo.getStream());
+ streamSession.remove(deviceId, channelId, ssrcInfo.getStream());
+ }
+
+ msg.setData(wvpResult);
+ // 鍥炲涔嬪墠鎵�鏈夌殑鐐规挱璇锋眰
+ resultHolder.invokeAllResult(msg);
+ }
+ }, userSetup.getPlayTimeout());
cmder.playStreamCmd(mediaServerItem, ssrcInfo, device, channelId, (MediaServerItem mediaServerItemInuse, JSONObject response) -> {
logger.info("鏀跺埌璁㈤槄娑堟伅锛� " + response.toJSONString());
onPublishHandlerForPlay(mediaServerItemInuse, response, deviceId, channelId, uuid);
}, (event) -> {
mediaServerService.closeRTPServer(playResult.getDevice().getDeviceId(), channelId, ssrcInfo.getStream());
// 閲婃斁ssrc
- mediaServerService.releaseSsrc(mediaServerItem, ssrcInfo.getSsrc());
+ mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
streamSession.remove(deviceId, channelId, ssrcInfo.getStream());
WVPResult wvpResult = new WVPResult();
wvpResult.setCode(-1);
@@ -306,14 +348,33 @@
msg.setId(uuid);
msg.setKey(key);
PlayBackResult<RequestMessage> playBackResult = new PlayBackResult<>();
- result.onTimeout(()->{
- msg.setData("鍥炴斁瓒呮椂");
- playBackResult.setCode(-1);
- playBackResult.setData(msg);
- callback.call(playBackResult);
- });
+
+ Timer timer = new Timer();
+ timer.schedule(new TimerTask() {
+ @Override
+ public void run() {
+ logger.warn(String.format("璁惧鍥炴斁瓒呮椂锛宒eviceId锛�%s 锛宑hannelId锛�%s", deviceId, channelId));
+ playBackResult.setCode(-1);
+ playBackResult.setData(msg);
+ callback.call(playBackResult);
+ SIPDialog dialog = streamSession.getDialogByStream(deviceId, channelId, ssrcInfo.getStream());
+ // 鐐规挱瓒呮椂鍥炲BYE 鍚屾椂閲婃斁ssrc浠ュ強姝ゆ鐐规挱鐨勮祫婧�
+ if (dialog != null) {
+ // 鐐规挱瓒呮椂鍥炲BYE 鍚屾椂閲婃斁ssrc浠ュ強姝ゆ鐐规挱鐨勮祫婧�
+ cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream());
+ }else {
+ mediaServerService.releaseSsrc(newMediaServerItem.getId(), ssrcInfo.getSsrc());
+ mediaServerService.closeRTPServer(deviceId, channelId, ssrcInfo.getStream());
+ streamSession.remove(deviceId, channelId, ssrcInfo.getStream());
+ }
+ cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream());
+ // 鍥炲涔嬪墠鎵�鏈夌殑鐐规挱璇锋眰
+ callback.call(playBackResult);
+ }
+ }, userSetup.getPlayTimeout());
cmder.playbackStreamCmd(newMediaServerItem, ssrcInfo, device, channelId, startTime, endTime, (MediaServerItem mediaServerItem, JSONObject response) -> {
logger.info("鏀跺埌璁㈤槄娑堟伅锛� " + response.toJSONString());
+ timer.cancel();
StreamInfo streamInfo = onPublishHandler(mediaServerItem, response, deviceId, channelId);
if (streamInfo == null) {
logger.warn("璁惧鍥炴斁API璋冪敤澶辫触锛�");
@@ -331,6 +392,7 @@
playBackResult.setResponse(response);
callback.call(playBackResult);
}, event -> {
+ timer.cancel();
msg.setData(String.format("鍥炴斁澶辫触锛� 閿欒鐮侊細 %s, %s", event.statusCode, event.msg));
playBackResult.setCode(-1);
playBackResult.setData(msg);
@@ -370,4 +432,26 @@
return streamInfo;
}
+ @Override
+ public void zlmServerOffline(String mediaServerId) {
+ // 澶勭悊姝ｅ湪鍚戜笂鎺ㄦ祦鐨勪笂绾у钩鍙�
+ List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServer(null);
+ if (sendRtpItems.size() > 0) {
+ for (SendRtpItem sendRtpItem : sendRtpItems) {
+ if (sendRtpItem.getMediaServerId().equals(mediaServerId)) {
+ ParentPlatform platform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
+ sipCommanderFroPlatform.streamByeCmd(platform, sendRtpItem.getCallId());
+ }
+ }
+ }
+ // 澶勭悊姝ｅ湪瑙傜湅鐨勫浗鏍囪澶�
+ List<SsrcTransaction> allSsrc = streamSession.getAllSsrc();
+ if (allSsrc.size() > 0) {
+ for (SsrcTransaction ssrcTransaction : allSsrc) {
+ if(ssrcTransaction.getMediaServerId().equals(mediaServerId)) {
+ cmder.streamByeCmd(ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId(), ssrcTransaction.getStream());
+ }
+ }
+ }
+ }
}
--
Gitblit v1.8.0