diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SendRtpItem.java b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SendRtpItem.java index f3ec337d..e0220dc9 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SendRtpItem.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SendRtpItem.java @@ -22,7 +22,12 @@ public class SendRtpItem { */ private String platformId; - /** + /** + * 对应设备id + */ + private String deviceId; + + /** * 通道id */ private String channelId; @@ -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; } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorFactory.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorFactory.java index 829bc7f4..2d1ca091 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorFactory.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorFactory.java @@ -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(); diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/AckRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/AckRequestProcessor.java index b72f0139..7f58de5e 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/AckRequestProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/AckRequestProcessor.java @@ -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 查询并开始推流 - } - 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(); + //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 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 { + 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; + } + } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/InviteRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/InviteRequestProcessor.java index ac19d6f6..0c3f127f 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/InviteRequestProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/InviteRequestProcessor.java @@ -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); diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/MessageRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/MessageRequestProcessor.java index 0bd78bd4..a2b37099 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/MessageRequestProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/MessageRequestProcessor.java @@ -370,31 +370,36 @@ 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(); - - Device device = storager.queryVideoDevice(deviceId); - if (device == null) { - return; - } - device.setName(XmlUtil.getText(rootElement, "DeviceName")); - device.setManufacturer(XmlUtil.getText(rootElement, "Manufacturer")); - device.setModel(XmlUtil.getText(rootElement, "Model")); - device.setFirmware(XmlUtil.getText(rootElement, "Firmware")); - if (StringUtils.isEmpty(device.getStreamMode())) { - device.setStreamMode("UDP"); - } - storager.updateDevice(device); - - RequestMessage msg = new RequestMessage(); - msg.setDeviceId(deviceId); - msg.setType(DeferredResultHolder.CALLBACK_CMD_DEVICEINFO); - msg.setData(device); - deferredResultHolder.invokeResult(msg); - // 回复200 OK - responseAck(evt); - if (offLineDetector.isOnline(deviceId)) { - publisher.onlineEventPublish(deviceId, VideoManagerConstants.EVENT_ONLINE_KEEPLIVE); + if (requestName.equals("Query")) { + // 回复200 OK + responseAck(evt); + } else { + Device device = storager.queryVideoDevice(deviceId); + if (device == null) { + return; + } + device.setName(XmlUtil.getText(rootElement, "DeviceName")); + device.setManufacturer(XmlUtil.getText(rootElement, "Manufacturer")); + device.setModel(XmlUtil.getText(rootElement, "Model")); + device.setFirmware(XmlUtil.getText(rootElement, "Firmware")); + if (StringUtils.isEmpty(device.getStreamMode())) { + device.setStreamMode("UDP"); + } + storager.updateDevice(device); + + RequestMessage msg = new RequestMessage(); + msg.setDeviceId(deviceId); + msg.setType(DeferredResultHolder.CALLBACK_CMD_DEVICEINFO); + msg.setData(device); + deferredResultHolder.invokeResult(msg); + // 回复200 OK + responseAck(evt); + if (offLineDetector.isOnline(deviceId)) { + publisher.onlineEventPublish(deviceId, VideoManagerConstants.EVENT_ONLINE_KEEPLIVE); + } } } catch (DocumentException | SipException | InvalidArgumentException | ParseException e) { e.printStackTrace(); diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRESTfulUtils.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRESTfulUtils.java index 58de847e..e3cc8a48 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRESTfulUtils.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRESTfulUtils.java @@ -119,4 +119,8 @@ public class ZLMRESTfulUtils { public JSONObject closeRtpServer(Map param) { return sendPost("closeRtpServer",param); } + + public JSONObject startSendRtp(Map param) { + return sendPost("startSendRtp",param); + } } diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java index a556ba7f..819c6a82 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java @@ -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(Mapparam) { + 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; + } + } } diff --git a/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java b/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java index 4ca25e27..ca70620f 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java @@ -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); + } diff --git a/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java b/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java index a9abdd9e..6153e5f7 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java @@ -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); + } + }