实现级联点播RTP推送

lawrencehj
1 parent b7857027
Showing 9 changed files with 207 additions and 70 deletions
src/main/java/com/genersoft/iot/vmp/gb28181/bean/SendRtpItem.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorFactory.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/AckRequestProcessor.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/InviteRequestProcessor.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/MessageRequestProcessor.java
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRESTfulUtils.java
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java
src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java
src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.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;
     }
@@ -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();
 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<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();
+		// }
 	}
+	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;
+	}
+
 }
@@ -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);
@@ -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();
@@ -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);
+    }
 }
@@ -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;
+        }
+    }
 }
@@ -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);
+
 }
@@ -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);
+    }
+
 }