From cb16cabb64e99d7f505822a49665725ff5b61ff6 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期一, 15 八月 2022 15:37:13 +0800
Subject: [PATCH] Merge branch 'wvp-28181-2.0'
---
src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java | 242 +++++++++++++++++++++++++++++++++++++++--------
1 files changed, 199 insertions(+), 43 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 811507d..b615d33 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
@@ -1,18 +1,40 @@
package com.genersoft.iot.vmp.service.impl;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.util.*;
+
+import javax.sip.ResponseEvent;
+
+import com.genersoft.iot.vmp.gb28181.bean.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.stereotype.Service;
+import org.springframework.web.context.request.async.DeferredResult;
+
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.common.StreamInfo;
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.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
+import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager;
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.ZLMRTPServerFactory;
+import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory;
+import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange;
+import com.genersoft.iot.vmp.media.zlm.dto.HookType;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.media.zlm.AssistRESTfulUtils;
import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
@@ -27,8 +49,13 @@
import com.genersoft.iot.vmp.service.bean.SSRCInfo;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
+import com.genersoft.iot.vmp.utils.redis.RedisUtil;
+import com.genersoft.iot.vmp.vmanager.bean.AudioBroadcastResult;
+import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
+import com.genersoft.iot.vmp.vmanager.gb28181.play.bean.AudioBroadcastEvent;
import com.genersoft.iot.vmp.vmanager.gb28181.play.bean.PlayResult;
+
import gov.nist.javax.sip.stack.SIPDialog;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -40,10 +67,14 @@
import org.springframework.web.context.request.async.DeferredResult;
import javax.sip.ResponseEvent;
+import javax.sip.SipException;
import java.io.FileNotFoundException;
import java.math.BigDecimal;
+import java.text.ParseException;
import java.math.RoundingMode;
import java.util.*;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
@SuppressWarnings(value = {"rawtypes", "unchecked"})
@Service
@@ -58,10 +89,16 @@
private SIPCommander cmder;
@Autowired
+ private AudioBroadcastManager audioBroadcastManager;
+
+ @Autowired
private SIPCommanderFroPlatform sipCommanderFroPlatform;
@Autowired
private IRedisCatchStorage redisCatchStorage;
+
+ @Autowired
+ private ZLMRTPServerFactory zlmrtpServerFactory;
@Autowired
private DeferredResultHolder resultHolder;
@@ -85,11 +122,18 @@
private UserSetting userSetting;
@Autowired
+ private SipConfig sipConfig;
+
+ @Autowired
private DynamicTask dynamicTask;
@Autowired
private ZLMHttpHookSubscribe subscribe;
+
+ @Qualifier("taskExecutor")
+ @Autowired
+ private ThreadPoolTaskExecutor taskExecutor;
@@ -122,22 +166,9 @@
result.onCompletion(()->{
// 鐐规挱缁撴潫鏃惰皟鐢ㄦ埅鍥炬帴鍙�
- // TODO 搴旇鍦ㄤ笂娴佹椂璋冪敤鏇村ソ锛岀粨鏉熶篃鍙兘鏄敊璇粨鏉�
- try {
- String classPath = ResourceUtils.getURL("classpath:").getPath();
- // 鍏煎鎵撳寘涓簀ar鐨刢lass璺緞
- if(classPath.contains("jar")) {
- classPath = classPath.substring(0, classPath.lastIndexOf("."));
- classPath = classPath.substring(0, classPath.lastIndexOf("/") + 1);
- }
- if (classPath.startsWith("file:")) {
- classPath = classPath.substring(classPath.indexOf(":") + 1);
- }
- String path = classPath + "static/static/snap/";
- // 鍏煎Windows绯荤粺璺緞锛堝幓闄ゅ墠闈㈢殑鈥�/鈥濓級
- if(System.getProperty("os.name").contains("indows")) {
- path = path.substring(1);
- }
+ taskExecutor.execute(()->{
+ // TODO 搴旇鍦ㄤ笂娴佹椂璋冪敤鏇村ソ锛岀粨鏉熶篃鍙兘鏄敊璇粨鏉�
+ String path = "snap";
String fileName = deviceId + "_" + channelId + ".jpg";
ResponseEntity responseEntity = (ResponseEntity)result.getResult();
if (responseEntity != null && responseEntity.getStatusCode() == HttpStatus.OK) {
@@ -151,9 +182,7 @@
zlmresTfulUtils.getSnap(mediaInfo, streamUrl, 15, 1, path, fileName);
}
}
- } catch (FileNotFoundException e) {
- e.printStackTrace();
- }
+ });
});
if (streamInfo != null) {
String streamId = streamInfo.getStream();
@@ -169,24 +198,30 @@
MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId);
JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(mediaInfo, streamId);
- if (rtpInfo != null && rtpInfo.getBoolean("exist")) {
+ if(rtpInfo.getInteger("code") == 0){
+ if (rtpInfo.getBoolean("exist")) {
- WVPResult wvpResult = new WVPResult();
- wvpResult.setCode(0);
- wvpResult.setMsg("success");
- wvpResult.setData(streamInfo);
- msg.setData(wvpResult);
+ WVPResult wvpResult = new WVPResult();
+ wvpResult.setCode(0);
+ wvpResult.setMsg("success");
+ wvpResult.setData(streamInfo);
+ msg.setData(wvpResult);
- resultHolder.invokeAllResult(msg);
- if (hookEvent != null) {
- hookEvent.response(mediaServerItem, JSONObject.parseObject(JSON.toJSONString(streamInfo)));
+ resultHolder.invokeAllResult(msg);
+ if (hookEvent != null) {
+ hookEvent.response(mediaServerItem, JSONObject.parseObject(JSON.toJSONString(streamInfo)));
+ }
+ }else {
+ redisCatchStorage.stopPlay(streamInfo);
+ storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
+ streamInfo = null;
}
}else {
+ //zlm杩炴帴澶辫触
redisCatchStorage.stopPlay(streamInfo);
storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
streamInfo = null;
}
-
}
if (streamInfo == null) {
String streamId = null;
@@ -261,6 +296,11 @@
}, userSetting.getPlayTimeout());
final String ssrc = ssrcInfo.getSsrc();
final String stream = ssrcInfo.getStream();
+ //绔彛鑾峰彇澶辫触鐨剆srcInfo 娌℃湁蹇呰鍙戦�佺偣鎾寚浠�
+ if(ssrcInfo.getPort() <= 0){
+ logger.info("[鐐规挱绔彛鍒嗛厤寮傚父]锛宒eviceId={},channelId={},ssrcInfo={}", device.getDeviceId(), channelId, ssrcInfo);
+ return;
+ }
cmder.playStreamCmd(mediaServerItem, ssrcInfo, device, channelId, (MediaServerItem mediaServerItemInuse, JSONObject response) -> {
logger.info("鏀跺埌璁㈤槄娑堟伅锛� " + response.toJSONString());
dynamicTask.stop(timeOutTaskKey);
@@ -300,16 +340,10 @@
// 鍗曠鍙ｆā寮弒treamId涔熸湁鍙樺寲锛岄渶瑕侀噸鏂拌缃洃鍚�
if (!mediaServerItem.isRtpEnable()) {
// 娣诲姞璁㈤槄
- JSONObject subscribeKey = new JSONObject();
- subscribeKey.put("app", "rtp");
- subscribeKey.put("stream", stream);
- subscribeKey.put("regist", true);
- subscribeKey.put("schema", "rtmp");
- subscribeKey.put("mediaServerId", mediaServerItem.getId());
- subscribe.removeSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed,subscribeKey);
- subscribeKey.put("stream", String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase());
- subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,
- (MediaServerItem mediaServerItemInUse, JSONObject response)->{
+ HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", stream, true, "rtmp", mediaServerItem.getId());
+ subscribe.removeSubscribe(hookSubscribe);
+ hookSubscribe.getContent().put("stream", String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase());
+ subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject response)->{
logger.info("[ZLM HOOK] ssrc淇鍚庢敹鍒拌闃呮秷鎭細 " + response.toJSONString());
dynamicTask.stop(timeOutTaskKey);
// hook鍝嶅簲
@@ -320,7 +354,7 @@
// 鍏抽棴rtp server
mediaServerService.closeRTPServer(device.getDeviceId(), channelId, finalSsrcInfo.getStream());
// 閲嶆柊寮�鍚痵src server
- mediaServerService.openRTPServer(mediaServerItem, finalSsrcInfo.getStream(), ssrcInResponse, device.isSsrcCheck(), false);
+ mediaServerService.openRTPServer(mediaServerItem, finalSsrcInfo.getStream(), ssrcInResponse, device.isSsrcCheck(), false, finalSsrcInfo.getPort());
}
}
@@ -624,7 +658,7 @@
public StreamInfo onPublishHandler(MediaServerItem mediaServerItem, JSONObject resonse, String deviceId, String channelId) {
String streamId = resonse.getString("stream");
JSONArray tracks = resonse.getJSONArray("tracks");
- StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(mediaServerItem,"rtp", streamId, tracks);
+ StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(mediaServerItem,"rtp", streamId, tracks, null);
streamInfo.setDeviceID(deviceId);
streamInfo.setChannelId(channelId);
return streamInfo;
@@ -655,7 +689,129 @@
}
@Override
+ public void audioBroadcast(Device device, String channelId, int timeout, AudioBroadcastEvent event) {
+ if (device == null || channelId == null) {
+ return;
+ }
+ DeviceChannel deviceChannel = storager.queryChannel(device.getDeviceId(), channelId);
+ if (deviceChannel == null) {
+ logger.warn("寮�鍚闊冲箍鎾殑鏃跺�欐湭鎵惧埌閫氶亾锛� {}", channelId);
+ event.call("寮�鍚闊冲箍鎾殑鏃跺�欐湭鎵惧埌閫氶亾");
+ return;
+ }
+ // 鏌ヨ閫氶亾浣跨敤鐘舵��
+ if (audioBroadcastManager.exit(device.getDeviceId(), channelId)) {
+ SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(device.getDeviceId(), channelId, null, null);
+ if (sendRtpItem != null && sendRtpItem.isOnlyAudio()) {
+ // 鏌ヨ娴佹槸鍚﹀瓨鍦紝涓嶅瓨鍦ㄥ垯璁や负鏄紓甯哥姸鎬�
+ MediaServerItem mediaServerItem = mediaServerService.getOne(sendRtpItem.getMediaServerId());
+ Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, sendRtpItem.getApp(), sendRtpItem.getStreamId());
+ if (streamReady) {
+ logger.warn("璇煶骞挎挱宸茬粡寮�鍚細 {}", channelId);
+ event.call("璇煶骞挎挱宸茬粡寮�鍚�");
+ return;
+ }else {
+ audioBroadcastManager.del(deviceChannel.getDeviceId(),channelId);
+ redisCatchStorage.deleteSendRTPServer(device.getDeviceId(), channelId, sendRtpItem.getCallId(), sendRtpItem.getStreamId());
+ }
+ }
+ }
+
+ // 鍙戦�侀�氱煡
+ cmder.audioBroadcastCmd(device, channelId, eventResultForOk -> {
+ // 鍙戦�佹垚鍔�
+ AudioBroadcastCatch audioBroadcastCatch = new AudioBroadcastCatch(device.getDeviceId(), channelId, AudioBroadcastCatchStatus.Ready);
+ audioBroadcastManager.add(audioBroadcastCatch);
+ }, eventResultForError -> {
+ // 鍙戦�佸け璐�
+ logger.error("璇煶骞挎挱鍙戦�佸け璐ワ細 {}:{}", channelId, eventResultForError.msg);
+ event.call("璇煶骞挎挱鍙戦�佸け璐�");
+ stopAudioBroadcast(device.getDeviceId(), channelId);
+ });
+ }
+
+ @Override
+ public void stopAudioBroadcast(String deviceId, String channelId){
+ AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(deviceId, channelId);
+ if (audioBroadcastCatch != null) {
+
+ try {
+ SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(deviceId, audioBroadcastCatch.getChannelId(), null, null);
+ if (sendRtpItem != null) {
+ redisCatchStorage.deleteSendRTPServer(deviceId, sendRtpItem.getChannelId(), null, null);
+ MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
+ Map<String, Object> param = new HashMap<>();
+ param.put("vhost", "__defaultVhost__");
+ param.put("app", sendRtpItem.getApp());
+ param.put("stream", sendRtpItem.getStreamId());
+ zlmresTfulUtils.stopSendRtp(mediaInfo, param);
+ // 绔嬪埢缁撴潫璁惧鐨勬帹娴侊紝绛夊緟鑷缁撴潫澶參
+ zlmresTfulUtils.closeStreams(mediaInfo, sendRtpItem.getApp(), sendRtpItem.getStreamId());
+ }
+ if (audioBroadcastCatch.getStatus() == AudioBroadcastCatchStatus.Ok) {
+ cmder.streamByeCmd(audioBroadcastCatch.getDialog(), audioBroadcastCatch.getRequest(), null);
+ }
+ audioBroadcastManager.del(deviceId, channelId);
+
+ } catch (SipException e) {
+ throw new RuntimeException(e);
+ } catch (ParseException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+
+ }
+
+ @Override
public void zlmServerOnline(String mediaServerId) {
- // 浼间箮娌″暐闇�瑕佸仛鐨�
+ // TODO 鏌ユ壘涔嬪墠鐨勭偣鎾紝娴佸鏋滀笉瀛樺湪鍒欑粰涓嬬骇鍙戦�乥ye
+// MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
+// zlmresTfulUtils.getMediaList(mediaServerItem, (mediaList ->{
+// Integer code = mediaList.getInteger("code");
+// if (code == 0) {
+// JSONArray data = mediaList.getJSONArray("data");
+// if (data == null || data.size() == 0) {
+// zlmServerOffline(mediaServerId);
+// }else {
+// Map<String, JSONObject> mediaListMap = new HashMap<>();
+// for (int i = 0; i < data.size(); i++) {
+// JSONObject json = data.getJSONObject(i);
+// String app = json.getString("app");
+// if ("rtp".equals(app)) {
+// String stream = json.getString("stream");
+// if (mediaListMap.get(stream) != null) {
+// continue;
+// }
+// mediaListMap.put(stream, json);
+// // 澶勭悊姝ｅ湪瑙傜湅鐨勫浗鏍囪澶�
+// List<SsrcTransaction> ssrcTransactions = streamSession.getSsrcTransactionForAll(null, null, null, stream);
+// if (ssrcTransactions.size() > 0) {
+// for (SsrcTransaction ssrcTransaction : ssrcTransactions) {
+// if(ssrcTransaction.getMediaServerId().equals(mediaServerId)) {
+// cmder.streamByeCmd(ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId(),
+// ssrcTransaction.getStream(), null);
+// }
+// }
+// }
+// }
+// }
+// if (mediaListMap.size() > 0 ) {
+// // 澶勭悊姝ｅ湪鍚戜笂鎺ㄦ祦鐨勪笂绾у钩鍙�
+// List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServer(null);
+// if (sendRtpItems.size() > 0) {
+// for (SendRtpItem sendRtpItem : sendRtpItems) {
+// if (sendRtpItem.getMediaServerId().equals(mediaServerId)) {
+// if (mediaListMap.get(sendRtpItem.getStreamId()) == null) {
+// ParentPlatform platform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
+// sipCommanderFroPlatform.streamByeCmd(platform, sendRtpItem.getCallId());
+// }
+// }
+// }
+// }
+// }
+// }
+// }
+// }));
}
}
--
Gitblit v1.8.0