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

添加拉流代理与国标关联, 支持代理rtsp/rtmp/...，转发到国标

Browse Source
This commit is contained in:
panlinlin
2021-04-01 18:06:21 +08:00
parent 56859d09df
commit 7dc8fd4a1e
42 changed files with 1394 additions and 224 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
src/main/java/com/genersoft/iot/vmp/gb28181/bean/GbStream.java Normal file
View File

@@ -0,0 +1,71 @@
package com.genersoft.iot.vmp.gb28181.bean;
/**
* 直播流关联国标上级平台
*/
public class GbStream extends PlatformGbStream{
private String app;
private String stream;
private String gbId;
private String name;
private double longitude;
private double latitude;
private String streamType;
public String getApp() {
return app;
}
public void setApp(String app) {
this.app = app;
}
public String getStream() {
return stream;
}
public void setStream(String stream) {
this.stream = stream;
}
public String getGbId() {
return gbId;
}
public void setGbId(String gbId) {
this.gbId = gbId;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public double getLongitude() {
return longitude;
}
public void setLongitude(double longitude) {
this.longitude = longitude;
}
public double getLatitude() {
return latitude;
}
public void setLatitude(double latitude) {
this.latitude = latitude;
}
public String getStreamType() {
return streamType;
}
public void setStreamType(String streamType) {
this.streamType = streamType;
}
}

31
src/main/java/com/genersoft/iot/vmp/gb28181/bean/PlatformGbStream.java Normal file
View File

@@ -0,0 +1,31 @@
package com.genersoft.iot.vmp.gb28181.bean;
public class PlatformGbStream {
private String app;
private String stream;
private String platformId;
public String getApp() {
return app;
}
public void setApp(String app) {
this.app = app;
}
public String getStream() {
return stream;
}
public void setStream(String stream) {
this.stream = stream;
}
public String getPlatformId() {
return platformId;
}
public void setPlatformId(String platformId) {
this.platformId = platformId;
}
}

9
src/main/java/com/genersoft/iot/vmp/gb28181/bean/SendRtpItem.java
View File

@@ -27,6 +27,11 @@ public class SendRtpItem {
*/
private String deviceId;
/**
* 直播流的应用名
*/
private String app;
/**
* 通道id
*/
@@ -40,10 +45,6 @@ public class SendRtpItem {
*/
private int status = 0;
/**
* 设备推流的app
*/
private String app = "rtp";
/**
* 设备推流的streamId

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

@@ -91,7 +91,9 @@ public class SIPCommander implements ISIPCommander {
@Autowired
private SipSubscribe sipSubscribe;
public SipConfig getSipConfig() {
return sipConfig;
}
/**
* 云台方向放控制，使用配置文件中的默认镜头移动速度

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

@@ -197,7 +197,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
@Override
public boolean catalogQuery(DeviceChannel channel, ParentPlatform parentPlatform, String sn, String fromTag, int size) {
if (channel == null || parentPlatform ==null) {
if ( parentPlatform ==null) {
return false;
}
try {
@@ -210,20 +210,22 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
catalogXml.append("<SumNum>" + size + "</SumNum>\r\n");
catalogXml.append("<DeviceList Num=\"1\">\r\n");
catalogXml.append("<Item>\r\n");
if (channel != null) {
catalogXml.append("<DeviceID>" + channel.getChannelId() + "</DeviceID>\r\n");
catalogXml.append("<Name>" + channel.getName() + "</Name>\r\n");
catalogXml.append("<Manufacturer>" + channel.getManufacture() + "</Manufacturer>\r\n");
catalogXml.append("<Model>" + channel.getModel() + "</Model>\r\n");
catalogXml.append("<Owner>" + channel.getOwner() + "</Owner>\r\n");
catalogXml.append("<CivilCode>" + channel.getCivilCode() + "</CivilCode>\r\n");
catalogXml.append("<Address>" + channel.getAddress() + "</Address>\r\n");
catalogXml.append("<Parental>" + channel.getParental() + "</Parental>\r\n");// TODO 当前不能添加分组， 所以暂时没有父节点
catalogXml.append("<ParentID>" + channel.getParentId() + "</ParentID>\r\n"); // TODO 当前不能添加分组， 所以暂时没有父节点
catalogXml.append("<Secrecy>" + channel.getSecrecy() + "</Secrecy>\r\n");
catalogXml.append("<RegisterWay>" + channel.getRegisterWay() + "</RegisterWay>\r\n");
catalogXml.append("<Status>" + (channel.getStatus() == 0?"OFF":"ON") + "</Status>\r\n");
catalogXml.append("<Info></Info>\r\n");
}
catalogXml.append("<DeviceID>" + channel.getChannelId() + "</DeviceID>\r\n");
catalogXml.append("<Name>" + channel.getName() + "</Name>\r\n");
catalogXml.append("<Manufacturer>" + channel.getManufacture() + "</Manufacturer>\r\n");
catalogXml.append("<Model>" + channel.getModel() + "</Model>\r\n");
catalogXml.append("<Owner>" + channel.getOwner() + "</Owner>\r\n");
catalogXml.append("<CivilCode>" + channel.getCivilCode() + "</CivilCode>\r\n");
catalogXml.append("<Address>" + channel.getAddress() + "</Address>\r\n");
catalogXml.append("<Parental>" + channel.getParental() + "</Parental>\r\n");// TODO 当前不能添加分组， 所以暂时没有父节点
catalogXml.append("<ParentID>" + channel.getParentId() + "</ParentID>\r\n"); // TODO 当前不能添加分组， 所以暂时没有父节点
catalogXml.append("<Secrecy>" + channel.getSecrecy() + "</Secrecy>\r\n");
catalogXml.append("<RegisterWay>" + channel.getRegisterWay() + "</RegisterWay>\r\n");
catalogXml.append("<Status>" + (channel.getStatus() == 0?"OFF":"ON") + "</Status>\r\n");
catalogXml.append("<Info></Info>\r\n");
catalogXml.append("</Item>\r\n");
catalogXml.append("</DeviceList>\r\n");

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

@@ -43,14 +43,23 @@ public class AckRequestProcessor extends SIPRequestAbstractProcessor {
SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(platformGbId, channelId);
String is_Udp = sendRtpItem.isTcp() ? "0" : "1";
String deviceId = sendRtpItem.getDeviceId();
StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId);
sendRtpItem.setStreamId(streamInfo.getStreamId());
StreamInfo streamInfo = null;
if (deviceId == null) {
streamInfo = new StreamInfo();
streamInfo.setApp(sendRtpItem.getApp());
streamInfo.setStreamId(sendRtpItem.getStreamId());
}else {
streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId);
sendRtpItem.setStreamId(streamInfo.getStreamId());
streamInfo.setApp("rtp");
}
redisCatchStorage.updateSendRTPSever(sendRtpItem);
System.out.println(platformGbId);
System.out.println(channelId);
Map<String, Object> param = new HashMap<>();
param.put("vhost","__defaultVhost__");
param.put("app","rtp");
param.put("app",streamInfo.getApp());
param.put("stream",streamInfo.getStreamId());
param.put("ssrc", sendRtpItem.getSsrc());
param.put("dst_url",sendRtpItem.getIp());
@@ -63,7 +72,7 @@ public class AckRequestProcessor extends SIPRequestAbstractProcessor {
while (!rtpPushed) {
try {
if (System.currentTimeMillis() - startTime < 30 * 1000) {
if (zlmrtpServerFactory.isRtpReady(streamInfo.getStreamId())) {
if (zlmrtpServerFactory.isStreamReady(streamInfo.getApp(), streamInfo.getStreamId())) {
rtpPushed = true;
System.out.println("已获取设备推流，开始向上级推流");
zlmrtpServerFactory.startSendRtpStream(param);

4
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/ByeRequestProcessor.java
View File

@@ -51,12 +51,12 @@ public class ByeRequestProcessor extends SIPRequestAbstractProcessor {
String streamId = sendRtpItem.getStreamId();
Map<String, Object> param = new HashMap<>();
param.put("vhost","__defaultVhost__");
param.put("app","rtp");
param.put("app",sendRtpItem.getApp());
param.put("stream",streamId);
System.out.println("停止向上级推流：" + streamId);
zlmrtpServerFactory.stopSendRtpStream(param);
redisCatchStorage.deleteSendRTPServer(platformGbId, channelId);
if (zlmrtpServerFactory.totalReaderCount(streamId) == 0) {
if (zlmrtpServerFactory.totalReaderCount(sendRtpItem.getApp(), streamId) == 0) {
System.out.println(streamId + "无其它观看者，通知设备停止推流");
cmder.streamByeCmd(streamId);
}

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

@@ -12,13 +12,12 @@ import javax.sip.message.Request;
import javax.sip.message.Response;
import com.genersoft.iot.vmp.conf.MediaServerConfig;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
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.gb28181.transmit.request.SIPRequestAbstractProcessor;
import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
@@ -30,6 +29,8 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.text.ParseException;
import java.util.HashMap;
import java.util.Map;
import java.util.Vector;
/**
@@ -93,12 +94,14 @@ public class InviteRequestProcessor extends SIPRequestAbstractProcessor {
if (platform != null) {
// 查询平台下是否有该通道
DeviceChannel channel = storager.queryChannelInParentPlatform(requesterId, channelId);
if (channel == null) {
GbStream gbStream = storager.queryStreamInParentPlatform(requesterId, channelId);
// 不是通道可能是直播流
if (channel != null || gbStream != null ) {
responseAck(evt, Response.CALL_IS_BEING_FORWARDED); // 通道存在，发181，呼叫转接中
}else {
logger.info("通道不存在，返回404");
responseAck(evt, Response.NOT_FOUND); // 通道不存在，发404，资源不存在
return;
}else {
responseAck(evt, Response.CALL_IS_BEING_FORWARDED); // 通道存在，发181，呼叫转接中
}
// 解析sdp消息, 使用jainsip 自带的sdp解析方式
String contentString = new String(request.getRawContent());
@@ -153,67 +156,120 @@ public class InviteRequestProcessor extends SIPRequestAbstractProcessor {
String addressStr = sdp.getOrigin().getAddress();
//String sessionName = sdp.getSessionName().getValue();
logger.info("[上级点播]用户：{}， 地址：{}:{}， ssrc：{}", username, addressStr, port, ssrc);
Device device = null;
// 通过 channel 和 gbStream 是否为null 值判断来源是直播流合适国标
if (channel != null) {
device = storager.queryVideoDeviceByPlatformIdAndChannelId(requesterId, channelId);
if (device == null) {
logger.warn("点播平台{}的通道{}时未找到设备信息", requesterId, channel);
responseAck(evt, Response.SERVER_INTERNAL_ERROR);
return;
}
SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(addressStr, port, ssrc, requesterId,
device.getDeviceId(), channelId,
mediaTransmissionTCP);
if (tcpActive != null) {
sendRtpItem.setTcpActive(tcpActive);
}
if (sendRtpItem == null) {
logger.warn("服务器端口资源不足");
responseAck(evt, Response.BUSY_HERE);
return;
}
Device device = storager.queryVideoDeviceByPlatformIdAndChannelId(requesterId, channelId);
if (device == null) {
logger.warn("点播平台{}的通道{}时未找到设备信息", requesterId, channel);
responseAck(evt, Response.SERVER_INTERNAL_ERROR);
return;
}
SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(addressStr, port, ssrc, requesterId, device.getDeviceId(), channelId,
mediaTransmissionTCP);
if (tcpActive != null) {
sendRtpItem.setTcpActive(tcpActive);
}
if (sendRtpItem == null) {
logger.warn("服务器端口资源不足");
responseAck(evt, Response.BUSY_HERE);
return;
}
// 写入redis， 超时时回复
redisCatchStorage.updateSendRTPSever(sendRtpItem);
// 通知下级推流，
PlayResult playResult = playService.play(device.getDeviceId(), channelId, (responseJSON)->{
// 收到推流， 回复200OK, 等待ack
sendRtpItem.setStatus(1);
// 写入redis， 超时时回复
redisCatchStorage.updateSendRTPSever(sendRtpItem);
// TODO 添加对tcp的支持
MediaServerConfig mediaInfo = redisCatchStorage.getMediaInfo();
StringBuffer content = new StringBuffer(200);
content.append("v=0\r\n");
content.append("o="+"00000"+" 0 0 IN IP4 "+mediaInfo.getWanIp()+"\r\n");
content.append("s=Play\r\n");
content.append("c=IN IP4 "+mediaInfo.getWanIp()+"\r\n");
content.append("t=0 0\r\n");
content.append("m=video "+ sendRtpItem.getLocalPort()+" RTP/AVP 96\r\n");
content.append("a=sendonly\r\n");
content.append("a=rtpmap:96 PS/90000\r\n");
content.append("y="+ ssrc + "\r\n");
content.append("f=\r\n");
// 通知下级推流，
PlayResult playResult = playService.play(device.getDeviceId(), channelId, (responseJSON)->{
// 收到推流， 回复200OK, 等待ack
if (sendRtpItem == null) return;
sendRtpItem.setStatus(1);
redisCatchStorage.updateSendRTPSever(sendRtpItem);
// TODO 添加对tcp的支持
MediaServerConfig mediaInfo = redisCatchStorage.getMediaInfo();
StringBuffer content = new StringBuffer(200);
content.append("v=0\r\n");
content.append("o="+"00000"+" 0 0 IN IP4 "+mediaInfo.getWanIp()+"\r\n");
content.append("s=Play\r\n");
content.append("c=IN IP4 "+mediaInfo.getWanIp()+"\r\n");
content.append("t=0 0\r\n");
content.append("m=video "+ sendRtpItem.getLocalPort()+" RTP/AVP 96\r\n");
content.append("a=sendonly\r\n");
content.append("a=rtpmap:96 PS/90000\r\n");
content.append("y="+ ssrc + "\r\n");
content.append("f=\r\n");
try {
responseAck(evt, content.toString());
} catch (SipException e) {
e.printStackTrace();
} catch (InvalidArgumentException e) {
e.printStackTrace();
} catch (ParseException e) {
e.printStackTrace();
try {
responseAck(evt, content.toString());
} catch (SipException e) {
e.printStackTrace();
} catch (InvalidArgumentException e) {
e.printStackTrace();
} catch (ParseException e) {
e.printStackTrace();
}
} ,(event -> {
// 未知错误。直接转发设备点播的错误
Response response = null;
try {
response = getMessageFactory().createResponse(event.getResponse().getStatusCode(), evt.getRequest());
getServerTransaction(evt).sendResponse(response);
} catch (ParseException | SipException | InvalidArgumentException e) {
e.printStackTrace();
}
}));
if (logger.isDebugEnabled()) {
logger.debug(playResult.getResult().toString());
}
},(event -> {
// 未知错误。直接转发设备点播的错误
Response response = null;
try {
response = getMessageFactory().createResponse(event.getResponse().getStatusCode(), evt.getRequest());
getServerTransaction(evt).sendResponse(response);
} catch (ParseException | SipException | InvalidArgumentException e) {
e.printStackTrace();
}else if (gbStream != null) {
SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(addressStr, port, ssrc, requesterId,
gbStream.getApp(), gbStream.getStream(), channelId,
mediaTransmissionTCP);
if (tcpActive != null) {
sendRtpItem.setTcpActive(tcpActive);
}
if (sendRtpItem == null) {
logger.warn("服务器端口资源不足");
responseAck(evt, Response.BUSY_HERE);
return;
}
// 写入redis， 超时时回复
redisCatchStorage.updateSendRTPSever(sendRtpItem);
// 检测直播流是否在线
Boolean streamReady = zlmrtpServerFactory.isStreamReady(gbStream.getApp(), gbStream.getStream());
if (streamReady) {
sendRtpItem.setStatus(1);
redisCatchStorage.updateSendRTPSever(sendRtpItem);
// TODO 添加对tcp的支持
MediaServerConfig mediaInfo = redisCatchStorage.getMediaInfo();
StringBuffer content = new StringBuffer(200);
content.append("v=0\r\n");
content.append("o="+"00000"+" 0 0 IN IP4 "+mediaInfo.getWanIp()+"\r\n");
content.append("s=Play\r\n");
content.append("c=IN IP4 "+mediaInfo.getWanIp()+"\r\n");
content.append("t=0 0\r\n");
content.append("m=video "+ sendRtpItem.getLocalPort()+" RTP/AVP 96\r\n");
content.append("a=sendonly\r\n");
content.append("a=rtpmap:96 PS/90000\r\n");
content.append("y="+ ssrc + "\r\n");
content.append("f=\r\n");
try {
responseAck(evt, content.toString());
} catch (SipException e) {
e.printStackTrace();
} catch (InvalidArgumentException e) {
e.printStackTrace();
} catch (ParseException e) {
e.printStackTrace();
}
}
}));
if (logger.isDebugEnabled()) {
logger.debug(playResult.getResult().toString());
}
} else {
// 非上级平台请求，查询是否设备请求（通常为接收语音广播的设备）
Device device = storager.queryVideoDevice(requesterId);
@@ -298,6 +354,7 @@ public class InviteRequestProcessor extends SIPRequestAbstractProcessor {
}
}
/***
* 回复状态码
* 100 trying

33
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/MessageRequestProcessor.java
View File

@@ -529,13 +529,44 @@ public class MessageRequestProcessor extends SIPRequestAbstractProcessor {
String sn = snElement.getText();
// 准备回复通道信息
List<ChannelReduce> channelReduces = storager.queryChannelListInParentPlatform(parentPlatform.getServerGBId());
// 查询关联的直播通道
List<GbStream> gbStreams = storager.queryGbStreamListInPlatform(parentPlatform.getServerGBId());
int size = channelReduces.size() + gbStreams.size();
// 回复级联的通道
if (channelReduces.size() > 0) {
for (ChannelReduce channelReduce : channelReduces) {
DeviceChannel deviceChannel = storager.queryChannel(channelReduce.getDeviceId(), channelReduce.getChannelId());
cmderFroPlatform.catalogQuery(deviceChannel, parentPlatform, sn, fromHeader.getTag(), channelReduces.size());
cmderFroPlatform.catalogQuery(deviceChannel, parentPlatform, sn, fromHeader.getTag(), size);
}
}
// 回复直播的通道
if (gbStreams.size() > 0) {
for (GbStream gbStream : gbStreams) {
DeviceChannel deviceChannel = new DeviceChannel();
deviceChannel.setChannelId(gbStream.getGbId());
deviceChannel.setName(gbStream.getName());
deviceChannel.setLongitude(gbStream.getLongitude());
deviceChannel.setLatitude(gbStream.getLatitude());
deviceChannel.setDeviceId(parentPlatform.getDeviceGBId());
deviceChannel.setManufacture("wvp-pro");
deviceChannel.setStatus(1);
// deviceChannel.setParentId(parentPlatform.getDeviceGBId());
deviceChannel.setRegisterWay(1);
deviceChannel.setCivilCode(cmder.getSipConfig().getSipDomain());
deviceChannel.setModel("live");
deviceChannel.setOwner("wvp-pro");
// deviceChannel.setAddress("test");
deviceChannel.setParental(0);
deviceChannel.setSecrecy("0");
deviceChannel.setSecrecy("0");
cmderFroPlatform.catalogQuery(deviceChannel, parentPlatform, sn, fromHeader.getTag(), size);
}
}
if (size == 0) {
// 回复无通道
cmderFroPlatform.catalogQuery(null, parentPlatform, sn, fromHeader.getTag(), size);
}
}