XW-AIOT/wvp-platform
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

合并主线

Browse Source
This commit is contained in:
648540858
2022-11-18 21:52:02 +08:00
parent f223aad705
commit ecf84bb0f1
28 changed files with 231 additions and 350 deletions
Download Patch File Download Diff File Expand all files Collapse all files

120
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
View File

@@ -9,21 +9,16 @@ import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
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.dto.HookType;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.StreamAuthorityInfo;
import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
import com.genersoft.iot.vmp.media.zlm.dto.hook.*;
import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.media.zlm.dto.*;
import com.genersoft.iot.vmp.service.*;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.vmanager.bean.AudioBroadcastResult;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -294,9 +289,9 @@ public class ZLMHttpHookListener {
JSONObject json = (JSONObject) JSON.toJSON(param);
taskExecutor.execute(()->{
taskExecutor.execute(()-> {
ZlmHttpHookSubscribe.Event subscribe = this.subscribe.sendNotify(HookType.on_stream_changed, json);
if (subscribe != null ) {
if (subscribe != null) {
MediaServerItem mediaInfo = mediaServerService.getOne(param.getMediaServerId());
if (mediaInfo != null) {
subscribe.response(mediaInfo, json);
@@ -312,15 +307,16 @@ public class ZLMHttpHookListener {
StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(param.getApp(), param.getStream());
if (streamAuthorityInfo == null) {
streamAuthorityInfo = StreamAuthorityInfo.getInstanceByHook(param);
}else {
} else {
streamAuthorityInfo.setOriginType(param.getOriginType());
streamAuthorityInfo.setOriginTypeStr(param.getOriginTypeStr());
}
redisCatchStorage.updateStreamAuthorityInfo(param.getApp(), param.getStream(), streamAuthorityInfo);
}
}else {
} else {
redisCatchStorage.removeStreamAuthorityInfo(param.getApp(), param.getStream());
}
});
if ("rtsp".equals(param.getSchema())){
logger.info("on_stream_changed：注册->{}, app->{}, stream->{}", param.isRegist(), param.getApp(), param.getStream());
@@ -329,12 +325,12 @@ public class ZLMHttpHookListener {
}else {
mediaServerService.removeCount(param.getMediaServerId());
}
if (item.getOriginType() == OriginType.PULL.ordinal()
|| item.getOriginType() == OriginType.FFMPEG_PULL.ordinal()) {
if (param.getOriginType() == OriginType.PULL.ordinal()
|| param.getOriginType() == OriginType.FFMPEG_PULL.ordinal()) {
// 设置拉流代理上线/离线
streamProxyService.updateStatus(param.isRegist(), app, param.getStream());
streamProxyService.updateStatus(param.isRegist(), param.getApp(), param.getStream());
}
if ("rtp".equals(app) && !regist ) {
if ("rtp".equals(param.getApp()) && !param.isRegist() ) {
StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(param.getStream());
if (streamInfo!=null){
redisCatchStorage.stopPlay(streamInfo);
@@ -346,47 +342,49 @@ public class ZLMHttpHookListener {
streamInfo.getStream(), null);
}
}
}else if ("broadcast".equals(app)){
}else if ("broadcast".equals(param.getApp())){
// 语音对讲推流 stream需要满足格式deviceId_channelId
if (regist && param.getStream().indexOf("_") > 0) {
if (param.isRegist() && param.getStream().indexOf("_") > 0) {
String[] streamArray = param.getStream().split("_");
if (streamArray.length == 2) {
String deviceId = streamArray[0];
String channelId = streamArray[1];
Device device = deviceService.queryDevice(deviceId);
Device device = deviceService.getDevice(deviceId);
if (device != null) {
DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
if (deviceChannel != null) {
if (audioBroadcastManager.exit(deviceId, channelId)) {
// 直接推流
SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(null, null, stream, null);
SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(null, null, param.getStream(), null);
if (sendRtpItem == null) {
// TODO 可能数据错误，重新开启语音通道
}else {
String is_Udp = sendRtpItem.isTcp() ? "0" : "1";
MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
logger.info("rtp/{}开始向上级推流, 目标={}:{}，SSRC={}", sendRtpItem.getStreamId(), sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc());
Map<String, Object> param = new HashMap<>(12);
param.put("vhost","__defaultVhost__");
param.put("app",sendRtpItem.getApp());
param.put("stream",sendRtpItem.getStreamId());
param.put("ssrc", sendRtpItem.getSsrc());
param.put("src_port", sendRtpItem.getLocalPort());
param.put("pt", sendRtpItem.getPt());
param.put("use_ps", sendRtpItem.isUsePs() ? "1" : "0");
param.put("only_audio", sendRtpItem.isOnlyAudio() ? "1" : "0");
Map<String, Object> sendParam = new HashMap<>(12);
sendParam.put("vhost","__defaultVhost__");
sendParam.put("app",sendRtpItem.getApp());
sendParam.put("stream",sendRtpItem.getStreamId());
sendParam.put("ssrc", sendRtpItem.getSsrc());
sendParam.put("src_port", sendRtpItem.getLocalPort());
sendParam.put("pt", sendRtpItem.getPt());
sendParam.put("use_ps", sendRtpItem.isUsePs() ? "1" : "0");
sendParam.put("only_audio", sendRtpItem.isOnlyAudio() ? "1" : "0");
JSONObject jsonObject;
if (sendRtpItem.isTcpActive()) {
jsonObject = zlmrtpServerFactory.startSendRtpPassive(mediaInfo, param);
jsonObject = zlmrtpServerFactory.startSendRtpPassive(mediaInfo, sendParam);
} else {
param.put("is_udp", is_Udp);
param.put("dst_url", sendRtpItem.getIp());
param.put("dst_port", sendRtpItem.getPort());
jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
sendParam.put("is_udp", is_Udp);
sendParam.put("dst_url", sendRtpItem.getIp());
sendParam.put("dst_port", sendRtpItem.getPort());
jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaInfo, sendParam);
}
if (jsonObject != null && jsonObject.getInteger("code") == 0) {
logger.info("[语音对讲] 自动推流成功, device: {}, channel: {}", deviceId, channelId);
}else {
logger.info("[语音对讲] 推流失败, 结果： {}", jsonObject);
}
}
@@ -406,43 +404,43 @@ public class ZLMHttpHookListener {
}
}
}else if ("talk".equals(app)){
}else if ("talk".equals(param.getApp())){
// 语音对讲推流 stream需要满足格式deviceId_channelId
if (regist && stream.indexOf("_") > 0) {
String[] streamArray = stream.split("_");
if (param.isRegist() && param.getStream().indexOf("_") > 0) {
String[] streamArray = param.getStream().split("_");
if (streamArray.length == 2) {
String deviceId = streamArray[0];
String channelId = streamArray[1];
Device device = deviceService.queryDevice(deviceId);
Device device = deviceService.getDevice(deviceId);
if (device != null) {
DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
if (deviceChannel != null) {
if (audioBroadcastManager.exit(deviceId, channelId)) {
// 直接推流
SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(null, null, stream, null);
SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(null, null, param.getStream(), null);
if (sendRtpItem == null) {
// TODO 可能数据错误，重新开启语音通道
}else {
MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
logger.info("rtp/{}开始向上级推流, 目标={}:{}，SSRC={}", sendRtpItem.getStreamId(), sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc());
Map<String, Object> param = new HashMap<>(12);
param.put("vhost","__defaultVhost__");
param.put("app",sendRtpItem.getApp());
param.put("stream",sendRtpItem.getStreamId());
param.put("ssrc", sendRtpItem.getSsrc());
param.put("src_port", sendRtpItem.getLocalPort());
param.put("pt", sendRtpItem.getPt());
param.put("use_ps", sendRtpItem.isUsePs() ? "1" : "0");
param.put("only_audio", sendRtpItem.isOnlyAudio() ? "1" : "0");
Map<String, Object> sendParam = new HashMap<>(12);
sendParam.put("vhost","__defaultVhost__");
sendParam.put("app",sendRtpItem.getApp());
sendParam.put("stream",sendRtpItem.getStreamId());
sendParam.put("ssrc", sendRtpItem.getSsrc());
sendParam.put("src_port", sendRtpItem.getLocalPort());
sendParam.put("pt", sendRtpItem.getPt());
sendParam.put("use_ps", sendRtpItem.isUsePs() ? "1" : "0");
sendParam.put("only_audio", sendRtpItem.isOnlyAudio() ? "1" : "0");
JSONObject jsonObject;
if (sendRtpItem.isTcpActive()) {
jsonObject = zlmrtpServerFactory.startSendRtpPassive(mediaInfo, param);
jsonObject = zlmrtpServerFactory.startSendRtpPassive(mediaInfo, sendParam);
} else {
param.put("is_udp", sendRtpItem.isTcp() ? "0" : "1");
param.put("dst_url", sendRtpItem.getIp());
param.put("dst_port", sendRtpItem.getPort());
jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
sendParam.put("is_udp", sendRtpItem.isTcp() ? "0" : "1");
sendParam.put("dst_url", sendRtpItem.getIp());
sendParam.put("dst_port", sendRtpItem.getPort());
jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaInfo, sendParam);
}
if (jsonObject != null && jsonObject.getInteger("code") == 0) {
logger.info("[语音对讲] 自动推流成功, device: {}, channel: {}", deviceId, channelId);
@@ -450,7 +448,7 @@ public class ZLMHttpHookListener {
}
}else {
// 开启语音对讲通道
MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
MediaServerItem mediaServerItem = mediaServerService.getOne(param.getMediaServerId());
playService.talk(mediaServerItem, device, channelId, (mediaServer, jsonObject)->{
System.out.println("开始推流");
}, eventResult -> {
@@ -466,9 +464,9 @@ public class ZLMHttpHookListener {
}
}else{
if (!"rtp".equals(app)){
String type = OriginType.values()[item.getOriginType()].getType();
MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
if (!"rtp".equals(param.getApp())){
String type = OriginType.values()[param.getOriginType()].getType();
MediaServerItem mediaServerItem = mediaServerService.getOne(param.getMediaServerId());
if (mediaServerItem != null){
if (param.isRegist()) {
@@ -478,7 +476,7 @@ public class ZLMHttpHookListener {
callId = streamAuthorityInfo.getCallId();
}
StreamInfo streamInfoByAppAndStream = mediaService.getStreamInfoByAppAndStream(mediaServerItem,
param.getApp(), param.getStream(), tracks, callId);
param.getApp(), param.getStream(), param.getTracks(), callId);
param.setStreamInfo(streamInfoByAppAndStream);
redisCatchStorage.addStream(mediaServerItem, type, param.getApp(), param.getStream(), param);
if (param.getOriginType() == OriginType.RTSP_PUSH.ordinal()
@@ -489,7 +487,8 @@ public class ZLMHttpHookListener {
}
}else {
// 兼容流注销时类型从redis记录获取
OnStreamChangedHookParam onStreamChangedHookParam = redisCatchStorage.getStreamInfo(param.getApp(), param.getStream(), param.getMediaServerId());
OnStreamChangedHookParam onStreamChangedHookParam = redisCatchStorage.getStreamInfo(
param.getApp(), param.getStream(), param.getMediaServerId());
if (onStreamChangedHookParam != null) {
type = OriginType.values()[onStreamChangedHookParam.getOriginType()].getType();
redisCatchStorage.removeStream(mediaServerItem.getId(), type, param.getApp(), param.getStream());
@@ -526,13 +525,13 @@ public class ZLMHttpHookListener {
if (platform != null) {
commanderFroPlatform.streamByeCmd(platform, sendRtpItem);
}else {
if ("talk".equals(app) && sendRtpItem.isOnlyAudio()) {
if ("talk".equals(param.getApp()) && sendRtpItem.isOnlyAudio()) {
AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
if (device != null && audioBroadcastCatch != null) {
// cmder.streamByeCmd(device, sendRtpItem.getChannelId(), audioBroadcastCatch.getSipTransactionInfo(), null);
}
}else {
cmder.streamByeCmd(device, sendRtpItem.getChannelId(), stream, sendRtpItem.getCallId());
cmder.streamByeCmd(device, sendRtpItem.getChannelId(), param.getStream(), sendRtpItem.getCallId());
}
}
@@ -575,6 +574,9 @@ public class ZLMHttpHookListener {
if (sendRtpItems.size() > 0) {
for (SendRtpItem sendRtpItem : sendRtpItems) {
ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
if (parentPlatform == null) {
continue;
}
try {
commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId());
} catch (SipException | InvalidArgumentException | ParseException e) {

7
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRESTfulUtils.java
View File

@@ -78,6 +78,7 @@ public class ZLMRESTfulUtils {
if (callback == null) {
try {
Response response = client.newCall(request).execute();
if (response.isSuccessful()) {
ResponseBody responseBody = response.body();
if (responseBody != null) {
@@ -85,6 +86,8 @@ public class ZLMRESTfulUtils {
responseJSON = JSON.parseObject(responseStr);
}
}else {
System.out.println( 2222);
System.out.println( response.code());
response.close();
Objects.requireNonNull(response.body()).close();
}
@@ -93,11 +96,11 @@ public class ZLMRESTfulUtils {
if(e instanceof SocketTimeoutException){
//读取超时超时异常
logger.error(String.format("读取ZLM数据失败: %s, %s", url, e.getMessage()));
logger.error(String.format("读取ZLM数据超时失败: %s, %s", url, e.getMessage()));
}
if(e instanceof ConnectException){
//判断连接异常，我这里是报Failed to connect to 10.7.5.144
logger.error(String.format("连接ZLM失败: %s, %s", url, e.getMessage()));
logger.error(String.format("连接ZLM连接失败: %s, %s", url, e.getMessage()));
}
}catch (Exception e){

23
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java
View File

@@ -5,9 +5,9 @@ import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.media.zlm.dto.MediaItem;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForRtpServerTimeout;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -346,23 +346,4 @@ public class ZLMRTPServerFactory {
return result;
}
public void closeAllSendRtpStream() {
}
public MediaItem getMediaInfo(MediaServerItem mediaServerItem, String app, String stream) {
JSONObject json = zlmresTfulUtils.getMediaList(mediaServerItem, app, stream);
MediaItem mediaItem = null;
if (json == null || json.getInteger("code") != 0) {
return null;
} else {
JSONArray data = json.getJSONArray("data");
if (data == null || data.size() == 0) {
return null;
}else {
mediaItem = JSONObject.toJavaObject(data.getJSONObject(0), MediaItem.class);
}
}
return mediaItem;
}
}

2
src/main/java/com/genersoft/iot/vmp/media/zlm/dto/HookSubscribeFactory.java
View File

@@ -37,7 +37,7 @@ public class HookSubscribeFactory {
public static HookSubscribeForStreamPush on_publish(String app, String stream, String scheam, String mediaServerId) {
HookSubscribeForStreamPush hookSubscribe = new HookSubscribeForStreamPush();
JSONObject subscribeKey = new com.alibaba.fastjson.JSONObject();
JSONObject subscribeKey = new JSONObject();
subscribeKey.put("app", app);
subscribeKey.put("stream", stream);
if (scheam != null) {

3
src/main/java/com/genersoft/iot/vmp/media/zlm/dto/HookSubscribeForStreamPush.java
View File

@@ -1,6 +1,7 @@
package com.genersoft.iot.vmp.media.zlm.dto;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson2.JSONObject;
import java.time.Instant;

17
src/main/java/com/genersoft/iot/vmp/media/zlm/dto/MediaServerItemLite.java
View File

@@ -1,12 +1,6 @@
package com.genersoft.iot.vmp.media.zlm.dto;
import com.genersoft.iot.vmp.gb28181.session.SsrcConfig;
import com.genersoft.iot.vmp.media.zlm.ZLMServerConfig;
import org.springframework.util.StringUtils;
import java.util.HashMap;
/**
* 精简的MediaServerItem信息，方便给前端返回数据
*/
@@ -38,8 +32,6 @@ public class MediaServerItemLite {
private String secret;
private int hookAliveInterval;
private int recordAssistPort;
@@ -58,7 +50,6 @@ public class MediaServerItemLite {
this.rtspPort = mediaServerItem.getRtspPort();
this.rtspSSLPort = mediaServerItem.getRtspSSLPort();
this.secret = mediaServerItem.getSecret();
this.hookAliveInterval = mediaServerItem.getHookAliveInterval();
this.recordAssistPort = mediaServerItem.getRecordAssistPort();
}
@@ -167,14 +158,6 @@ public class MediaServerItemLite {
this.secret = secret;
}
public int getHookAliveInterval() {
return hookAliveInterval;
}
public void setHookAliveInterval(int hookAliveInterval) {
this.hookAliveInterval = hookAliveInterval;
}
public int getRecordAssistPort() {
return recordAssistPort;
}