增加redis通知方式向设备/平台发送报警消息功能

648540858
1 parent 7f965a69
Showing 16 changed files with 486 additions and 25 deletions
pom.xml
src/main/java/com/genersoft/iot/vmp/conf/RedisConfig.java
src/main/java/com/genersoft/iot/vmp/conf/SipConfig.java
src/main/java/com/genersoft/iot/vmp/gb28181/bean/AlarmChannelMessage.java
src/main/java/com/genersoft/iot/vmp/gb28181/bean/DeviceAlarm.java
src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeInfo.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommander.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommanderForPlatform.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/AlarmNotifyMessageHandler.java
src/main/java/com/genersoft/iot/vmp/gb28181/utils/SipUtils.java
src/main/java/com/genersoft/iot/vmp/service/impl/RedisAlarmMsgListener.java
src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java
src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java
src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/alarm/AlarmController.java
@@ -11,7 +11,7 @@
 	<groupId>com.genersoft</groupId>
 	<artifactId>wvp-pro</artifactId>
-	<version>2.1.1</version>
+	<version>2.2.1</version>
 	<name>web video platform</name>
 	<description>国标28181视频平台</description>
 package com.genersoft.iot.vmp.conf;
 import com.genersoft.iot.vmp.common.VideoManagerConstants;
+import com.genersoft.iot.vmp.service.impl.RedisAlarmMsgListener;
 import com.genersoft.iot.vmp.service.impl.RedisGPSMsgListener;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -48,6 +49,9 @@ public class RedisConfig extends CachingConfigurerSupport {
 	@Autowired
 	private RedisGPSMsgListener redisGPSMsgListener;
+	@Autowired
+	private RedisAlarmMsgListener redisAlarmMsgListener;
+
 	@Bean
 	public JedisPool jedisPool() {
 		if (StringUtils.isBlank(password)) {
@@ -93,6 +97,7 @@ public class RedisConfig extends CachingConfigurerSupport {
         RedisMessageListenerContainer container = new RedisMessageListenerContainer();
         container.setConnectionFactory(connectionFactory);
 		container.addMessageListener(redisGPSMsgListener, new PatternTopic(VideoManagerConstants.VM_MSG_GPS));
+		container.addMessageListener(redisAlarmMsgListener, new PatternTopic(VideoManagerConstants.VM_MSG_SUBSCRIBE_ALARM_RECEIVE));
         return container;
     }
@@ -29,7 +29,7 @@ public class SipConfig {
 	Integer registerTimeInterval = 120;
-	private boolean alarm = false;
+	private boolean alarm;
 	public void setIp(String ip) {
 		this.ip = ip;
+package com.genersoft.iot.vmp.gb28181.bean;
+
+/**
+ * 通过redis分发报警消息
+ */
+public class AlarmChannelMessage {
+    /**
+     * 国标编号
+     */
+    private String gbId;
+
+    /**
+     * 报警编号
+     */
+    private int alarmSn;
+
+
+    /**
+     * 报警描述
+     */
+    private String alarmDescription;
+
+    public String getGbId() {
+        return gbId;
+    }
+
+    public void setGbId(String gbId) {
+        this.gbId = gbId;
+    }
+
+    public int getAlarmSn() {
+        return alarmSn;
+    }
+
+    public void setAlarmSn(int alarmSn) {
+        this.alarmSn = alarmSn;
+    }
+
+    public String getAlarmDescription() {
+        return alarmDescription;
+    }
+
+    public void setAlarmDescription(String alarmDescription) {
+        this.alarmDescription = alarmDescription;
+    }
+}
@@ -50,7 +50,30 @@ public class DeviceAlarm {
 	private double latitude;
 	/**
-	 * 报警类型
+	 * 报警类型,
+	 * 报警方式为2时,不携带 AlarmType为默认的报警设备报警,
+	 * 携带 AlarmType取值及对应报警类型如下:
+	 * 		1-视频丢失报警;
+	 * 		2-设备防拆报警;
+	 * 		3-存储设备磁盘满报警;
+	 * 		4-设备高温报警;
+	 * 		5-设备低温报警。
+	 * 报警方式为5时,取值如下:
+	 * 		1-人工视频报警;
+	 * 		2-运动目标检测报警;
+	 * 		3-遗留物检测报警;
+	 * 		4-物体移除检测报警;
+	 * 		5-绊线检测报警;
+	 * 		6-入侵检测报警;
+	 * 		7-逆行检测报警;
+	 * 		8-徘徊检测报警;
+	 * 		9-流量统计报警;
+	 * 		10-密度检测报警;
+	 * 		11-视频异常检测报警;
+	 * 		12-快速移动报警。
+	 * 报警方式为6时,取值下:
+	 * 		1-存储设备磁盘故障报警;
+	 * 		2-存储设备风扇故障报警。
 	 */
 	private String alarmType;
@@ -25,6 +25,9 @@ public class SubscribeInfo {
         this.callId = callIdHeader.getCallId();
     }
+    public SubscribeInfo() {
+    }
+
     private String id;
     private int expires;
     private String callId;
 package com.genersoft.iot.vmp.gb28181.transmit.cmd;
 import com.genersoft.iot.vmp.common.StreamInfo;
-import com.genersoft.iot.vmp.gb28181.bean.Device;
-import com.genersoft.iot.vmp.gb28181.bean.InviteStreamCallback;
+import com.genersoft.iot.vmp.gb28181.bean.*;
 import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
 import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
 import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
@@ -336,4 +335,13 @@ public interface ISIPCommander {
 	 * @param cmdString 前端控制指令串
 	 */
 	boolean dragZoomCmd(Device device, String channelId, String cmdString);
+
+
+	/**
+	 * 向设备发送报警NOTIFY消息， 用于互联结构下，此时将设备当成一个平级平台看待
+	 * @param device 设备
+	 * @param deviceAlarm 报警信息信息
+	 * @return
+	 */
+	boolean sendAlarmMessage(Device device, DeviceAlarm deviceAlarm);
 }
 package com.genersoft.iot.vmp.gb28181.transmit.cmd;
-import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
-import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
-import com.genersoft.iot.vmp.gb28181.bean.RecordInfo;
-import com.genersoft.iot.vmp.gb28181.bean.SubscribeInfo;
+import com.genersoft.iot.vmp.gb28181.bean.*;
 import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
 import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
@@ -76,6 +73,14 @@ public interface ISIPCommanderForPlatform {
     boolean sendNotifyMobilePosition(ParentPlatform parentPlatform, GPSMsgInfo gpsMsgInfo, SubscribeInfo subscribeInfo);
     /**
+     * 向上级回复报警消息
+     * @param parentPlatform 平台信息
+     * @param deviceAlarm 报警信息信息
+     * @return
+     */
+    boolean sendAlarmMessage(ParentPlatform parentPlatform, DeviceAlarm deviceAlarm);
+
+    /**
      * 回复catalog事件-增加/更新
      * @param parentPlatform
      * @param deviceChannels
@@ -5,10 +5,7 @@ import com.genersoft.iot.vmp.common.StreamInfo;
 import com.genersoft.iot.vmp.conf.DynamicTask;
 import com.genersoft.iot.vmp.conf.SipConfig;
 import com.genersoft.iot.vmp.conf.UserSetting;
-import com.genersoft.iot.vmp.gb28181.bean.Device;
-import com.genersoft.iot.vmp.gb28181.bean.InviteStreamCallback;
-import com.genersoft.iot.vmp.gb28181.bean.InviteStreamInfo;
-import com.genersoft.iot.vmp.gb28181.bean.SsrcTransaction;
+import com.genersoft.iot.vmp.gb28181.bean.*;
 import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
 import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
 import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
@@ -23,6 +20,7 @@ import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
 import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
 import gov.nist.javax.sip.SipProviderImpl;
 import gov.nist.javax.sip.SipStackImpl;
+import gov.nist.javax.sip.message.MessageFactoryImpl;
 import gov.nist.javax.sip.message.SIPRequest;
 import gov.nist.javax.sip.stack.SIPDialog;
 import org.slf4j.Logger;
@@ -35,10 +33,7 @@ import org.springframework.util.StringUtils;
 import javax.sip.*;
 import javax.sip.address.SipURI;
-import javax.sip.header.CallIdHeader;
-import javax.sip.header.ContentTypeHeader;
-import javax.sip.header.ExpiresHeader;
-import javax.sip.header.ViaHeader;
+import javax.sip.header.*;
 import javax.sip.message.Request;
 import java.lang.reflect.Field;
 import java.text.ParseException;
@@ -1777,4 +1772,101 @@ public class SIPCommander implements ISIPCommander {
 			e.printStackTrace();
 		}
 	}
+
+	@Override
+	public boolean sendAlarmMessage(Device device, DeviceAlarm deviceAlarm) {
+		if (device == null) {
+			return false;
+		}
+		logger.info("[发送 报警通知] {}/{}->{},{}", device.getDeviceId(), deviceAlarm.getChannelId(),
+				deviceAlarm.getLongitude(), deviceAlarm.getLatitude());
+		try {
+			String characterSet = device.getCharset();
+			StringBuffer deviceStatusXml = new StringBuffer(600);
+			deviceStatusXml.append("<?xml version=\"1.0\" encoding=\"" + characterSet + "\"?>\r\n");
+			deviceStatusXml.append("<Notify>\r\n");
+			deviceStatusXml.append("<CmdType>Alarm</CmdType>\r\n");
+			deviceStatusXml.append("<SN>" + (int)((Math.random()*9+1)*100000) + "</SN>\r\n");
+			deviceStatusXml.append("<DeviceID>" + deviceAlarm.getChannelId() + "</DeviceID>\r\n");
+			deviceStatusXml.append("<AlarmPriority>" + deviceAlarm.getAlarmPriority() + "</AlarmPriority>\r\n");
+			deviceStatusXml.append("<AlarmMethod>" + deviceAlarm.getAlarmMethod() + "</AlarmMethod>\r\n");
+			deviceStatusXml.append("<AlarmTime>" + deviceAlarm.getAlarmTime() + "</AlarmTime>\r\n");
+			deviceStatusXml.append("<AlarmDescription>" + deviceAlarm.getAlarmDescription() + "</AlarmDescription>\r\n");
+			deviceStatusXml.append("<Longitude>" + deviceAlarm.getLongitude() + "</Longitude>\r\n");
+			deviceStatusXml.append("<Latitude>" + deviceAlarm.getLatitude() + "</Latitude>\r\n");
+			deviceStatusXml.append("<info>\r\n");
+			deviceStatusXml.append("<AlarmType>" + deviceAlarm.getAlarmType() + "</AlarmType>\r\n");
+			deviceStatusXml.append("</info>\r\n");
+			deviceStatusXml.append("</Notify>\r\n");
+
+			CallIdHeader callIdHeader = device.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()
+					: udpSipProvider.getNewCallId();
+			String tm = Long.toString(System.currentTimeMillis());
+			Request request = headerProvider.createMessageRequest(device, deviceStatusXml.toString(), "z9hG4bK-ViaPtz-" + tm, "FromPtz" + tm, null, callIdHeader);
+			transmitRequest(device, request);
+
+
+		} catch (SipException | ParseException  e) {
+			e.printStackTrace();
+			return false;
+		} catch (InvalidArgumentException e) {
+			throw new RuntimeException(e);
+		}
+		return true;
+	}
+
+	private void sendNotify(Device device, String catalogXmlContent,
+							SubscribeInfo subscribeInfo, SipSubscribe.Event errorEvent,  SipSubscribe.Event okEvent )
+			throws NoSuchFieldException, IllegalAccessException, SipException, ParseException {
+		MessageFactoryImpl messageFactory = (MessageFactoryImpl) sipFactory.createMessageFactory();
+		String characterSet = device.getCharset();
+		// 设置编码， 防止中文乱码
+		messageFactory.setDefaultContentEncodingCharset(characterSet);
+		Dialog dialog  = subscribeInfo.getDialog();
+		if (dialog == null || !dialog.getState().equals(DialogState.CONFIRMED)) return;
+		SIPRequest notifyRequest = (SIPRequest)dialog.createRequest(Request.NOTIFY);
+		ContentTypeHeader contentTypeHeader = sipFactory.createHeaderFactory().createContentTypeHeader("Application", "MANSCDP+xml");
+		notifyRequest.setContent(catalogXmlContent, contentTypeHeader);
+
+		SubscriptionStateHeader subscriptionState = sipFactory.createHeaderFactory()
+				.createSubscriptionStateHeader(SubscriptionStateHeader.ACTIVE);
+		notifyRequest.addHeader(subscriptionState);
+
+		EventHeader event = sipFactory.createHeaderFactory().createEventHeader(subscribeInfo.getEventType());
+		if (subscribeInfo.getEventId() != null) {
+			event.setEventId(subscribeInfo.getEventId());
+		}
+		notifyRequest.addHeader(event);
+
+		SipURI sipURI = (SipURI) notifyRequest.getRequestURI();
+		if (subscribeInfo.getTransaction() != null) {
+			SIPRequest request = (SIPRequest) subscribeInfo.getTransaction().getRequest();
+			sipURI.setHost(request.getRemoteAddress().getHostAddress());
+			sipURI.setPort(request.getRemotePort());
+		}else {
+			sipURI.setHost(device.getIp());
+			sipURI.setPort(device.getPort());
+		}
+
+		ClientTransaction transaction = null;
+		if ("TCP".equals(device.getTransport())) {
+			transaction = tcpSipProvider.getNewClientTransaction(notifyRequest);
+		} else if ("UDP".equals(device.getTransport())) {
+			transaction = udpSipProvider.getNewClientTransaction(notifyRequest);
+		}
+		// 添加错误订阅
+		if (errorEvent != null) {
+			sipSubscribe.addErrorSubscribe(subscribeInfo.getCallId(), errorEvent);
+		}
+		// 添加订阅
+		if (okEvent != null) {
+			sipSubscribe.addOkSubscribe(subscribeInfo.getCallId(), okEvent);
+		}
+		if (transaction == null) {
+			logger.error("平台{}的Transport错误：{}",device.getDeviceId(), device.getTransport());
+			return;
+		}
+		dialog.sendRequest(transaction);
+
+	}
 }
@@ -436,6 +436,48 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
     }
     @Override
+    public boolean sendAlarmMessage(ParentPlatform parentPlatform, DeviceAlarm deviceAlarm) {
+        if (parentPlatform == null) {
+            return false;
+        }
+        logger.info("[发送 报警订阅] {}/{}->{},{}", parentPlatform.getServerGBId(), deviceAlarm.getChannelId(),
+                deviceAlarm.getLongitude(), deviceAlarm.getLatitude());
+        try {
+            String characterSet = parentPlatform.getCharacterSet();
+            StringBuffer deviceStatusXml = new StringBuffer(600);
+            deviceStatusXml.append("<?xml version=\"1.0\" encoding=\"" + characterSet + "\"?>\r\n");
+            deviceStatusXml.append("<Notify>\r\n");
+            deviceStatusXml.append("<CmdType>Alarm</CmdType>\r\n");
+            deviceStatusXml.append("<SN>" + (int)((Math.random()*9+1)*100000) + "</SN>\r\n");
+            deviceStatusXml.append("<DeviceID>" + deviceAlarm.getChannelId() + "</DeviceID>\r\n");
+            deviceStatusXml.append("<AlarmPriority>" + deviceAlarm.getAlarmPriority() + "</AlarmPriority>\r\n");
+            deviceStatusXml.append("<AlarmMethod>" + deviceAlarm.getAlarmMethod() + "</AlarmMethod>\r\n");
+            deviceStatusXml.append("<AlarmTime>" + deviceAlarm.getAlarmTime() + "</AlarmTime>\r\n");
+            deviceStatusXml.append("<AlarmDescription>" + deviceAlarm.getAlarmDescription() + "</AlarmDescription>\r\n");
+            deviceStatusXml.append("<Longitude>" + deviceAlarm.getLongitude() + "</Longitude>\r\n");
+            deviceStatusXml.append("<Latitude>" + deviceAlarm.getLatitude() + "</Latitude>\r\n");
+            deviceStatusXml.append("<info>\r\n");
+            deviceStatusXml.append("<AlarmType>" + deviceAlarm.getAlarmType() + "</AlarmType>\r\n");
+            deviceStatusXml.append("</info>\r\n");
+            deviceStatusXml.append("</Notify>\r\n");
+
+            CallIdHeader callIdHeader = parentPlatform.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()
+                    : udpSipProvider.getNewCallId();
+
+            String tm = Long.toString(System.currentTimeMillis());
+            Request request = headerProviderPlarformProvider.createMessageRequest(parentPlatform, deviceStatusXml.toString(), "FromPtz" + tm, callIdHeader);
+            transmitRequest(parentPlatform, request);
+
+        } catch (SipException | ParseException  e) {
+            e.printStackTrace();
+            return false;
+        } catch (InvalidArgumentException e) {
+            throw new RuntimeException(e);
+        }
+        return true;
+    }
+
+    @Override
     public boolean sendNotifyForCatalogAddOrUpdate(String type, ParentPlatform parentPlatform, List<DeviceChannel> deviceChannels, SubscribeInfo subscribeInfo, Integer index) {
         if (parentPlatform == null || deviceChannels == null || deviceChannels.size() == 0 || subscribeInfo == null) {
             return false;
@@ -495,11 +537,16 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
             event.setEventId(subscribeInfo.getEventId());
         }
         notifyRequest.addHeader(event);
-
         SipURI sipURI = (SipURI) notifyRequest.getRequestURI();
-        SIPRequest request = (SIPRequest) subscribeInfo.getTransaction().getRequest();
-        sipURI.setHost(request.getRemoteAddress().getHostAddress());
-        sipURI.setPort(request.getRemotePort());
+        if (subscribeInfo.getTransaction() != null) {
+            SIPRequest request = (SIPRequest) subscribeInfo.getTransaction().getRequest();
+            sipURI.setHost(request.getRemoteAddress().getHostAddress());
+            sipURI.setPort(request.getRemotePort());
+        }else {
+            sipURI.setHost(parentPlatform.getServerIP());
+            sipURI.setPort(parentPlatform.getServerPort());
+        }
+
         ClientTransaction transaction = null;
         if ("TCP".equals(parentPlatform.getTransport())) {
             transaction = tcpSipProvider.getNewClientTransaction(notifyRequest);
 package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.notify.cmd;
+import com.alibaba.fastjson.JSON;
 import com.genersoft.iot.vmp.conf.SipConfig;
 import com.genersoft.iot.vmp.conf.UserSetting;
 import com.genersoft.iot.vmp.gb28181.bean.*;
@@ -11,6 +12,7 @@ import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.notify.
 import com.genersoft.iot.vmp.gb28181.utils.Coordtransform;
 import com.genersoft.iot.vmp.gb28181.utils.NumericUtil;
 import com.genersoft.iot.vmp.service.IDeviceAlarmService;
+import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
 import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
 import com.genersoft.iot.vmp.utils.GpsUtil;
 import org.dom4j.Element;
@@ -21,7 +23,12 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 import org.springframework.util.StringUtils;
+import javax.sip.InvalidArgumentException;
 import javax.sip.RequestEvent;
+import javax.sip.SipException;
+import javax.sip.message.Response;
+
+import java.text.ParseException;
 import static com.genersoft.iot.vmp.gb28181.utils.XmlUtil.*;
@@ -47,6 +54,9 @@ public class AlarmNotifyMessageHandler extends SIPRequestProcessorParent impleme
     private IVideoManagerStorage storager;
     @Autowired
+    private IRedisCatchStorage redisCatchStorage;
+
+    @Autowired
     private IDeviceAlarmService deviceAlarmService;
     @Autowired
@@ -59,11 +69,22 @@ public class AlarmNotifyMessageHandler extends SIPRequestProcessorParent impleme
     @Override
     public void handForDevice(RequestEvent evt, Device device, Element rootElement) {
-        if (!sipConfig.isAlarm()) {
-            return;
+        logger.info("收到来自设备[{}]的报警通知", device.getDeviceId());
+        // 回复200 OK
+        try {
+            responseAck(evt, Response.OK);
+        } catch (SipException e) {
+            throw new RuntimeException(e);
+        } catch (InvalidArgumentException e) {
+            throw new RuntimeException(e);
+        } catch (ParseException e) {
+            throw new RuntimeException(e);
         }
+
         Element deviceIdElement = rootElement.element("DeviceID");
         String channelId = deviceIdElement.getText().toString();
+
+
         DeviceAlarm deviceAlarm = new DeviceAlarm();
         deviceAlarm.setDeviceId(device.getDeviceId());
         deviceAlarm.setChannelId(channelId);
@@ -111,9 +132,24 @@ public class AlarmNotifyMessageHandler extends SIPRequestProcessorParent impleme
                 deviceAlarm.setAlarmType(getText(rootElement.element("Info"), "AlarmType"));
             }
         }
+
+        if (channelId.equals(sipConfig.getId())) {
+            // 发送给平台的报警信息。 发送redis通知
+            AlarmChannelMessage alarmChannelMessage = new AlarmChannelMessage();
+            alarmChannelMessage.setAlarmSn(Integer.parseInt(deviceAlarm.getAlarmMethod()));
+            alarmChannelMessage.setAlarmDescription(deviceAlarm.getAlarmDescription());
+            alarmChannelMessage.setGbId(channelId);
+            redisCatchStorage.sendAlarmMsg(alarmChannelMessage);
+
+            return;
+        }
+
         logger.debug("存储报警信息、报警分类");
         // 存储报警信息、报警分类
-        deviceAlarmService.add(deviceAlarm);
+        if (sipConfig.isAlarm()) {
+            deviceAlarmService.add(deviceAlarm);
+        }
+
         if (offLineDetector.isOnline(device.getDeviceId())) {
             publisher.deviceAlarmEventPublish(deviceAlarm);
@@ -121,7 +157,59 @@ public class AlarmNotifyMessageHandler extends SIPRequestProcessorParent impleme
     }
     @Override
-    public void handForPlatform(RequestEvent evt, ParentPlatform parentPlatform, Element element) {
+    public void handForPlatform(RequestEvent evt, ParentPlatform parentPlatform, Element rootElement) {
+        logger.info("收到来自平台[{}]的报警通知", parentPlatform.getServerGBId());
+        // 回复200 OK
+        try {
+            responseAck(evt, Response.OK);
+        } catch (SipException e) {
+            throw new RuntimeException(e);
+        } catch (InvalidArgumentException e) {
+            throw new RuntimeException(e);
+        } catch (ParseException e) {
+            throw new RuntimeException(e);
+        }
+        Element deviceIdElement = rootElement.element("DeviceID");
+        String channelId = deviceIdElement.getText().toString();
+
+
+        DeviceAlarm deviceAlarm = new DeviceAlarm();
+        deviceAlarm.setDeviceId(parentPlatform.getServerGBId());
+        deviceAlarm.setChannelId(channelId);
+        deviceAlarm.setAlarmPriority(getText(rootElement, "AlarmPriority"));
+        deviceAlarm.setAlarmMethod(getText(rootElement, "AlarmMethod"));
+        deviceAlarm.setAlarmTime(getText(rootElement, "AlarmTime"));
+        if (getText(rootElement, "AlarmDescription") == null) {
+            deviceAlarm.setAlarmDescription("");
+        } else {
+            deviceAlarm.setAlarmDescription(getText(rootElement, "AlarmDescription"));
+        }
+        if (NumericUtil.isDouble(getText(rootElement, "Longitude"))) {
+            deviceAlarm.setLongitude(Double.parseDouble(getText(rootElement, "Longitude")));
+        } else {
+            deviceAlarm.setLongitude(0.00);
+        }
+        if (NumericUtil.isDouble(getText(rootElement, "Latitude"))) {
+            deviceAlarm.setLatitude(Double.parseDouble(getText(rootElement, "Latitude")));
+        } else {
+            deviceAlarm.setLatitude(0.00);
+        }
+
+        if (!StringUtils.isEmpty(deviceAlarm.getAlarmMethod())) {
+
+            if (deviceAlarm.getAlarmMethod().equals("5")) {
+                deviceAlarm.setAlarmType(getText(rootElement.element("Info"), "AlarmType"));
+            }
+        }
+        if (channelId.equals(parentPlatform.getDeviceGBId())) {
+            // 发送给平台的报警信息。 发送redis通知
+            AlarmChannelMessage alarmChannelMessage = new AlarmChannelMessage();
+            alarmChannelMessage.setAlarmSn(Integer.parseInt(deviceAlarm.getAlarmMethod()));
+            alarmChannelMessage.setAlarmDescription(deviceAlarm.getAlarmDescription());
+            alarmChannelMessage.setGbId(channelId);
+            redisCatchStorage.sendAlarmMsg(alarmChannelMessage);
+            return;
+        }
     }
 }
@@ -25,6 +25,7 @@ public class SipUtils {
      * */
     public static String getChannelIdFromHeader(Request request) {
         Header subject = request.getHeader("subject");
+        if (subject == null) return null;
         return ((Subject) subject).getSubject().split(":")[0];
     }
+package com.genersoft.iot.vmp.service.impl;
+
+import com.alibaba.fastjson.JSON;
+import com.genersoft.iot.vmp.gb28181.bean.*;
+import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
+import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
+import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
+import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
+import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
+import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.connection.Message;
+import org.springframework.data.redis.connection.MessageListener;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.stereotype.Component;
+
+import java.text.SimpleDateFormat;
+
+@Component
+public class RedisAlarmMsgListener implements MessageListener {
+
+    private final static Logger logger = LoggerFactory.getLogger(RedisAlarmMsgListener.class);
+
+    @Autowired
+    private ISIPCommander commander;
+
+    @Autowired
+    private ISIPCommanderForPlatform commanderForPlatform;
+
+    @Autowired
+    private IVideoManagerStorage storage;
+
+    private final SimpleDateFormat formatForGB = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss");
+
+    @Override
+    public void onMessage(Message message, byte[] bytes) {
+        logger.info("收到来自REDIS的ALARM通知： {}", new String(message.getBody()));
+        AlarmChannelMessage alarmChannelMessage = JSON.parseObject(message.getBody(), AlarmChannelMessage.class);
+        if (alarmChannelMessage == null) {
+            logger.warn("[REDIS的ALARM通知]消息解析失败");
+            return;
+        }
+        String gbId = alarmChannelMessage.getGbId();
+        Device device = storage.queryVideoDevice(gbId);
+        ParentPlatform platform = storage.queryParentPlatByServerGBId(gbId);
+
+        DeviceAlarm deviceAlarm = new DeviceAlarm();
+        deviceAlarm.setChannelId(gbId);
+        deviceAlarm.setAlarmDescription(alarmChannelMessage.getAlarmDescription());
+        deviceAlarm.setAlarmMethod("" + alarmChannelMessage.getAlarmSn());
+        deviceAlarm.setAlarmPriority("1");
+        deviceAlarm.setAlarmTime(formatForGB.format(System.currentTimeMillis()));
+        deviceAlarm.setAlarmType("1");
+        deviceAlarm.setLongitude(0);
+        deviceAlarm.setLatitude(0);
+
+
+        if (device != null && platform == null) {
+            commander.sendAlarmMessage(device, deviceAlarm);
+        }else if (device == null && platform != null){
+            commanderForPlatform.sendAlarmMessage(platform, deviceAlarm);
+        }else {
+           logger.warn("无法确定" + gbId + "是平台还是设备");
+        }
+    }
+}
@@ -140,6 +140,12 @@ public interface IRedisCatchStorage {
     void sendStreamChangeMsg(String type, JSONObject jsonObject);
     /**
+     * 发送报警消息
+     * @param msg 消息内容
+     */
+    void sendAlarmMsg(AlarmChannelMessage msg);
+
+    /**
      * 添加流信息到redis
      * @param mediaServerItem
      * @param app
@@ -639,4 +639,10 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
         redis.convertAndSend(key, (JSONObject)JSON.toJSON(msg));
     }
+    @Override
+    public void sendAlarmMsg(AlarmChannelMessage msg) {
+        String key = VideoManagerConstants.VM_MSG_SUBSCRIBE_ALARM;
+        logger.info("[redis 报警通知] {}: {}", key, JSON.toJSON(msg));
+        redis.convertAndSend(key, (JSONObject)JSON.toJSON(msg));
+    }
 }
 package com.genersoft.iot.vmp.vmanager.gb28181.alarm;
+import com.genersoft.iot.vmp.gb28181.bean.Device;
 import com.genersoft.iot.vmp.gb28181.bean.DeviceAlarm;
+import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
+import com.genersoft.iot.vmp.gb28181.bean.SubscribeInfo;
+import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
+import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
 import com.genersoft.iot.vmp.service.IDeviceAlarmService;
 import com.genersoft.iot.vmp.service.IGbStreamService;
+import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
 import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
 import com.github.pagehelper.PageInfo;
 import io.swagger.annotations.Api;
@@ -31,7 +37,17 @@ public class AlarmController {
     @Autowired
     private IDeviceAlarmService deviceAlarmService;
+    @Autowired
+    private ISIPCommander commander;
+
+    @Autowired
+    private ISIPCommanderForPlatform commanderForPlatform;
+
+    @Autowired
+    private IVideoManagerStorage storage;
+
     private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+    private SimpleDateFormat formatForGB = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss");
     /**
      *  分页查询报警
@@ -133,5 +149,51 @@ public class AlarmController {
         return new ResponseEntity<WVPResult<String>>(wvpResult, HttpStatus.OK);
     }
+    /**
+     *  测试向上级/设备发送模拟报警通知
+     *
+     * @param deviceId 报警id
+     * @return
+     */
+    @ApiOperation("测试向上级/设备发送模拟报警通知")
+    @GetMapping("/test/notify/alarm")
+    @ApiImplicitParams({
+            @ApiImplicitParam(name="deviceId", value = "deviceId", required = true ,dataTypeClass = Integer.class)
+    })
+    public ResponseEntity<WVPResult<String>> delete(
+            @RequestParam(required = false) String deviceId
+    ) {
+        if (StringUtils.isEmpty(deviceId)) {
+            return new ResponseEntity<>(HttpStatus.NOT_FOUND);
+        }
+        Device device = storage.queryVideoDevice(deviceId);
+        ParentPlatform platform = storage.queryParentPlatByServerGBId(deviceId);
+        DeviceAlarm deviceAlarm = new DeviceAlarm();
+        deviceAlarm.setChannelId(deviceId);
+        deviceAlarm.setAlarmDescription("test");
+        deviceAlarm.setAlarmMethod("1");
+        deviceAlarm.setAlarmPriority("1");
+        deviceAlarm.setAlarmTime(formatForGB.format(System.currentTimeMillis()));
+        deviceAlarm.setAlarmType("1");
+        deviceAlarm.setLongitude(115.33333);
+        deviceAlarm.setLatitude(39.33333);
+
+        if (device != null && platform == null) {
+            commander.sendAlarmMessage(device, deviceAlarm);
+        }else if (device == null && platform != null){
+            commanderForPlatform.sendAlarmMessage(platform, deviceAlarm);
+        }else {
+            WVPResult wvpResult = new WVPResult();
+            wvpResult.setCode(0);
+            wvpResult.setMsg("无法确定" + deviceId + "是平台还是设备");
+            return new ResponseEntity<WVPResult<String>>(wvpResult, HttpStatus.OK);
+        }
+
+        WVPResult wvpResult = new WVPResult();
+        wvpResult.setCode(0);
+        wvpResult.setMsg("success");
+        return new ResponseEntity<WVPResult<String>>(wvpResult, HttpStatus.OK);
+    }
+
 }