Shikong/wvp-gb28181-project
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

实现级联点播RTP推送

Browse Source
This commit is contained in:
lawrencehj 2021-03-09 20:05:37 +08:00
parent f83bb3639a
commit 66b7d6e1ae
9 changed files with 207 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -22,6 +22,11 @@ public class SendRtpItem {
*/
private String platformId;
/**
* 对应设备id
*/
private String deviceId;
/**
* 通道id
*/
@ -92,6 +97,14 @@ public class SendRtpItem {
this.platformId = platformId;
}
public String getDeviceId() {
return deviceId;
}
public void setDeviceId(String deviceId) {
this.deviceId = deviceId;
}
public String getChannelId() {
return channelId;
}

12
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorFactory.java
View File

@ -4,25 +4,17 @@ import javax.sip.RequestEvent;
import javax.sip.ResponseEvent;
import javax.sip.SipProvider;
import javax.sip.header.CSeqHeader;
import javax.sip.header.CallIdHeader;
import javax.sip.header.Header;
import javax.sip.message.Request;
import javax.sip.message.Response;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.alibaba.fastjson.JSON;
import com.genersoft.iot.vmp.gb28181.transmit.response.impl.*;
import com.genersoft.iot.vmp.gb28181.transmit.response.impl.*;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.vmanager.service.IPlayService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
// 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.context.annotation.Lazy;
import org.springframework.stereotype.Component;
@ -59,7 +51,7 @@ import com.genersoft.iot.vmp.utils.redis.RedisUtil;
@Component
public class SIPProcessorFactory {
private final static Logger logger = LoggerFactory.getLogger(SIPProcessorFactory.class);
// private final static Logger logger = LoggerFactory.getLogger(SIPProcessorFactory.class);
@Autowired
private SipConfig sipConfig;
@ -156,6 +148,8 @@ public class SIPProcessorFactory {
} else if (Request.ACK.equals(method)) {
AckRequestProcessor processor = new AckRequestProcessor();
processor.setRequestEvent(evt);
processor.setRedisCatchStorage(redisCatchStorage);
processor.setZlmrtpServerFactory(zlmrtpServerFactory);
return processor;
} else if (Request.BYE.equals(method)) {
ByeRequestProcessor processor = new ByeRequestProcessor();

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

@ -1,19 +1,34 @@
package com.genersoft.iot.vmp.gb28181.transmit.request.impl;
import java.util.HashMap;
import java.util.Map;
import javax.sip.*;
import javax.sip.message.Request;
//import javax.sip.message.Request;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.gb28181.transmit.request.SIPRequestAbstractProcessor;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import gov.nist.javax.sip.header.CSeq;
import org.springframework.stereotype.Component;
/**
* @Description:ACK请求处理器
* @author: swwheihei
* @date: 2020年5月3日 下午5:31:45
*/
@Component
public class AckRequestProcessor extends SIPRequestAbstractProcessor {
//@Autowired
private IRedisCatchStorage redisCatchStorage;
//@Autowired
private ZLMRTPServerFactory zlmrtpServerFactory;
/**
* 处理 ACK请求
*
@ -21,25 +36,77 @@ public class AckRequestProcessor extends SIPRequestAbstractProcessor {
*/
@Override
public void process(RequestEvent evt) {
Request request = evt.getRequest();
//Request request = evt.getRequest();
Dialog dialog = evt.getDialog();
DialogState state = dialog.getState();
if (dialog == null) return;
if (request.getMethod().equals(Request.INVITE) && dialog.getState()== DialogState.CONFIRMED) {
// TODO 查询并开始推流
}
//DialogState state = dialog.getState();
if (/*request.getMethod().equals(Request.INVITE) &&*/ dialog.getState()== DialogState.CONFIRMED) {
String remoteUri = dialog.getRemoteParty().getURI().toString();
String localUri = dialog.getLocalParty().getURI().toString();
String platformGbId = remoteUri.substring(remoteUri.indexOf(":") + 1, remoteUri.indexOf("@"));
String channelId = localUri.substring(remoteUri.indexOf(":") + 1, remoteUri.indexOf("@"));
SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(platformGbId, channelId);
String is_Udp = sendRtpItem.isTcp() ? "0" : "1";
String deviceId = sendRtpItem.getDeviceId();
StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId);
System.out.println(platformGbId);
System.out.println(channelId);
Map<String, Object> param = new HashMap<>();
param.put("vhost","__defaultVhost__");
param.put("app","rtp");
param.put("stream",streamInfo.getStreamId());
param.put("ssrc", sendRtpItem.getSsrc());
param.put("dst_url",sendRtpItem.getIp());
param.put("dst_port", sendRtpItem.getPort());
param.put("is_udp", is_Udp);
//param.put ("src_port", sendRtpItem.getLocalPort());
// 设备推流查询，成功后才能转推
boolean rtpPushed = false;
long startTime = System.currentTimeMillis();
while (!rtpPushed) {
try {
Request ackRequest = null;
CSeq csReq = (CSeq) request.getHeader(CSeq.NAME);
ackRequest = dialog.createAck(csReq.getSeqNumber());
dialog.sendAck(ackRequest);
System.out.println("send ack to callee:" + ackRequest.toString());
} catch (SipException e) {
e.printStackTrace();
} catch (InvalidArgumentException e) {
if (System.currentTimeMillis() - startTime < 30 * 1000) {
if (zlmrtpServerFactory.isRtpReady(streamInfo.getStreamId())) {
rtpPushed = true;
zlmrtpServerFactory.startSendRtpStream(param);
} else {
Thread.sleep(2000);
continue;
}
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
// try {
// Request ackRequest = null;
// CSeq csReq = (CSeq) request.getHeader(CSeq.NAME);
// ackRequest = dialog.createAck(csReq.getSeqNumber());
// dialog.sendAck(ackRequest);
// System.out.println("send ack to callee:" + ackRequest.toString());
// } catch (SipException e) {
// e.printStackTrace();
// } catch (InvalidArgumentException e) {
// e.printStackTrace();
// }
}
public IRedisCatchStorage getRedisCatchStorage() {
return redisCatchStorage;
}
public void setRedisCatchStorage(IRedisCatchStorage redisCatchStorage) {
this.redisCatchStorage = redisCatchStorage;
}
public ZLMRTPServerFactory getZlmrtpServerFactory() {
return zlmrtpServerFactory;
}
public void setZlmrtpServerFactory(ZLMRTPServerFactory zlmrtpServerFactory) {
this.zlmrtpServerFactory = zlmrtpServerFactory;
}
}

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

@ -7,14 +7,10 @@ import javax.sip.SipException;
import javax.sip.SipFactory;
import javax.sip.address.Address;
import javax.sip.address.SipURI;
import javax.sip.header.ContentTypeHeader;
import javax.sip.header.FromHeader;
import javax.sip.header.HeaderFactory;
import javax.sip.header.SubjectHeader;
import javax.sip.header.*;
import javax.sip.message.Request;
import javax.sip.message.Response;
import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.conf.MediaServerConfig;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
@ -27,17 +23,12 @@ import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
import com.genersoft.iot.vmp.vmanager.play.bean.PlayResult;
import com.genersoft.iot.vmp.vmanager.service.IPlayService;
import gov.nist.javax.sdp.fields.SDPFormat;
import gov.nist.javax.sip.address.AddressImpl;
import gov.nist.javax.sip.address.SipUri;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import java.io.IOException;
import java.text.ParseException;
import java.util.List;
import java.util.UUID;
import java.util.Vector;
/**
@ -45,6 +36,7 @@ import java.util.Vector;
* @author: panll
* @date: 2021年1月14日
*/
@SuppressWarnings("rawtypes")
public class InviteRequestProcessor extends SIPRequestAbstractProcessor {
private final static Logger logger = LoggerFactory.getLogger(MessageRequestProcessor.class);
@ -122,8 +114,10 @@ public class InviteRequestProcessor extends SIPRequestAbstractProcessor {
// jainSip不支持y=字段， 移除移除以解析。
int ssrcIndex = contentString.indexOf("y=");
String ssrc = contentString.substring(ssrcIndex + 2, contentString.length())
.replace("\r\n", "").replace("\n", "");
String ssrc = contentString.substring(ssrcIndex + 2, ssrcIndex + 12);
//ssrc规定长度为10字节，不取余下长度以避免后续还有“f=”字段
// String ssrc = contentString.substring(ssrcIndex + 2, contentString.length())
// .replace("\r\n", "").replace("\n", "");
String substring = contentString.substring(0, contentString.indexOf("y="));
SessionDescription sdp = SdpFactory.getInstance().createSessionDescription(substring);
@ -131,9 +125,9 @@ public class InviteRequestProcessor extends SIPRequestAbstractProcessor {
// 获取支持的格式
Vector mediaDescriptions = sdp.getMediaDescriptions(true);
// 查看是否支持PS 负载96
String ip = null;
//String ip = null;
int port = -1;
boolean recvonly = false;
//boolean recvonly = false;
boolean mediaTransmissionTCP = false;
Boolean tcpActive = null;
for (int i = 0; i < mediaDescriptions.size(); i++) {
@ -143,7 +137,7 @@ public class InviteRequestProcessor extends SIPRequestAbstractProcessor {
Vector mediaFormats = media.getMediaFormats(false);
if (mediaFormats.contains("96")) {
port = media.getMediaPort();
String mediaType = media.getMediaType();
//String mediaType = media.getMediaType();
String protocol = media.getProtocol();
// 区分TCP发流还是udp， 当前默认udp
@ -169,7 +163,7 @@ public class InviteRequestProcessor extends SIPRequestAbstractProcessor {
}
String username = sdp.getOrigin().getUsername();
String addressStr = sdp.getOrigin().getAddress();
String sessionName = sdp.getSessionName().getValue();
//String sessionName = sdp.getSessionName().getValue();
logger.info("[上级点播]用户：{}， 地址：{}:{}， ssrc：{}", username, addressStr, port, ssrc);
Device device = storager.queryVideoDeviceByPlatformIdAndChannelId(platformId, channelId);
@ -178,7 +172,7 @@ public class InviteRequestProcessor extends SIPRequestAbstractProcessor {
responseAck(evt, Response.SERVER_INTERNAL_ERROR);
return;
}
SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(ip, port, platformId, ssrc, channelId,
SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(addressStr, port, ssrc, platformId, device.getDeviceId(), channelId,
mediaTransmissionTCP);
if (tcpActive != null) {
sendRtpItem.setTcpActive(tcpActive);

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

@ -370,9 +370,13 @@ public class MessageRequestProcessor extends SIPRequestAbstractProcessor {
private void processMessageDeviceInfo(RequestEvent evt) {
try {
Element rootElement = getRootElement(evt);
String requestName = rootElement.getName();
Element deviceIdElement = rootElement.element("DeviceID");
String deviceId = deviceIdElement.getTextTrim().toString();
if (requestName.equals("Query")) {
// 回复200 OK
responseAck(evt);
} else {
Device device = storager.queryVideoDevice(deviceId);
if (device == null) {
return;
@ -396,6 +400,7 @@ public class MessageRequestProcessor extends SIPRequestAbstractProcessor {
if (offLineDetector.isOnline(deviceId)) {
publisher.onlineEventPublish(deviceId, VideoManagerConstants.EVENT_ONLINE_KEEPLIVE);
}
}
} catch (DocumentException | SipException | InvalidArgumentException | ParseException e) {
e.printStackTrace();
}

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

@ -119,4 +119,8 @@ public class ZLMRESTfulUtils {
public JSONObject closeRtpServer(Map<String, Object> param) {
return sendPost("closeRtpServer",param);
}
public JSONObject startSendRtp(Map<String, Object> param) {
return sendPost("startSendRtp",param);
}
}

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

@ -105,7 +105,7 @@ public class ZLMRTPServerFactory {
* @param tcp 是否为tcp
* @return SendRtpItem
*/
public SendRtpItem createSendRtpItem(String ip, int port, String ssrc, String platformId, String channelId, boolean tcp){
public SendRtpItem createSendRtpItem(String ip, int port, String ssrc, String platformId, String deviceId, String channelId, boolean tcp){
String playSsrc = SsrcUtil.getPlaySsrc();
int localPort = createRTPServer(SsrcUtil.getPlaySsrc());
if (localPort != -1) {
@ -119,9 +119,54 @@ public class ZLMRTPServerFactory {
sendRtpItem.setPort(port);
sendRtpItem.setSsrc(ssrc);
sendRtpItem.setPlatformId(platformId);
sendRtpItem.setDeviceId(deviceId);
sendRtpItem.setChannelId(channelId);
sendRtpItem.setTcp(tcp);
sendRtpItem.setLocalPort(localPort);
return sendRtpItem;
}
/**
*
*/
public Boolean startSendRtpStream(Map<String, Object>param) {
Boolean result = false;
JSONObject jsonObject = zlmresTfulUtils.startSendRtp(param);
System.out.println(jsonObject);
if (jsonObject != null) {
switch (jsonObject.getInteger("code")){
case 0:
result= true;
logger.error("RTP推流请求成功，本地推流端口：" + jsonObject.getString("local_port"));
break;
// case -300: // id已经存在
// result = false;
// break;
// case -400: // 端口占用
// result= false;
// break;
default:
logger.error("RTP推流失败: " + jsonObject.getString("msg"));
break;
}
}else {
// 检查ZLM状态
logger.error("RTP推流失败: 请检查ZLM服务");
}
return result;
}
/**
*
*/
public Boolean isRtpReady(String streamId) {
JSONObject mediaInfo = zlmresTfulUtils.getMediaInfo("rtp", "rtmp", streamId);
if (mediaInfo.getInteger("code") == 0 && mediaInfo.getBoolean("online")) {
logger.info("设备RTP推流成功");
return true;
} else {
logger.info("设备RTP推流未完成");
return false;
}
}
}

10
src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java
View File

@ -4,7 +4,6 @@ import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.MediaServerConfig;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatformCatch;
import com.genersoft.iot.vmp.gb28181.bean.PlatformRegister;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import java.util.Map;
@ -81,4 +80,13 @@ public interface IRedisCatchStorage {
void delPlatformRegisterInfo(String callId);
void updateSendRTPSever(SendRtpItem sendRtpItem);
/**
* 查询RTP推送信息缓存
* @param platformGbId
* @param channelId
* @return sendRtpItem
*/
SendRtpItem querySendRTPServer(String platformGbId, String channelId);
}

7
src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java
View File

@ -218,4 +218,11 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
String key = VideoManagerConstants.PLATFORM_SEND_RTP_INFO_PREFIX + sendRtpItem.getPlatformId() + "_" + sendRtpItem.getChannelId();
redis.set(key, sendRtpItem);
}
@Override
public SendRtpItem querySendRTPServer(String platformGbId, String channelId) {
String key = VideoManagerConstants.PLATFORM_SEND_RTP_INFO_PREFIX + platformGbId + "_" + channelId;
return (SendRtpItem)redis.get(key);
}
}