尝试修复catalog获取失败。服务重启后设备未注册仍上报keeplive处理

songww
1 parent d072017b
Showing 11 changed files with 145 additions and 48 deletions
src/main/java/com/genersoft/iot/vmp/gb28181/event/DeviceOffLineDetector.java
src/main/java/com/genersoft/iot/vmp/gb28181/event/EventPublisher.java
src/main/java/com/genersoft/iot/vmp/gb28181/event/outline/KeepliveTimeoutListener.java → src/main/java/com/genersoft/iot/vmp/gb28181/event/offline/KeepliveTimeoutListener.java
src/main/java/com/genersoft/iot/vmp/gb28181/event/outline/OutlineEvent.java → src/main/java/com/genersoft/iot/vmp/gb28181/event/offline/OfflineEvent.java
src/main/java/com/genersoft/iot/vmp/gb28181/event/outline/OutlineEventListener.java → src/main/java/com/genersoft/iot/vmp/gb28181/event/offline/OfflineEventListener.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommander.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/MessageRequestProcessor.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/response/impl/InviteResponseProcessor.java
src/main/java/com/genersoft/iot/vmp/vmanager/playback/PlaybackController.java
src/main/resources/application.yml
+package com.genersoft.iot.vmp.gb28181.event;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import com.genersoft.iot.vmp.common.VideoManagerConstants;
+import com.genersoft.iot.vmp.utils.redis.RedisUtil;
+
+/**    
+ * @Description:设备离在线状态检测器，用于检测设备状态
+ * @author: songww
+ * @date:   2020年5月13日 下午2:40:29     
+ */
+@Component
+public class DeviceOffLineDetector {
+
+	@Autowired
+    private RedisUtil redis;
+	
+	public boolean isOnline(String deviceId) {
+		String key = VideoManagerConstants.KEEPLIVEKEY_PREFIX + deviceId;
+		return redis.hasKey(key);
+	}
+}
@@ -4,8 +4,8 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.stereotype.Component;
+import com.genersoft.iot.vmp.gb28181.event.offline.OfflineEvent;
 import com.genersoft.iot.vmp.gb28181.event.online.OnlineEvent;
-import com.genersoft.iot.vmp.gb28181.event.outline.OutlineEvent;
 /**    
  * @Description:Event事件通知推送器，支持推送在线事件、离线事件
@@ -26,7 +26,7 @@ public class EventPublisher {
     }
 	public void outlineEventPublish(String deviceId, String from){
-		OutlineEvent outEvent = new OutlineEvent(this);
+		OfflineEvent outEvent = new OfflineEvent(this);
 		outEvent.setDeviceId(deviceId);
 		outEvent.setFrom(from);
         applicationEventPublisher.publishEvent(outEvent);
-package com.genersoft.iot.vmp.gb28181.event.outline;
+package com.genersoft.iot.vmp.gb28181.event.offline;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.redis.connection.Message;
-package com.genersoft.iot.vmp.gb28181.event.outline;
+package com.genersoft.iot.vmp.gb28181.event.offline;
 import org.springframework.context.ApplicationEvent;
@@ -7,7 +7,7 @@ import org.springframework.context.ApplicationEvent;
  * @author: songww
  * @date:   2020年5月6日 上午11:33:13     
  */
-public class OutlineEvent extends ApplicationEvent {
+public class OfflineEvent extends ApplicationEvent {
 	/**   
 	 * @Title:  OutlineEvent   
@@ -15,7 +15,7 @@ public class OutlineEvent extends ApplicationEvent {
 	 * @param:  @param source  
 	 * @throws   
 	 */  
-	public OutlineEvent(Object source) {
+	public OfflineEvent(Object source) {
 		super(source);
 	}
-package com.genersoft.iot.vmp.gb28181.event.outline;
+package com.genersoft.iot.vmp.gb28181.event.offline;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -13,14 +13,14 @@ import com.genersoft.iot.vmp.utils.redis.RedisUtil;
 /**
  * @Description: 离线事件监听器，监听到离线后，修改设备离在线状态。 设备离线有两个来源：
  *               1、设备主动注销，发送注销指令，{@link com.genersoft.iot.vmp.gb28181.transmit.request.impl.RegisterRequestProcessor}
- *               2、设备未知原因离线，心跳超时,{@link com.genersoft.iot.vmp.gb28181.event.outline.OutlineEventListener}
+ *               2、设备未知原因离线，心跳超时,{@link com.genersoft.iot.vmp.gb28181.event.offline.OfflineEventListener}
  * @author: songww
  * @date: 2020年5月6日 下午1:51:23
  */
 @Component
-public class OutlineEventListener implements ApplicationListener<OutlineEvent> {
+public class OfflineEventListener implements ApplicationListener<OfflineEvent> {
-	private final static Logger logger = LoggerFactory.getLogger(OutlineEventListener.class);
+	private final static Logger logger = LoggerFactory.getLogger(OfflineEventListener.class);
 	@Autowired
 	private IVideoManagerStorager storager;
@@ -29,7 +29,7 @@ public class OutlineEventListener implements ApplicationListener&lt;OutlineEvent&gt; {
     private RedisUtil redis;
 	@Override
-	public void onApplicationEvent(OutlineEvent event) {
+	public void onApplicationEvent(OfflineEvent event) {
 		if (logger.isDebugEnabled()) {
 			logger.debug("设备离线事件触发，deviceId：" + event.getDeviceId() + ",from:" + event.getFrom());
@@ -79,7 +79,7 @@ public interface ISIPCommander {
 	 * @param startTime 开始时间,格式要求：yyyy-MM-dd HH:mm:ss
 	 * @param endTime 结束时间,格式要求：yyyy-MM-dd HH:mm:ss
 	 */
-	public String playbackStreamCmd(Device device,String channelId, String recordId, String startTime, String endTime);
+	public String playbackStreamCmd(Device device,String channelId, String startTime, String endTime);
 	/**
 	 * 语音广播
 package com.genersoft.iot.vmp.gb28181.transmit.cmd.impl;
 import java.text.ParseException;
-import java.util.Random;
+import javax.sip.ClientTransaction;
 import javax.sip.InvalidArgumentException;
 import javax.sip.SipException;
 import javax.sip.message.Request;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.stereotype.Component;
 import com.genersoft.iot.vmp.conf.SipConfig;
@@ -19,8 +18,6 @@ import com.genersoft.iot.vmp.gb28181.transmit.cmd.SIPRequestHeaderProvider;
 import com.genersoft.iot.vmp.gb28181.utils.DateUtil;
 import com.genersoft.iot.vmp.gb28181.utils.SsrcUtil;
-import tk.mybatis.mapper.util.StringUtil;
-
 /**    
  * @Description:设备能力接口，用于定义设备的控制、查询能力   
  * @author: songww
@@ -181,16 +178,16 @@ public class SIPCommander implements ISIPCommander {
 	 * @param endTime 结束时间,格式要求：yyyy-MM-dd HH:mm:ss
 	 */ 
 	@Override
-	public String playbackStreamCmd(Device device, String channelId, String recordId, String startTime, String endTime) {
+	public String playbackStreamCmd(Device device, String channelId, String startTime, String endTime) {
 		try {
 			String ssrc = SsrcUtil.getPlayBackSsrc();
 			//
 			StringBuffer content = new StringBuffer(200);
 	        content.append("v=0\r\n");
-	        content.append("o="+channelId+" 0 0 IN IP4 "+sipConfig.getSipIp()+"\r\n");
+	        content.append("o="+device.getDeviceId()+" 0 0 IN IP4 "+sipConfig.getSipIp()+"\r\n");
 	        content.append("s=Playback\r\n");
-	        content.append("u="+recordId+":3\r\n");
+	        content.append("u="+channelId+":3\r\n");
 	        content.append("c=IN IP4 "+sipConfig.getMediaIp()+"\r\n");
 	        content.append("t="+DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(startTime)+" "+DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(endTime) +"\r\n");
 	        if(device.getTransport().equals("TCP")) {
@@ -439,11 +436,15 @@ public class SIPCommander implements ISIPCommander {
 	}
 	private void transmitRequest(Device device, Request request) throws SipException {
+		ClientTransaction clientTransaction = null;
 		if(device.getTransport().equals("TCP")) {
-			sipLayer.getTcpSipProvider().sendRequest(request);
+			clientTransaction = sipLayer.getTcpSipProvider().getNewClientTransaction(request);
+			//sipLayer.getTcpSipProvider().sendRequest(request);
 		} else if(device.getTransport().equals("UDP")) {
-			sipLayer.getUdpSipProvider().sendRequest(request);
+			clientTransaction = sipLayer.getUdpSipProvider().getNewClientTransaction(request);
+			//sipLayer.getUdpSipProvider().sendRequest(request);
 		}
+		clientTransaction.sendRequest();
 	}
 }
@@ -30,6 +30,7 @@ import com.genersoft.iot.vmp.gb28181.bean.Device;
 import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
 import com.genersoft.iot.vmp.gb28181.bean.RecordInfo;
 import com.genersoft.iot.vmp.gb28181.bean.RecordItem;
+import com.genersoft.iot.vmp.gb28181.event.DeviceOffLineDetector;
 import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
 import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
 import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
@@ -69,8 +70,21 @@ public class MessageRequestProcessor implements ISIPRequestProcessor {
 	@Autowired
 	private DeferredResultHolder deferredResultHolder;
+	@Autowired
+	private DeviceOffLineDetector offLineDetector;
+	
 	private final static String CACHE_RECORDINFO_KEY = "CACHE_RECORDINFO_";
+	private static final String MESSAGE_CATALOG = "Catalog";
+	private static final String MESSAGE_DEVICE_INFO = "DeviceInfo";
+	private static final String MESSAGE_KEEP_ALIVE = "Keepalive";
+	private static final String MESSAGE_ALARM = "Alarm";
+	private static final String MESSAGE_RECORD_INFO = "RecordInfo";
+//	private static final String MESSAGE_BROADCAST = "Broadcast";
+//	private static final String MESSAGE_DEVICE_STATUS = "DeviceStatus";
+//	private static final String MESSAGE_MOBILE_POSITION = "MobilePosition";
+//	private static final String MESSAGE_MOBILE_POSITION_INTERVAL = "Interval";
+	
 	/**   
 	 * 处理MESSAGE请求
 	 *  
@@ -85,22 +99,31 @@ public class MessageRequestProcessor implements ISIPRequestProcessor {
 		this.transaction = transaction;
 		Request request = evt.getRequest();
-		
-		if (new String(request.getRawContent()).contains("<CmdType>Keepalive</CmdType>")) {
-			logger.info("接收到KeepAlive消息");
-			processMessageKeepAlive(evt);
-		} else if (new String(request.getRawContent()).contains("<CmdType>Catalog</CmdType>")) {
-			logger.info("接收到Catalog消息");
-			processMessageCatalogList(evt);
-		} else if (new String(request.getRawContent()).contains("<CmdType>DeviceInfo</CmdType>")) {
-			logger.info("接收到DeviceInfo消息");
-			processMessageDeviceInfo(evt);
-		} else if (new String(request.getRawContent()).contains("<CmdType>Alarm</CmdType>")) {
-			logger.info("接收到Alarm消息");
-			processMessageAlarm(evt);
-		} else if (new String(request.getRawContent()).contains("<CmdType>RecordInfo</CmdType>")) {
-			logger.info("接收到RecordInfo消息");
-			processMessageRecordInfo(evt);
+		SAXReader reader = new SAXReader();
+		Document xml;
+		try {
+			xml = reader.read(new ByteArrayInputStream(request.getRawContent()));
+			Element rootElement = xml.getRootElement();
+			String cmd = rootElement.element("CmdType").getStringValue();
+			
+			if (MESSAGE_KEEP_ALIVE.equals(cmd)) {
+				logger.info("接收到KeepAlive消息");
+				processMessageKeepAlive(evt);
+			} else if (MESSAGE_CATALOG.equals(cmd)) {
+				logger.info("接收到Catalog消息");
+				processMessageCatalogList(evt);
+			} else if (MESSAGE_DEVICE_INFO.equals(cmd)) {
+				logger.info("接收到DeviceInfo消息");
+				processMessageDeviceInfo(evt);
+			} else if (MESSAGE_ALARM.equals(cmd)) {
+				logger.info("接收到Alarm消息");
+				processMessageAlarm(evt);
+			} else if (MESSAGE_RECORD_INFO.equals(cmd)) {
+				logger.info("接收到RecordInfo消息");
+				processMessageRecordInfo(evt);
+			}
+		} catch (DocumentException e) {
+			e.printStackTrace();
 		}
 	}
@@ -247,12 +270,17 @@ public class MessageRequestProcessor implements ISIPRequestProcessor {
 	 */
 	private void processMessageKeepAlive(RequestEvent evt){
 		try {
-			Request request = evt.getRequest();
-			Response response = layer.getMessageFactory().createResponse(Response.OK,request);
 			Element rootElement = getRootElement(evt);
-			Element deviceIdElement = rootElement.element("DeviceID");
+			String deviceId = XmlUtil.getText(rootElement,"DeviceID");
+			Request request = evt.getRequest();
+			Response response = null;
+			if (offLineDetector.isOnline(deviceId)) {
+				response = layer.getMessageFactory().createResponse(Response.OK,request);
+				publisher.onlineEventPublish(deviceId, VideoManagerConstants.EVENT_ONLINE_KEEPLIVE);
+			} else {
+				response = layer.getMessageFactory().createResponse(Response.BAD_REQUEST,request);
+			}
 			transaction.sendResponse(response);
-			publisher.onlineEventPublish(deviceIdElement.getText(), VideoManagerConstants.EVENT_ONLINE_KEEPLIVE);
 		} catch (ParseException | SipException | InvalidArgumentException | DocumentException e) {
 			e.printStackTrace();
 		}
 package com.genersoft.iot.vmp.gb28181.transmit.response.impl;
+import java.text.ParseException;
+
+import javax.sip.ClientTransaction;
 import javax.sip.Dialog;
 import javax.sip.InvalidArgumentException;
 import javax.sip.ResponseEvent;
 import javax.sip.SipException;
+import javax.sip.address.SipURI;
+import javax.sip.header.CSeqHeader;
+import javax.sip.header.ViaHeader;
 import javax.sip.message.Request;
+import javax.sip.message.Response;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Component;
 import com.genersoft.iot.vmp.conf.SipConfig;
 import com.genersoft.iot.vmp.gb28181.SipLayer;
+import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorFactory;
 import com.genersoft.iot.vmp.gb28181.transmit.response.ISIPResponseProcessor;
 /**    
@@ -20,20 +30,51 @@ import com.genersoft.iot.vmp.gb28181.transmit.response.ISIPResponseProcessor;
 @Component
 public class InviteResponseProcessor implements ISIPResponseProcessor {
+	private final static Logger logger = LoggerFactory.getLogger(SIPProcessorFactory.class);
+	
 	/**
 	 * 处理invite响应
 	 * 
-	 * @param request
+	 * @param evt
 	 *            响应消息
 	 */ 
 	@Override
 	public void process(ResponseEvent evt, SipLayer layer, SipConfig config) {
 		try {
-			Dialog dialog = evt.getDialog();
-			Request reqAck =dialog.createAck(1L);
-			dialog.sendAck(reqAck);
+			Response response = evt.getResponse();
+			int statusCode = response.getStatusCode();
+			//trying不会回复
+			if(statusCode == Response.TRYING){
+
+			}
+			//成功响应
+			//下发ack
+			if(statusCode == Response.OK){
+				ClientTransaction clientTransaction = evt.getClientTransaction();
+				if(clientTransaction == null){
+					logger.error("回复ACK时，clientTransaction为null >>> {}",response);
+					return;
+				}
+				Dialog clientDialog = clientTransaction.getDialog();
+
+				CSeqHeader clientCSeqHeader = (CSeqHeader) response.getHeader(CSeqHeader.NAME);
+				long cseqId = clientCSeqHeader.getSeqNumber();
+				/*
+				createAck函数，创建的ackRequest，会采用Invite响应的200OK，中的contact字段中的地址，作为目标地址。
+				有的终端传上来的可能还是内网地址，会造成ack发送不出去。接受不到音视频流
+				所以在此处统一替换地址。和响应消息的Via头中的地址保持一致。
+				 */
+				Request ackRequest = clientDialog.createAck(cseqId);
+				SipURI requestURI = (SipURI) ackRequest.getRequestURI();
+				ViaHeader viaHeader = (ViaHeader) response.getHeader(ViaHeader.NAME);
+				requestURI.setHost(viaHeader.getHost());
+				requestURI.setPort(viaHeader.getPort());
+				clientDialog.sendAck(ackRequest);
+			}
 		} catch (InvalidArgumentException | SipException e) {
 			e.printStackTrace();
+		} catch (ParseException e) {
+			e.printStackTrace();
 		}
 	}
@@ -10,6 +10,7 @@ import org.springframework.web.bind.annotation.PathVariable;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
+import com.alibaba.fastjson.JSONObject;
 import com.genersoft.iot.vmp.gb28181.bean.Device;
 import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
 import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
@@ -30,7 +31,7 @@ public class PlaybackController {
 	public ResponseEntity<String> play(@PathVariable String deviceId,@PathVariable String channelId, String startTime,  String endTime){
 		Device device = storager.queryVideoDevice(deviceId);
-		String ssrc = cmder.playStreamCmd(device, channelId);
+		String ssrc = cmder.playbackStreamCmd(device, channelId, startTime, endTime);
 		if (logger.isDebugEnabled()) {
 			logger.debug(String.format("设备预览 API调用，deviceId：%s ，channelId：%s",deviceId, channelId));
@@ -38,7 +39,9 @@ public class PlaybackController {
 		}
 		if(ssrc!=null) {
-			return new ResponseEntity<String>(ssrc,HttpStatus.OK);
+			JSONObject json = new JSONObject();
+			json.put("ssrc", ssrc);
+			return new ResponseEntity<String>(json.toString(),HttpStatus.OK);
 		} else {
 			logger.warn("设备预览API调用失败！");
 			return new ResponseEntity<String>(HttpStatus.INTERNAL_SERVER_ERROR);
@@ -26,7 +26,7 @@ spring:
 server:
     port: 8080
 sip:
-    ip: 10.200.64.63
+    ip: 127.0.0.1
     port: 5060
     # 根据国标6.1.2中规定，domain宜采用ID统一编码的前十位编码。国标附录D中定义前8位为中心编码（由省级、市级、区级、基层编号组成，参照GB/T 2260-2007）
     # 后两位为行业编码，定义参照附录D.3