src/main/java/com/genersoft/iot/vmp/gb28181/bean/SendRtpItem.java
@@ -22,7 +22,12 @@ */ private String platformId; /** /** * 对应设备id */ private String deviceId; /** * 通道id */ private String channelId; @@ -92,6 +97,14 @@ this.platformId = platformId; } public String getDeviceId() { return deviceId; } public void setDeviceId(String deviceId) { this.deviceId = deviceId; } public String getChannelId() { return channelId; } src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorFactory.java
@@ -4,25 +4,17 @@ 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 @@ @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 @@ } 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(); 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 @@ */ @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 { 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(); } // 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; } } src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/InviteRequestProcessor.java
@@ -7,14 +7,10 @@ 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.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 @@ * @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 @@ // 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 @@ // 获取支持的格式 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 @@ 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 @@ } 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 @@ 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); src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/MessageRequestProcessor.java
@@ -370,31 +370,36 @@ 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(); src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRESTfulUtils.java
@@ -119,4 +119,8 @@ public JSONObject closeRtpServer(Map<String, Object> param) { return sendPost("closeRtpServer",param); } public JSONObject startSendRtp(Map<String, Object> param) { return sendPost("startSendRtp",param); } } src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java
@@ -105,7 +105,7 @@ * @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 @@ 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; } } } src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java
@@ -4,7 +4,6 @@ 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 @@ void delPlatformRegisterInfo(String callId); void updateSendRTPSever(SendRtpItem sendRtpItem); /** * 查询RTP推送信息缓存 * @param platformGbId * @param channelId * @return sendRtpItem */ SendRtpItem querySendRTPServer(String platformGbId, String channelId); } src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java
@@ -218,4 +218,11 @@ 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); } }
