Thirdparty
/
wvp-GB28181-pro
mirror of https://github.com/648540858/wvp-GB28181-pro
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

合并主线

结构优化
648540858 2022-06-27 14:40:07 +08:00
parent 26739237e2
commit 1d1f174852
6 changed files with 286 additions and 280 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java
View File

@ -4,7 +4,6 @@ import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe; import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform; import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.SIPRequestHeaderPlarformProvider; import com.genersoft.iot.vmp.gb28181.transmit.cmd.SIPRequestHeaderPlarformProvider;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory; import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService; import com.genersoft.iot.vmp.service.IMediaServerService;
@ -26,12 +25,13 @@ import org.springframework.lang.Nullable;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils; import org.springframework.util.StringUtils;
import com.genersoft.iot.vmp.utils.DateUtil;
import javax.sip.*; import javax.sip.*;
import javax.sip.address.SipURI; import javax.sip.address.SipURI;
import javax.sip.header.*; import javax.sip.header.*;
import javax.sip.message.Request; import javax.sip.message.Request;
import java.lang.reflect.Field; import java.lang.reflect.Field;
import java.net.InetAddress;
import java.text.ParseException; import java.text.ParseException;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.HashSet; import java.util.HashSet;

57
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java
View File

@ -1,8 +1,6 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl; package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject; 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.DynamicTask;
import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager; import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager;
@ -22,7 +20,6 @@ import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import gov.nist.javax.sip.message.SIPRequest; import gov.nist.javax.sip.message.SIPRequest;
import gov.nist.javax.sip.stack.SIPDialog; import gov.nist.javax.sip.stack.SIPDialog;
import com.genersoft.iot.vmp.utils.SerializeUtils; import com.genersoft.iot.vmp.utils.SerializeUtils;
import org.ehcache.shadow.org.terracotta.offheapstore.storage.IntegerStorageEngine;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.InitializingBean;
@ -37,7 +34,6 @@ import javax.sip.address.SipURI;
import javax.sip.header.CallIdHeader; import javax.sip.header.CallIdHeader;
import javax.sip.header.FromHeader; import javax.sip.header.FromHeader;
import javax.sip.header.HeaderAddress; import javax.sip.header.HeaderAddress;
import javax.sip.header.ToHeader;
import java.text.ParseException; import java.text.ParseException;
import java.util.*; import java.util.*;
@ -99,28 +95,28 @@ public class AckRequestProcessor extends SIPRequestProcessorParent implements In
@Override @Override
public void process(RequestEvent evt) { public void process(RequestEvent evt) {
Dialog dialog = evt.getDialog(); Dialog dialog = evt.getDialog();
CallIdHeader callIdHeader = (CallIdHeader)evt.getRequest().getHeader(CallIdHeader.NAME); CallIdHeader callIdHeader = (CallIdHeader) evt.getRequest().getHeader(CallIdHeader.NAME);
if (dialog == null) { if (dialog == null) {
return; return;
} }
if (dialog.getState()== DialogState.CONFIRMED) { if (dialog.getState() == DialogState.CONFIRMED) {
String platformGbId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(FromHeader.NAME)).getAddress().getURI()).getUser(); String platformGbId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(FromHeader.NAME)).getAddress().getURI()).getUser();
logger.info("ACK请求： platformGbId->{}", platformGbId); logger.info("ACK请求： platformGbId->{}", platformGbId);
ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(platformGbId); ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(platformGbId);
// 取消设置的超时任务 // 取消设置的超时任务
dynamicTask.stop(callIdHeader.getCallId()); dynamicTask.stop(callIdHeader.getCallId());
// String channelId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(ToHeader.NAME)).getAddress().getURI()).getUser(); // String channelId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(ToHeader.NAME)).getAddress().getURI()).getUser();
SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(platformGbId, null, null, callIdHeader.getCallId()); SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(platformGbId, null, null, callIdHeader.getCallId());
String is_Udp = sendRtpItem.isTcp() ? "0" : "1"; String is_Udp = sendRtpItem.isTcp() ? "0" : "1";
MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId()); MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
logger.info("[收到ACK]，开始使用{}向上级推流 {}/{}->{}:{}({})", sendRtpItem.isTcp() ? "TCP" : "UDP", logger.info("[收到ACK]，开始使用{}向上级推流 {}/{}->{}:{}({})", sendRtpItem.isTcp() ? "TCP" : "UDP",
sendRtpItem.getApp(), sendRtpItem.getStreamId(), sendRtpItem.getApp(), sendRtpItem.getStreamId(),
sendRtpItem.getIp() ,sendRtpItem.getPort(), sendRtpItem.getIp(), sendRtpItem.getPort(),
sendRtpItem.getSsrc()); sendRtpItem.getSsrc());
Map<String, Object> param = new HashMap<>(); Map<String, Object> param = new HashMap<>();
param.put("vhost","__defaultVhost__"); param.put("vhost", "__defaultVhost__");
param.put("app",sendRtpItem.getApp()); param.put("app", sendRtpItem.getApp());
param.put("stream",sendRtpItem.getStreamId()); param.put("stream", sendRtpItem.getStreamId());
param.put("ssrc", sendRtpItem.getSsrc()); param.put("ssrc", sendRtpItem.getSsrc());
param.put("src_port", sendRtpItem.getLocalPort()); param.put("src_port", sendRtpItem.getLocalPort());
param.put("pt", sendRtpItem.getPt()); param.put("pt", sendRtpItem.getPt());
@ -129,9 +125,9 @@ public class AckRequestProcessor extends SIPRequestProcessorParent implements In
JSONObject jsonObject; JSONObject jsonObject;
if (sendRtpItem.isTcpActive()) { if (sendRtpItem.isTcpActive()) {
jsonObject = zlmrtpServerFactory.startSendRtpPassive(mediaInfo, param); jsonObject = zlmrtpServerFactory.startSendRtpPassive(mediaInfo, param);
}else { } else {
param.put("is_udp", is_Udp); param.put("is_udp", is_Udp);
param.put("dst_url",sendRtpItem.getIp()); param.put("dst_url", sendRtpItem.getIp());
param.put("dst_port", sendRtpItem.getPort()); param.put("dst_port", sendRtpItem.getPort());
jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param); jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
} }
@ -149,36 +145,37 @@ public class AckRequestProcessor extends SIPRequestProcessorParent implements In
String waiteStreamTimeoutTaskKey = "waite-stream-" + audioBroadcastCatch.getDeviceId() + audioBroadcastCatch.getChannelId(); String waiteStreamTimeoutTaskKey = "waite-stream-" + audioBroadcastCatch.getDeviceId() + audioBroadcastCatch.getChannelId();
dynamicTask.stop(waiteStreamTimeoutTaskKey); dynamicTask.stop(waiteStreamTimeoutTaskKey);
} }
logger.info("RTP推流成功[ {}/{} ]，{}->{}:{}, " ,param.get("app"), param.get("stream"), jsonObject.getString("local_port"), param.get("dst_url"), param.get("dst_port")); logger.info("RTP推流成功[ {}/{} ]，{}->{}:{}, ", param.get("app"), param.get("stream"), jsonObject.getString("local_port"), param.get("dst_url"), param.get("dst_port"));
} else { } else {
logger.error("RTP推流失败: {}, 参数：{}",jsonObject.getString("msg"),JSONObject.toJSON(param)); logger.error("RTP推流失败: {}, 参数：{}", jsonObject.getString("msg"), JSONObject.toJSON(param));
if (sendRtpItem.isOnlyAudio()) { if (sendRtpItem.isOnlyAudio()) {
// 语音对讲 // 语音对讲
try { try {
cmder.streamByeCmd((SIPDialog) evt.getDialog(), (SIPRequest)evt.getRequest(), null); cmder.streamByeCmd((SIPDialog) evt.getDialog(), (SIPRequest) evt.getRequest(), null);
} catch (SipException e) { } catch (SipException e) {
throw new RuntimeException(e); throw new RuntimeException(e);
} catch (ParseException e) { } catch (ParseException e) {
throw new RuntimeException(e); throw new RuntimeException(e);
} }
}else { } else {
// 向上级平台 // 向上级平台
commanderForPlatform.streamByeCmd(parentPlatform, callIdHeader.getCallId()); commanderForPlatform.streamByeCmd(parentPlatform, callIdHeader.getCallId());
} }
if (mediaInfo == null) { if (mediaInfo == null) {
RequestPushStreamMsg requestPushStreamMsg = RequestPushStreamMsg.getInstance( RequestPushStreamMsg requestPushStreamMsg = RequestPushStreamMsg.getInstance(
sendRtpItem.getMediaServerId(), sendRtpItem.getApp(), sendRtpItem.getStreamId(), sendRtpItem.getMediaServerId(), sendRtpItem.getApp(), sendRtpItem.getStreamId(),
sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc(), sendRtpItem.isTcp(), sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc(), sendRtpItem.isTcp(),
sendRtpItem.getLocalPort(), sendRtpItem.getPt(), sendRtpItem.isUsePs(), sendRtpItem.isOnlyAudio()); sendRtpItem.getLocalPort(), sendRtpItem.getPt(), sendRtpItem.isUsePs(), sendRtpItem.isOnlyAudio());
redisGbPlayMsgListener.sendMsgForStartSendRtpStream(sendRtpItem.getServerId(), requestPushStreamMsg, jsonObject->{ redisGbPlayMsgListener.sendMsgForStartSendRtpStream(sendRtpItem.getServerId(), requestPushStreamMsg, json -> {
startSendRtpStreamHand(evt, sendRtpItem, parentPlatform, jsonObject, param, callIdHeader); startSendRtpStreamHand(evt, sendRtpItem, parentPlatform, json, param, callIdHeader);
}); });
}else { } else {
JSONObject jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param); JSONObject startSendRtpStreamResult = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
startSendRtpStreamHand(evt, sendRtpItem, parentPlatform, jsonObject, param, callIdHeader); startSendRtpStreamHand(evt, sendRtpItem, parentPlatform, startSendRtpStreamResult, param, callIdHeader);
}
} }
} }
} }
private void startSendRtpStreamHand(RequestEvent evt, SendRtpItem sendRtpItem, ParentPlatform parentPlatform, private void startSendRtpStreamHand(RequestEvent evt, SendRtpItem sendRtpItem, ParentPlatform parentPlatform,

483
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java
View File

@ -19,15 +19,14 @@ import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.ISIPRequestProcessor; 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.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.gb28181.utils.SipUtils; import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
import com.genersoft.iot.vmp.gb28181.utils.XmlUtil;
import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe; import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager; import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager;
import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils; import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory; import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.MediaItem; import com.genersoft.iot.vmp.media.zlm.dto.MediaItem;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItemLite; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItemLite;
import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
import com.genersoft.iot.vmp.service.IMediaServerService; import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IPlayService; import com.genersoft.iot.vmp.service.IPlayService;
import com.genersoft.iot.vmp.service.IStreamPushService; import com.genersoft.iot.vmp.service.IStreamPushService;
@ -58,8 +57,6 @@ import javax.sip.message.Request;
import javax.sip.message.Response; import javax.sip.message.Response;
import java.text.ParseException; import java.text.ParseException;
import java.time.Instant; import java.time.Instant;
import java.util.Date;
import java.util.List;
import java.util.Vector; import java.util.Vector;
/** /**
@ -109,9 +106,6 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
@Autowired @Autowired
private ZLMRESTfulUtils zlmresTfulUtils; private ZLMRESTfulUtils zlmresTfulUtils;
@Autowired
private IMediaServerService mediaServerService;
@Autowired @Autowired
private SIPProcessorObserver sipProcessorObserver; private SIPProcessorObserver sipProcessorObserver;
@ -171,7 +165,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
// 查询请求是否来自上级平台\设备 // 查询请求是否来自上级平台\设备
ParentPlatform platform = storager.queryParentPlatByServerGBId(requesterId); ParentPlatform platform = storager.queryParentPlatByServerGBId(requesterId);
if (platform == null) { if (platform == null) {
inviteFromDeviceHandle(evt, requesterId); inviteFromDeviceHandle(evt, requesterId, channelId);
} else { } else {
// 查询平台下是否有该通道 // 查询平台下是否有该通道
DeviceChannel channel = storager.queryChannelInParentPlatform(requesterId, channelId); DeviceChannel channel = storager.queryChannelInParentPlatform(requesterId, channelId);
@ -724,10 +718,16 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
} }
} }
public void inviteFromDeviceHandle(RequestEvent evt, String requesterId) throws InvalidArgumentException, ParseException, SipException, SdpException { public void inviteFromDeviceHandle(RequestEvent evt, String requesterId, String channelId1) throws InvalidArgumentException, ParseException, SipException, SdpException {
// 非上级平台请求，查询是否设备请求（通常为接收语音广播的设备） // 非上级平台请求，查询是否设备请求（通常为接收语音广播的设备）
Device device = redisCatchStorage.getDevice(requesterId); Device device = redisCatchStorage.getDevice(requesterId);
AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(requesterId, channelId1);
if (audioBroadcastCatch == null) {
logger.warn("来自设备的Invite请求非语音广播，已忽略");
responseAck(evt, Response.FORBIDDEN);
return;
}
Request request = evt.getRequest(); Request request = evt.getRequest();
if (device != null) { if (device != null) {
logger.info("收到设备" + requesterId + "的语音广播Invite请求"); logger.info("收到设备" + requesterId + "的语音广播Invite请求");
@ -740,7 +740,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
int ssrcIndex = contentString.indexOf("y="); int ssrcIndex = contentString.indexOf("y=");
if (ssrcIndex > 0) { if (ssrcIndex > 0) {
substring = contentString.substring(0, ssrcIndex); substring = contentString.substring(0, ssrcIndex);
ssrc = contentString.substring(ssrcIndex + 2, ssrcIndex + 12); ssrc = contentString.substring(ssrcIndex + 2, ssrcIndex + 12).trim();
} }
ssrcIndex = substring.indexOf("f="); ssrcIndex = substring.indexOf("f=");
if (ssrcIndex > 0) { if (ssrcIndex > 0) {
@ -750,13 +750,13 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
// 获取支持的格式 // 获取支持的格式
Vector mediaDescriptions = sdp.getMediaDescriptions(true); Vector mediaDescriptions = sdp.getMediaDescriptions(true);
// 查看是否支持PS 负载96 // 查看是否支持PS 负载96
int port = -1; int port = -1;
//boolean recvonly = false;
boolean mediaTransmissionTCP = false; boolean mediaTransmissionTCP = false;
Boolean tcpActive = null; Boolean tcpActive = null;
for (int i = 0; i < mediaDescriptions.size(); i++) { for (int i = 0; i < mediaDescriptions.size(); i++) {
MediaDescription mediaDescription = (MediaDescription) mediaDescriptions.get(i); MediaDescription mediaDescription = (MediaDescription)mediaDescriptions.get(i);
Media media = mediaDescription.getMedia(); Media media = mediaDescription.getMedia();
Vector mediaFormats = media.getMediaFormats(false); Vector mediaFormats = media.getMediaFormats(false);
@ -784,233 +784,248 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
responseAck(evt, Response.UNSUPPORTED_MEDIA_TYPE); // 不支持的格式，发415 responseAck(evt, Response.UNSUPPORTED_MEDIA_TYPE); // 不支持的格式，发415
return; return;
} }
String username = sdp.getOrigin().getUsername();
String addressStr = sdp.getOrigin().getAddress(); String addressStr = sdp.getOrigin().getAddress();
logger.info("设备{}请求语音流，地址：{}:{}，ssrc：{}", username, addressStr, port, ssrc); logger.info("设备{}请求语音流，地址：{}:{}，ssrc：{}", requesterId, addressStr, port, ssrc);
MediaServerItem mediaServerItem = playService.getNewMediaServerItem(device);
if (mediaServerItem == null) {
logger.warn("未找到可用的zlm");
responseAck(evt, Response.BUSY_HERE);
return;
}
SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
device.getDeviceId(), audioBroadcastCatch.getChannelId(),
mediaTransmissionTCP);
if (sendRtpItem == null) {
logger.warn("服务器端口资源不足");
responseAck(evt, Response.BUSY_HERE);
return;
}
sendRtpItem.setTcp(mediaTransmissionTCP);
if (tcpActive != null) {
sendRtpItem.setTcpActive(tcpActive);
}
String app = "broadcast";
String stream = device.getDeviceId() + "_" + audioBroadcastCatch.getChannelId();
CallIdHeader callIdHeader = (CallIdHeader) request.getHeader(CallIdHeader.NAME);
sendRtpItem.setPlayType(InviteStreamType.PLAY);
sendRtpItem.setCallId(callIdHeader.getCallId());
sendRtpItem.setPlatformId(requesterId);
sendRtpItem.setStatus(1);
sendRtpItem.setApp(app);
sendRtpItem.setStreamId(stream);
sendRtpItem.setPt(8);
sendRtpItem.setUsePs(false);
sendRtpItem.setOnlyAudio(true);
redisCatchStorage.updateSendRTPSever(sendRtpItem);
// hook监听等待设备推流上来
// 添加订阅
JSONObject subscribeKey = new JSONObject();
subscribeKey.put("app", app);
subscribeKey.put("stream", stream);
subscribeKey.put("regist", true);
subscribeKey.put("schema", "rtmp");
subscribeKey.put("mediaServerId", mediaServerItem.getId());
String finalSsrc = ssrc;
// 流已经存在时直接推流
JSONObject mediaInfo = zlmresTfulUtils.getMediaInfo(mediaServerItem, app, "rtsp", stream);
JSONArray tracks = mediaInfo.getJSONArray("tracks");
Integer codecId = null;
if (tracks != null && tracks.size() > 0) {
for (int i = 0; i < tracks.size(); i++) {
MediaItem.MediaTrack track = JSON.toJavaObject((JSON)tracks.get(i),MediaItem.MediaTrack.class);
if (track.getCodecType() == 1) {
codecId = track.getCodecId();
break;
}
}
}
if ((mediaInfo.getInteger("code") == 0 && mediaInfo.getBoolean("online"))) {
logger.info("发现已经在推流");
sendRtpItem.setStatus(2);
redisCatchStorage.updateSendRTPSever(sendRtpItem);
StringBuffer content = new StringBuffer(200);
content.append("v=0\r\n");
content.append("o="+ config.getId() +" "+ sdp.getOrigin().getSessionId() +" " + sdp.getOrigin().getSessionVersion() + " IN IP4 "+mediaServerItem.getSdpIp()+"\r\n");
content.append("s=Play\r\n");
content.append("c=IN IP4 "+mediaServerItem.getSdpIp()+"\r\n");
content.append("t=0 0\r\n");
if (codecId == null) {
if (mediaTransmissionTCP) {
content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 8\r\n");
}else {
content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n");
}
content.append("a=rtpmap:8 PCMA/8000\r\n");
}else {
if (codecId == 4) {
if (mediaTransmissionTCP) {
content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 0\r\n");
}else {
content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 0\r\n");
}
content.append("a=rtpmap:0 PCMU/8000\r\n");
}else {
if (mediaTransmissionTCP) {
content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 8\r\n");
}else {
content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n");
}
content.append("a=rtpmap:8 PCMA/8000\r\n");
}
}
if (sendRtpItem.isTcp()) {
content.append("a=connection:new\r\n");
if (!sendRtpItem.isTcpActive()) {
content.append("a=setup:active\r\n");
}else {
content.append("a=setup:passive\r\n");
}
}
content.append("a=sendonly\r\n");
content.append("y="+ finalSsrc + "\r\n");
content.append("f=v/////a/1/8/1\r\n");
ParentPlatform parentPlatform = new ParentPlatform();
parentPlatform.setServerIP(device.getIp());
parentPlatform.setServerPort(device.getPort());
parentPlatform.setServerGBId(device.getDeviceId());
try {
responseSdpAck(evt, content.toString(), parentPlatform);
Dialog dialog = evt.getDialog();
audioBroadcastCatch.setDialog((SIPDialog) dialog);
audioBroadcastCatch.setRequest((SIPRequest) request);
audioBroadcastManager.update(audioBroadcastCatch);
} catch (SipException e) {
throw new RuntimeException(e);
} catch (InvalidArgumentException e) {
throw new RuntimeException(e);
} catch (ParseException e) {
throw new RuntimeException(e);
}
}else {
// 流不存在时监听流上线
// 设置等待推流的超时; 默认20s
String waiteStreamTimeoutTaskKey = "waite-stream-" + device.getDeviceId() + audioBroadcastCatch.getChannelId();
dynamicTask.startDelay(waiteStreamTimeoutTaskKey, ()->{
logger.info("等待推流超时: {}/{}", app, stream);
subscribe.removeSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey);
playService.stopAudioBroadcast(device.getDeviceId(), audioBroadcastCatch.getChannelId());
// 发送bye
try {
responseAck(evt, Response.BUSY_HERE);
} catch (SipException e) {
throw new RuntimeException(e);
} catch (InvalidArgumentException e) {
throw new RuntimeException(e);
} catch (ParseException e) {
throw new RuntimeException(e);
}
}, 20*1000);
boolean finalMediaTransmissionTCP = mediaTransmissionTCP;
subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,
(MediaServerItem mediaServerItemInUse, JSONObject json)->{
logger.info("收到语音对讲推流");
MediaItem mediaItem = JSON.toJavaObject(json, MediaItem.class);
Integer audioCodecId = null;
if (mediaItem.getTracks() != null && mediaItem.getTracks().size() > 0) {
for (int i = 0; i < mediaItem.getTracks().size(); i++) {
MediaItem.MediaTrack mediaTrack = mediaItem.getTracks().get(i);
if (mediaTrack.getCodecType() == 1) {
audioCodecId = mediaTrack.getCodecId();
break;
}
}
}
try {
sendRtpItem.setStatus(2);
redisCatchStorage.updateSendRTPSever(sendRtpItem);
StringBuffer content = new StringBuffer(200);
content.append("v=0\r\n");
content.append("o="+ config.getId() +" "+ sdp.getOrigin().getSessionId() +" " + sdp.getOrigin().getSessionVersion() + " IN IP4 "+mediaServerItem.getSdpIp()+"\r\n");
content.append("s=Play\r\n");
content.append("c=IN IP4 "+mediaServerItem.getSdpIp()+"\r\n");
content.append("t=0 0\r\n");
if (audioCodecId == null) {
if (finalMediaTransmissionTCP) {
content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 8\r\n");
}else {
content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n");
}
content.append("a=rtpmap:8 PCMA/8000\r\n");
}else {
if (audioCodecId == 4) {
if (finalMediaTransmissionTCP) {
content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 0\r\n");
}else {
content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 0\r\n");
}
content.append("a=rtpmap:0 PCMU/8000\r\n");
}else {
if (finalMediaTransmissionTCP) {
content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 8\r\n");
}else {
content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n");
}
content.append("a=rtpmap:8 PCMA/8000\r\n");
}
}
content.append("a=sendonly\r\n");
if (sendRtpItem.isTcp()) {
content.append("a=connection:new\r\n");
if (!sendRtpItem.isTcpActive()) {
content.append("a=setup:active\r\n");
}else {
content.append("a=setup:passive\r\n");
}
}
content.append("y="+ finalSsrc + "\r\n");
content.append("f=v/////a/1/8/1\r\n");
ParentPlatform parentPlatform = new ParentPlatform();
parentPlatform.setServerIP(device.getIp());
parentPlatform.setServerPort(device.getPort());
parentPlatform.setServerGBId(device.getDeviceId());
responseSdpAck(evt, content.toString(), parentPlatform);
Dialog dialog = evt.getDialog();
audioBroadcastCatch.setDialog((SIPDialog) dialog);
audioBroadcastCatch.setRequest((SIPRequest) request);
audioBroadcastManager.update(audioBroadcastCatch);
} catch (SipException e) {
throw new RuntimeException(e);
} catch (InvalidArgumentException e) {
throw new RuntimeException(e);
} catch (ParseException e) {
throw new RuntimeException(e);
} catch (SdpParseException e) {
throw new RuntimeException(e);
}
});
}
String key = DeferredResultHolder.CALLBACK_CMD_BROADCAST + device.getDeviceId();
WVPResult<AudioBroadcastResult> wvpResult = new WVPResult<>();
wvpResult.setCode(0);
wvpResult.setMsg("success");
AudioBroadcastResult audioBroadcastResult = new AudioBroadcastResult();
audioBroadcastResult.setApp(app);
audioBroadcastResult.setStream(stream);
audioBroadcastResult.setMediaServerItem(new MediaServerItemLite(mediaServerItem));
audioBroadcastResult.setCodec("G.711");
wvpResult.setData(audioBroadcastResult);
RequestMessage requestMessage = new RequestMessage();
requestMessage.setKey(key);
requestMessage.setData(wvpResult);
resultHolder.invokeAllResult(requestMessage);
} else { } else {
logger.warn("来自无效设备/平台的请求"); logger.warn("来自无效设备/平台的请求");
responseAck(evt, Response.BAD_REQUEST); responseAck(evt, Response.BAD_REQUEST);
} }
} }
CallIdHeader callIdHeader = (CallIdHeader) request.getHeader(CallIdHeader.NAME);
sendRtpItem.setPlayType(InviteStreamType.PLAY);
sendRtpItem.setCallId(callIdHeader.getCallId());
sendRtpItem.setPlatformId(requesterId);
sendRtpItem.setStatus(1);
sendRtpItem.setApp(app);
sendRtpItem.setStreamId(stream);
sendRtpItem.setPt(8);
sendRtpItem.setUsePs(false);
sendRtpItem.setOnlyAudio(true);
redisCatchStorage.updateSendRTPSever(sendRtpItem);
// hook监听等待设备推流上来
// 添加订阅
JSONObject subscribeKey = new JSONObject();
subscribeKey.put("app", app);
subscribeKey.put("stream", stream);
subscribeKey.put("regist", true);
subscribeKey.put("schema", "rtmp");
subscribeKey.put("mediaServerId", mediaServerItem.getId());
String finalSsrc = ssrc;
// 流已经存在时直接推流
JSONObject mediaInfo = zlmresTfulUtils.getMediaInfo(mediaServerItem, app, "rtsp", stream);
JSONArray tracks = mediaInfo.getJSONArray("tracks");
Integer codecId = null;
if (tracks != null && tracks.size() > 0) {
for (int i = 0; i < tracks.size(); i++) {
MediaItem.MediaTrack track = JSON.toJavaObject((JSON)tracks.get(i),MediaItem.MediaTrack.class);
if (track.getCodecType() == 1) {
codecId = track.getCodecId();
break;
}
}
}
if ((mediaInfo.getInteger("code") == 0 && mediaInfo.getBoolean("online"))) {
logger.info("发现已经在推流");
sendRtpItem.setStatus(2);
redisCatchStorage.updateSendRTPSever(sendRtpItem);
StringBuffer content = new StringBuffer(200);
content.append("v=0\r\n");
content.append("o="+ config.getId() +" "+ sdp.getOrigin().getSessionId() +" " + sdp.getOrigin().getSessionVersion() + " IN IP4 "+mediaServerItem.getSdpIp()+"\r\n");
content.append("s=Play\r\n");
content.append("c=IN IP4 "+mediaServerItem.getSdpIp()+"\r\n");
content.append("t=0 0\r\n");
if (codecId == null) {
if (mediaTransmissionTCP) {
content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 8\r\n");
}else {
content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n");
}
content.append("a=rtpmap:8 PCMA/8000\r\n");
}else {
if (codecId == 4) {
if (mediaTransmissionTCP) {
content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 0\r\n");
}else {
content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 0\r\n");
}
content.append("a=rtpmap:0 PCMU/8000\r\n");
}else {
if (mediaTransmissionTCP) {
content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 8\r\n");
}else {
content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n");
}
content.append("a=rtpmap:8 PCMA/8000\r\n");
}
}
if (sendRtpItem.isTcp()) {
content.append("a=connection:new\r\n");
if (!sendRtpItem.isTcpActive()) {
content.append("a=setup:active\r\n");
}else {
content.append("a=setup:passive\r\n");
}
}
content.append("a=sendonly\r\n");
content.append("y="+ finalSsrc + "\r\n");
content.append("f=v/////a/1/8/1\r\n");
ParentPlatform parentPlatform = new ParentPlatform();
parentPlatform.setServerIP(device.getIp());
parentPlatform.setServerPort(device.getPort());
parentPlatform.setServerGBId(device.getDeviceId());
try {
responseSdpAck(evt, content.toString(), parentPlatform);
Dialog dialog = evt.getDialog();
audioBroadcastCatch.setDialog((SIPDialog) dialog);
audioBroadcastCatch.setRequest((SIPRequest) request);
audioBroadcastManager.update(audioBroadcastCatch);
} catch (SipException e) {
throw new RuntimeException(e);
} catch (InvalidArgumentException e) {
throw new RuntimeException(e);
} catch (ParseException e) {
throw new RuntimeException(e);
}
}else {
// 流不存在时监听流上线
// 设置等待推流的超时; 默认20s
String waiteStreamTimeoutTaskKey = "waite-stream-" + device.getDeviceId() + audioBroadcastCatch.getChannelId();
dynamicTask.startDelay(waiteStreamTimeoutTaskKey, ()->{
logger.info("等待推流超时: {}/{}", app, stream);
subscribe.removeSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey);
playService.stopAudioBroadcast(device.getDeviceId(), audioBroadcastCatch.getChannelId());
// 发送bye
try {
responseAck(evt, Response.BUSY_HERE);
} catch (SipException e) {
throw new RuntimeException(e);
} catch (InvalidArgumentException e) {
throw new RuntimeException(e);
} catch (ParseException e) {
throw new RuntimeException(e);
}
}, 20*1000);
boolean finalMediaTransmissionTCP = mediaTransmissionTCP;
subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,
(MediaServerItem mediaServerItemInUse, JSONObject json)->{
logger.info("收到语音对讲推流");
MediaItem mediaItem = JSON.toJavaObject(json, MediaItem.class);
Integer audioCodecId = null;
if (mediaItem.getTracks() != null && mediaItem.getTracks().size() > 0) {
for (int i = 0; i < mediaItem.getTracks().size(); i++) {
MediaItem.MediaTrack mediaTrack = mediaItem.getTracks().get(i);
if (mediaTrack.getCodecType() == 1) {
audioCodecId = mediaTrack.getCodecId();
break;
}
}
}
try {
sendRtpItem.setStatus(2);
redisCatchStorage.updateSendRTPSever(sendRtpItem);
StringBuffer content = new StringBuffer(200);
content.append("v=0\r\n");
content.append("o="+ config.getId() +" "+ sdp.getOrigin().getSessionId() +" " + sdp.getOrigin().getSessionVersion() + " IN IP4 "+mediaServerItem.getSdpIp()+"\r\n");
content.append("s=Play\r\n");
content.append("c=IN IP4 "+mediaServerItem.getSdpIp()+"\r\n");
content.append("t=0 0\r\n");
if (audioCodecId == null) {
if (finalMediaTransmissionTCP) {
content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 8\r\n");
}else {
content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n");
}
content.append("a=rtpmap:8 PCMA/8000\r\n");
}else {
if (audioCodecId == 4) {
if (finalMediaTransmissionTCP) {
content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 0\r\n");
}else {
content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 0\r\n");
}
content.append("a=rtpmap:0 PCMU/8000\r\n");
}else {
if (finalMediaTransmissionTCP) {
content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 8\r\n");
}else {
content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n");
}
content.append("a=rtpmap:8 PCMA/8000\r\n");
}
}
content.append("a=sendonly\r\n");
if (sendRtpItem.isTcp()) {
content.append("a=connection:new\r\n");
if (!sendRtpItem.isTcpActive()) {
content.append("a=setup:active\r\n");
}else {
content.append("a=setup:passive\r\n");
}
}
content.append("y="+ finalSsrc + "\r\n");
content.append("f=v/////a/1/8/1\r\n");
ParentPlatform parentPlatform = new ParentPlatform();
parentPlatform.setServerIP(device.getIp());
parentPlatform.setServerPort(device.getPort());
parentPlatform.setServerGBId(device.getDeviceId());
responseSdpAck(evt, content.toString(), parentPlatform);
Dialog dialog = evt.getDialog();
audioBroadcastCatch.setDialog((SIPDialog) dialog);
audioBroadcastCatch.setRequest((SIPRequest) request);
audioBroadcastManager.update(audioBroadcastCatch);
} catch (SipException e) {
throw new RuntimeException(e);
} catch (InvalidArgumentException e) {
throw new RuntimeException(e);
} catch (ParseException e) {
throw new RuntimeException(e);
} catch (SdpParseException e) {
throw new RuntimeException(e);
}
});
}
String key = DeferredResultHolder.CALLBACK_CMD_BROADCAST + device.getDeviceId();
WVPResult<AudioBroadcastResult> wvpResult = new WVPResult<>();
wvpResult.setCode(0);
wvpResult.setMsg("success");
AudioBroadcastResult audioBroadcastResult = new AudioBroadcastResult();
audioBroadcastResult.setApp(app);
audioBroadcastResult.setStream(stream);
audioBroadcastResult.setMediaServerItem(new MediaServerItemLite(mediaServerItem));
audioBroadcastResult.setCodec("G.711");
wvpResult.setData(audioBroadcastResult);
RequestMessage requestMessage = new RequestMessage();
requestMessage.setKey(key);
requestMessage.setData(wvpResult);
resultHolder.invokeAllResult(requestMessage);
} else {
logger.warn("来自无效设备/平台的请求");
responseAck(evt, Response.BAD_REQUEST);
}
}
} }

1
src/main/java/com/genersoft/iot/vmp/service/IMediaServerService.java
View File

@ -1,7 +1,6 @@
package com.genersoft.iot.vmp.service; package com.genersoft.iot.vmp.service;
import com.alibaba.fastjson.JSONObject; import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.media.zlm.ZLMServerConfig; import com.genersoft.iot.vmp.media.zlm.ZLMServerConfig;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.bean.SSRCInfo; import com.genersoft.iot.vmp.service.bean.SSRCInfo;

4
src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java
View File

@ -6,7 +6,6 @@ import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.common.VideoManagerConstants; import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.conf.SipConfig; import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.SsrcTransaction;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher; import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.session.SsrcConfig; import com.genersoft.iot.vmp.gb28181.session.SsrcConfig;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager; import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
@ -34,11 +33,8 @@ import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.TransactionStatus;
import org.springframework.util.StringUtils; import org.springframework.util.StringUtils;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime; import java.time.LocalDateTime;
import java.util.*; import java.util.*;
import java.util.stream.Collectors;
/** /**
* 媒体服务器节点管理 * 媒体服务器节点管理

17
web_src/package-lock.json generated
View File

@ -50,7 +50,7 @@
"postcss-url": "^7.2.1", "postcss-url": "^7.2.1",
"rimraf": "^2.6.0", "rimraf": "^2.6.0",
"semver": "^5.3.0", "semver": "^5.3.0",
"shelljs": "^0.7.6", "shelljs": "^0.8.5",
"uglifyjs-webpack-plugin": "^1.1.1", "uglifyjs-webpack-plugin": "^1.1.1",
"url-loader": "^0.5.8", "url-loader": "^0.5.8",
"vue-loader": "^13.3.0", "vue-loader": "^13.3.0",
@ -11655,9 +11655,9 @@
} }
}, },
"node_modules/shelljs": { "node_modules/shelljs": {
"version": "0.7.8", "version": "0.8.5",
"resolved": "https://registry.npm.taobao.org/shelljs/download/shelljs-0.7.8.tgz", "resolved": "https://registry.npmmirror.com/shelljs/-/shelljs-0.8.5.tgz",
"integrity": "sha1-3svPh0sNHl+3LhSxZKloMEjprLM=", "integrity": "sha512-TiwcRcrkhHvbrZbnRcFYMLl30Dfov3HKqzp5tO5b4pt6G/SezKcYhmDg15zXVBswHmctSAQKznqNW2LO5tTDow==",
"dev": true, "dev": true,
"dependencies": { "dependencies": {
"glob": "^7.0.0", "glob": "^7.0.0",
@ -11668,8 +11668,7 @@
"shjs": "bin/shjs" "shjs": "bin/shjs"
}, },
"engines": { "engines": {
"iojs": "*", "node": ">=4"
"node": ">=0.11.0"
} }
}, },
"node_modules/shellwords": { "node_modules/shellwords": {
@ -24156,9 +24155,9 @@
"dev": true "dev": true
}, },
"shelljs": { "shelljs": {
"version": "0.7.8", "version": "0.8.5",
"resolved": "https://registry.npm.taobao.org/shelljs/download/shelljs-0.7.8.tgz", "resolved": "https://registry.npmmirror.com/shelljs/-/shelljs-0.8.5.tgz",
"integrity": "sha1-3svPh0sNHl+3LhSxZKloMEjprLM=", "integrity": "sha512-TiwcRcrkhHvbrZbnRcFYMLl30Dfov3HKqzp5tO5b4pt6G/SezKcYhmDg15zXVBswHmctSAQKznqNW2LO5tTDow==",
"dev": true, "dev": true,
"requires": { "requires": {
"glob": "^7.0.0", "glob": "^7.0.0",