对级联点播信令进行处理

src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java

@@ -30,6 +30,8 @@ public class VideoManagerConstants {
 
 	public static final String PLATFORM_REGISTER_INFO_PREFIX = "VMP_platform_register_info_";
 
+	public static final String PLATFORM_SEND_RTP_INFO_PREFIX = "VMP_platform_send_rtp_info_";
+
 	public static final String Pattern_Topic = "VMP_keeplive_platform_";
 
 	public static final String EVENT_ONLINE_REGISTER = "1";

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

@@ -0,0 +1,150 @@
+package com.genersoft.iot.vmp.gb28181.bean;
+
+public class SendRtpItem {
+
+    /**
+     * 推流ip
+     */
+    private String ip;
+
+    /**
+     * 推流端口
+     */
+    private int port;
+
+    /**
+     * 推流标识
+     */
+    private String ssrc;
+
+    /**
+     * 平台id
+     */
+    private String platformId;
+
+    /**
+     * 通道id
+     */
+    private String channelId;
+
+    /**
+     * 推流状态
+     * 0 等待设备推流上来
+     * 1 等待上级平台回复ack
+     * 2 推流中
+     */
+    private int status = 0;
+
+    /**
+     * 设备推流的app
+     */
+    private String app = "rtp";
+
+    /**
+     * 设备推流的streamId
+     */
+    private String streamId;
+
+    /**
+     * 是否为tcp
+     */
+    private boolean tcp;
+
+    /**
+     * 是否为tcp主动模式
+     */
+    private boolean tcpActive;
+
+    /**
+     * 自己推流使用的端口
+     */
+    private int localPort;
+
+    public String getIp() {
+        return ip;
+    }
+
+    public void setIp(String ip) {
+        this.ip = ip;
+    }
+
+    public int getPort() {
+        return port;
+    }
+
+    public void setPort(int port) {
+        this.port = port;
+    }
+
+    public String getSsrc() {
+        return ssrc;
+    }
+
+    public void setSsrc(String ssrc) {
+        this.ssrc = ssrc;
+    }
+
+    public String getPlatformId() {
+        return platformId;
+    }
+
+    public void setPlatformId(String platformId) {
+        this.platformId = platformId;
+    }
+
+    public String getChannelId() {
+        return channelId;
+    }
+
+    public void setChannelId(String channelId) {
+        this.channelId = channelId;
+    }
+
+    public int getStatus() {
+        return status;
+    }
+
+    public void setStatus(int status) {
+        this.status = status;
+    }
+
+    public String getApp() {
+        return app;
+    }
+
+    public void setApp(String app) {
+        this.app = app;
+    }
+
+    public String getStreamId() {
+        return streamId;
+    }
+
+    public void setStreamId(String streamId) {
+        this.streamId = streamId;
+    }
+
+    public boolean isTcp() {
+        return tcp;
+    }
+
+    public void setTcp(boolean tcp) {
+        this.tcp = tcp;
+    }
+
+    public int getLocalPort() {
+        return localPort;
+    }
+
+    public void setLocalPort(int localPort) {
+        this.localPort = localPort;
+    }
+
+    public boolean isTcpActive() {
+        return tcpActive;
+    }
+
+    public void setTcpActive(boolean tcpActive) {
+        this.tcpActive = tcpActive;
+    }
+}

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

@@ -136,6 +136,7 @@ public class SIPProcessorFactory {
 			processor.setCmderFroPlatform(cmderFroPlatform);
 			processor.setPlayService(playService);
 			processor.setStorager(storager);
+			processor.setRedisCatchStorage(redisCatchStorage);
 			processor.setZlmrtpServerFactory(zlmrtpServerFactory);
 			return processor;
 		} else if (Request.REGISTER.equals(method)) {

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

@@ -1,9 +1,6 @@
 package com.genersoft.iot.vmp.gb28181.transmit.request.impl;
 
-import javax.sip.Dialog;
+import javax.sip.*;
-import javax.sip.InvalidArgumentException;
-import javax.sip.RequestEvent;
-import javax.sip.SipException;
 import javax.sip.message.Request;
 
 import com.genersoft.iot.vmp.gb28181.transmit.request.SIPRequestAbstractProcessor;
@@ -26,7 +23,11 @@ public class AckRequestProcessor extends SIPRequestAbstractProcessor {
 	public void process(RequestEvent evt) {
 		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);

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

@@ -4,9 +4,12 @@ import javax.sdp.*;
 import javax.sip.InvalidArgumentException;
 import javax.sip.RequestEvent;
 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.message.Request;
 import javax.sip.message.Response;
@@ -15,6 +18,7 @@ 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;
+import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
 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;
@@ -100,16 +104,18 @@ public class InviteRequestProcessor extends SIPRequestAbstractProcessor {
 			platformId = uri.getUser();
 
 			if (platformId == null || channelId == null) {
-				response400Ack(evt); // 参数不全， 发400，请求错误
+				logger.info("无法从FromHeader的Address中获取到平台id，返回404");
+				responseAck(evt, Response.BAD_REQUEST); // 参数不全， 发400，请求错误
 				return;
 			}
 			// 查询平台下是否有该通道
 			DeviceChannel channel = storager.queryChannelInParentPlatform(platformId, channelId);
 			if (channel == null) {
-				response404Ack(evt); // 通道不存在，发404，资源不存在
+				logger.info("通道不存在，返回404");
+				responseAck(evt, Response.NOT_FOUND); // 通道不存在，发404，资源不存在
 				return;
 			}else {
-				response100Ack(evt); // 通道存在，发100，trying
+				responseAck(evt, Response.TRYING); // 通道存在，发100，trying
 			}
 			// 解析sdp消息, 使用jainsip 自带的sdp解析方式
 			String contentString = new String(request.getRawContent());
@@ -152,107 +158,79 @@ public class InviteRequestProcessor extends SIPRequestAbstractProcessor {
 							}
 						}
 					}
-//					Vector attributes = mediaDescription.getAttributes(false);
-//					for (Object attributeObj : attributes) {
-//						Attribute attribute = (Attribute)attributeObj;
-//						String name = attribute.getName();
-//						switch (name){
-//							case "recvonly":
-//								recvonly = true;
-//								break;
-//							case "rtpmap":
-//							case "connection":
-//								break;
-//							case "setup":
-//								mediaTransmissionTCP = true;
-//								if ("active".equals(attribute.getValue())) {  // tcp主动模式
-//									tcpActive = true;
-//								}else if ("passive".equals(attribute.getValue())){ // tcp被动模式
-//									tcpActive = false;
-//								}
-//								break;
-//
-//						}
-//						if ("recvonly".equals(name)) {
-//							recvonly = true;
-//						}
-//
-//						String value = attribute.getValue();
-//					}
 					break;
 				}
 			}
 			if (port == -1) {
+				logger.info("不支持的媒体格式，返回415");
 				// 回复不支持的格式
-				response415Ack(evt); // 不支持的格式，发415
+				responseAck(evt, Response.UNSUPPORTED_MEDIA_TYPE); // 不支持的格式，发415
 				return;
 			}
 			String username = sdp.getOrigin().getUsername();
 			String addressStr = sdp.getOrigin().getAddress();
 			String sessionName = sdp.getSessionName().getValue();
 			logger.info("[上级点播]用户：{}， 地址：{}:{}， ssrc：{}", username, addressStr, port, ssrc);
-//
-//			Device device = storager.queryVideoDeviceByPlatformIdAndChannelId(platformId, channelId);
-//			if (device == null) {
-//				logger.warn("点播平台{}的通道{}时未找到设备信息", platformId, channel);
-//				response500Ack(evt);
-//				return;
-//			}
-//
-//			// 通知下级推流，
-//			PlayResult playResult = playService.play(device.getDeviceId(), channelId, (responseJSON)->{
-//				// 收到推流， 回复200OK
-//				UUID uuid = UUID.randomUUID();
-//				int rtpServer = zlmrtpServerFactory.createRTPServer(uuid.toString());
-//				if (rtpServer == -1) {
-//					logger.error("为获取到可用端口");
-//					return;
-//				}else {
-//					zlmrtpServerFactory.closeRTPServer(uuid.toString());
-//				}
-//				// 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 "+ rtpServer+" 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();
-//				}
-//
-//				// 写入redis， 超时时回复
-////				redisCatchStorage.waiteAck()
-//			},(event -> {
-//				// 未知错误。直接转发设备点播的错误
-//				Response response = null;
-//				try {
-//					response = getMessageFactory().createResponse(event.getResponse().getStatusCode(), evt.getRequest());
-//					getServerTransaction(evt).sendResponse(response);
-//
-//				} catch (ParseException | SipException | InvalidArgumentException e) {
-//					e.printStackTrace();
-//				}
-//			}));
-//			playResult.getResult();
-			// 查找合适的端口推流，
-			// 收到ack后调用推流接口
 
+			Device device = storager.queryVideoDeviceByPlatformIdAndChannelId(platformId, channelId);
+			if (device == null) {
+				logger.warn("点播平台{}的通道{}时未找到设备信息", platformId, channel);
+				responseAck(evt, Response.SERVER_INTERNAL_ERROR);
+				return;
+			}
+			SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(ip, port, platformId, ssrc, 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);
+				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();
+				}
+			},(event -> {
+				// 未知错误。直接转发设备点播的错误
+				Response response = null;
+				try {
+					response = getMessageFactory().createResponse(event.getResponse().getStatusCode(), evt.getRequest());
+					getServerTransaction(evt).sendResponse(response);
+
+				} catch (ParseException | SipException | InvalidArgumentException e) {
+					e.printStackTrace();
+				}
+			}));
+			playResult.getResult();
 
 		} catch (SipException | InvalidArgumentException | ParseException e) {
 			e.printStackTrace();
@@ -263,101 +241,47 @@ public class InviteRequestProcessor extends SIPRequestAbstractProcessor {
 		} catch (SdpException e) {
 			e.printStackTrace();
 		}
-
 	}
 
 	/***
-	 * 回复100 trying
+	 * 回复状态码
+	 * 100 trying
+	 * 200 OK
+	 * 400
+	 * 404
 	 * @param evt
 	 * @throws SipException
 	 * @throws InvalidArgumentException
 	 * @throws ParseException
 	 */
-	private void response100Ack(RequestEvent evt) throws SipException, InvalidArgumentException, ParseException {
+	private void responseAck(RequestEvent evt, int statusCode) throws SipException, InvalidArgumentException, ParseException {
 		Response response = getMessageFactory().createResponse(Response.TRYING, evt.getRequest());
 		getServerTransaction(evt).sendResponse(response);
 	}
 
-	/***
+	/**
-	 * 回复200 OK
+	 * 回复带sdp的200
 	 * @param evt
+	 * @param sdp
 	 * @throws SipException
 	 * @throws InvalidArgumentException
 	 * @throws ParseException
 	 */
 	private void responseAck(RequestEvent evt, String sdp) throws SipException, InvalidArgumentException, ParseException {
 		Response response = getMessageFactory().createResponse(Response.OK, evt.getRequest());
-		ContentTypeHeader contentTypeHeader = getHeaderFactory().createContentTypeHeader("APPLICATION", "SDP");
+		SipFactory sipFactory = SipFactory.getInstance();
+		ContentTypeHeader contentTypeHeader = sipFactory.createHeaderFactory().createContentTypeHeader("APPLICATION", "SDP");
 		response.setContent(sdp, contentTypeHeader);
+
+		SipURI sipURI = (SipURI)evt.getRequest().getRequestURI();
+
+		Address concatAddress = sipFactory.createAddressFactory().createAddress(
+				sipFactory.createAddressFactory().createSipURI(sipURI.getUser(),  sipURI.getHost()+":"+sipURI.getPort()
+				));
+		response.addHeader(sipFactory.createHeaderFactory().createContactHeader(concatAddress));
 		getServerTransaction(evt).sendResponse(response);
 	}
 
-	/***
-	 * 回复400
-	 * @param evt
-	 * @throws SipException
-	 * @throws InvalidArgumentException
-	 * @throws ParseException
-	 */
-	private void response400Ack(RequestEvent evt) throws SipException, InvalidArgumentException, ParseException {
-		Response response = getMessageFactory().createResponse(Response.BAD_REQUEST, evt.getRequest());
-		getServerTransaction(evt).sendResponse(response);
-	}
-
-	/***
-	 * 回复404
-	 * @param evt
-	 * @throws SipException
-	 * @throws InvalidArgumentException
-	 * @throws ParseException
-	 */
-	private void response404Ack(RequestEvent evt) throws SipException, InvalidArgumentException, ParseException {
-		Response response = getMessageFactory().createResponse(Response.NOT_FOUND, evt.getRequest());
-		getServerTransaction(evt).sendResponse(response);
-	}
-
-	/***
-	 * 回复415 不支持的媒体类型
-	 * @param evt
-	 * @throws SipException
-	 * @throws InvalidArgumentException
-	 * @throws ParseException
-	 */
-	private void response415Ack(RequestEvent evt) throws SipException, InvalidArgumentException, ParseException {
-		Response response = getMessageFactory().createResponse(Response.UNSUPPORTED_MEDIA_TYPE, evt.getRequest());
-		getServerTransaction(evt).sendResponse(response);
-	}
-
-	/***
-	 * 回复488
-	 * @param evt
-	 * @throws SipException
-	 * @throws InvalidArgumentException
-	 * @throws ParseException
-	 */
-	private void response488Ack(RequestEvent evt) throws SipException, InvalidArgumentException, ParseException {
-		Response response = getMessageFactory().createResponse(Response.NOT_ACCEPTABLE_HERE, evt.getRequest());
-		getServerTransaction(evt).sendResponse(response);
-	}
-
-	/***
-	 * 回复500
-	 * @param evt
-	 * @throws SipException
-	 * @throws InvalidArgumentException
-	 * @throws ParseException
-	 */
-	private void response500Ack(RequestEvent evt) throws SipException, InvalidArgumentException, ParseException {
-		Response response = getMessageFactory().createResponse(Response.SERVER_INTERNAL_ERROR, evt.getRequest());
-		getServerTransaction(evt).sendResponse(response);
-	}
-
-
-
-
-
-
-
 
 
 
@@ -394,4 +318,12 @@ public class InviteRequestProcessor extends SIPRequestAbstractProcessor {
 	public void setPlayService(IPlayService playService) {
 		this.playService = playService;
 	}
+
+	public IRedisCatchStorage getRedisCatchStorage() {
+		return redisCatchStorage;
+	}
+
+	public void setRedisCatchStorage(IRedisCatchStorage redisCatchStorage) {
+		this.redisCatchStorage = redisCatchStorage;
+	}
 }

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

@@ -1,6 +1,8 @@
 package com.genersoft.iot.vmp.media.zlm;
 
 import com.alibaba.fastjson.JSONObject;
+import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
+import com.genersoft.iot.vmp.gb28181.session.SsrcUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -92,4 +94,34 @@ public class ZLMRTPServerFactory {
             return currentPort++;
         }
     }
+
+    /**
+     * 创建一个推流
+     * @param ip 推流ip
+     * @param port 推流端口
+     * @param ssrc 推流唯一标识
+     * @param platformId 平台id
+     * @param channelId 通道id
+     * @param tcp 是否为tcp
+     * @return SendRtpItem
+     */
+    public SendRtpItem createSendRtpItem(String ip, int port, String ssrc, String platformId, String channelId, boolean tcp){
+        String playSsrc = SsrcUtil.getPlaySsrc();
+        int localPort = createRTPServer(SsrcUtil.getPlaySsrc());
+        if (localPort != -1) {
+            closeRTPServer(playSsrc);
+        }else {
+            logger.error("没有可用的端口");
+            return null;
+        }
+        SendRtpItem sendRtpItem = new SendRtpItem();
+        sendRtpItem.setIp(ip);
+        sendRtpItem.setPort(port);
+        sendRtpItem.setSsrc(ssrc);
+        sendRtpItem.setPlatformId(platformId);
+        sendRtpItem.setChannelId(channelId);
+        sendRtpItem.setTcp(tcp);
+        sendRtpItem.setLocalPort(localPort);
+        return sendRtpItem;
+    }
 }

src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java

@@ -5,6 +5,7 @@ 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;
 
@@ -78,4 +79,6 @@ public interface IRedisCatchStorage {
     String queryPlatformRegisterInfo(String callId);
 
     void delPlatformRegisterInfo(String callId);
+
+    void updateSendRTPSever(SendRtpItem sendRtpItem);
 }

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

@@ -3,10 +3,7 @@ package com.genersoft.iot.vmp.storager.impl;
 import com.genersoft.iot.vmp.common.StreamInfo;
 import com.genersoft.iot.vmp.common.VideoManagerConstants;
 import com.genersoft.iot.vmp.conf.MediaServerConfig;
-import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
+import com.genersoft.iot.vmp.gb28181.bean.*;
-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.storager.IRedisCatchStorage;
 import com.genersoft.iot.vmp.storager.dao.DeviceChannelMapper;
 import com.genersoft.iot.vmp.utils.redis.RedisUtil;
@@ -215,4 +212,10 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
     public void delPlatformRegisterInfo(String callId) {
         redis.del(VideoManagerConstants.PLATFORM_REGISTER_INFO_PREFIX + callId);
     }
+
+    @Override
+    public void updateSendRTPSever(SendRtpItem sendRtpItem) {
+        String key = VideoManagerConstants.PLATFORM_SEND_RTP_INFO_PREFIX + sendRtpItem.getPlatformId() + "_" + sendRtpItem.getChannelId();
+        redis.set(key, sendRtpItem);
+    }
 }