diff --git a/src/main/java/com/genersoft/iot/vmp/conf/UserSetting.java b/src/main/java/com/genersoft/iot/vmp/conf/UserSetting.java index 017b39db..cad6e69d 100644 --- a/src/main/java/com/genersoft/iot/vmp/conf/UserSetting.java +++ b/src/main/java/com/genersoft/iot/vmp/conf/UserSetting.java @@ -31,6 +31,8 @@ public class UserSetting { private Boolean logInDatebase = Boolean.TRUE; + private Boolean usePushingAsStatus = Boolean.TRUE; + private String serverId = "000000"; private String thirdPartyGBIdReg = "[\\s\\S]*"; @@ -136,4 +138,12 @@ public class UserSetting { public void setPlatformPlayTimeout(int platformPlayTimeout) { this.platformPlayTimeout = platformPlayTimeout; } + + public Boolean isUsePushingAsStatus() { + return usePushingAsStatus; + } + + public void setUsePushingAsStatus(Boolean usePushingAsStatus) { + this.usePushingAsStatus = usePushingAsStatus; + } } diff --git a/src/main/java/com/genersoft/iot/vmp/conf/redis/RedisConfig.java b/src/main/java/com/genersoft/iot/vmp/conf/redis/RedisConfig.java index 0b653cf5..e412f719 100644 --- a/src/main/java/com/genersoft/iot/vmp/conf/redis/RedisConfig.java +++ b/src/main/java/com/genersoft/iot/vmp/conf/redis/RedisConfig.java @@ -3,7 +3,7 @@ package com.genersoft.iot.vmp.conf.redis; import com.alibaba.fastjson.parser.ParserConfig; import com.genersoft.iot.vmp.common.VideoManagerConstants; -import com.genersoft.iot.vmp.service.impl.*; +import com.genersoft.iot.vmp.service.redisMsg.*; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.cache.annotation.CachingConfigurerSupport; import org.springframework.context.annotation.Bean; diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SipMsgInfo.java b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SipMsgInfo.java new file mode 100644 index 00000000..302539b6 --- /dev/null +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SipMsgInfo.java @@ -0,0 +1,56 @@ +package com.genersoft.iot.vmp.gb28181.bean; + +import org.dom4j.Element; + +import javax.sip.RequestEvent; + +public class SipMsgInfo { + private RequestEvent evt; + private Device device; + private ParentPlatform platform; + private Element rootElement; + + public SipMsgInfo(RequestEvent evt, Device device, Element rootElement) { + this.evt = evt; + this.device = device; + this.rootElement = rootElement; + } + + public SipMsgInfo(RequestEvent evt, ParentPlatform platform, Element rootElement) { + this.evt = evt; + this.platform = platform; + this.rootElement = rootElement; + } + + public RequestEvent getEvt() { + return evt; + } + + public void setEvt(RequestEvent evt) { + this.evt = evt; + } + + public Device getDevice() { + return device; + } + + public void setDevice(Device device) { + this.device = device; + } + + public ParentPlatform getPlatform() { + return platform; + } + + public void setPlatform(ParentPlatform platform) { + this.platform = platform; + } + + public Element getRootElement() { + return rootElement; + } + + public void setRootElement(Element rootElement) { + this.rootElement = rootElement; + } +} diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java index f8c3abfb..33cc6e6d 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java @@ -14,18 +14,15 @@ import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; import com.genersoft.iot.vmp.service.IMediaServerService; import com.genersoft.iot.vmp.service.bean.RequestPushStreamMsg; -import com.genersoft.iot.vmp.service.impl.RedisGbPlayMsgListener; +import com.genersoft.iot.vmp.service.redisMsg.RedisGbPlayMsgListener; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage; -import com.genersoft.iot.vmp.utils.SerializeUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; -import javax.sip.Dialog; -import javax.sip.DialogState; import javax.sip.RequestEvent; import javax.sip.address.SipURI; import javax.sip.header.CallIdHeader; diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java index 6ff5c0f7..ff3c78f0 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java @@ -22,8 +22,8 @@ import com.genersoft.iot.vmp.service.IStreamProxyService; import com.genersoft.iot.vmp.service.IStreamPushService; import com.genersoft.iot.vmp.service.bean.MessageForPushChannel; import com.genersoft.iot.vmp.service.bean.SSRCInfo; -import com.genersoft.iot.vmp.service.impl.RedisGbPlayMsgListener; -import com.genersoft.iot.vmp.service.impl.RedisPushStreamResponseListener; +import com.genersoft.iot.vmp.service.redisMsg.RedisGbPlayMsgListener; +import com.genersoft.iot.vmp.service.redisMsg.RedisPushStreamResponseListener; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage; import com.genersoft.iot.vmp.utils.DateUtil; diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/AlarmNotifyMessageHandler.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/AlarmNotifyMessageHandler.java index add32d1b..da2cb9c3 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/AlarmNotifyMessageHandler.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/AlarmNotifyMessageHandler.java @@ -21,6 +21,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.stereotype.Component; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; @@ -31,6 +33,7 @@ import javax.sip.SipException; import javax.sip.message.Response; import java.text.ParseException; +import java.util.concurrent.ConcurrentLinkedQueue; import static com.genersoft.iot.vmp.gb28181.utils.XmlUtil.*; @@ -67,6 +70,15 @@ public class AlarmNotifyMessageHandler extends SIPRequestProcessorParent impleme @Autowired private IDeviceChannelService deviceChannelService; + private boolean taskQueueHandlerRun = false; + + private final ConcurrentLinkedQueue taskQueue = new ConcurrentLinkedQueue<>(); + + @Qualifier("taskExecutor") + @Autowired + private ThreadPoolTaskExecutor taskExecutor; + + @Override public void afterPropertiesSet() throws Exception { notifyMessageHandler.addHandler(cmdType, this); @@ -75,114 +87,127 @@ public class AlarmNotifyMessageHandler extends SIPRequestProcessorParent impleme @Override public void handForDevice(RequestEvent evt, Device device, Element rootElement) { logger.info("[收到报警通知]设备：{}", device.getDeviceId()); - // 回复200 OK - try { - responseAck(getServerTransaction(evt), Response.OK); - } catch (SipException | InvalidArgumentException | ParseException e) { - logger.error("[收到报警通知], 回复200OK失败", e); - } - Element deviceIdElement = rootElement.element("DeviceID"); - String channelId = deviceIdElement.getText().toString(); + taskQueue.offer(new SipMsgInfo(evt, device, rootElement)); + if (!taskQueueHandlerRun) { + taskQueueHandlerRun = true; + taskExecutor.execute(() -> { + while (!taskQueue.isEmpty()) { + SipMsgInfo sipMsgInfo = taskQueue.poll(); + // 回复200 OK + try { + responseAck(getServerTransaction(sipMsgInfo.getEvt()), Response.OK); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[收到报警通知], 回复200OK失败", e); + } - DeviceAlarm deviceAlarm = new DeviceAlarm(); - deviceAlarm.setCreateTime(DateUtil.getNow()); - deviceAlarm.setDeviceId(device.getDeviceId()); - deviceAlarm.setChannelId(channelId); - deviceAlarm.setAlarmPriority(getText(rootElement, "AlarmPriority")); - deviceAlarm.setAlarmMethod(getText(rootElement, "AlarmMethod")); - String alarmTime = XmlUtil.getText(rootElement, "AlarmTime"); - if (alarmTime == null) { - return; - } - deviceAlarm.setAlarmTime(DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(alarmTime)); - String alarmDescription = getText(rootElement, "AlarmDescription"); - if (alarmDescription == null) { - deviceAlarm.setAlarmDescription(""); - } else { - deviceAlarm.setAlarmDescription(alarmDescription); - } - String longitude = getText(rootElement, "Longitude"); - if (longitude != null && NumericUtil.isDouble(longitude)) { - deviceAlarm.setLongitude(Double.parseDouble(longitude)); - } else { - deviceAlarm.setLongitude(0.00); - } - String latitude = getText(rootElement, "Latitude"); - if (latitude != null && NumericUtil.isDouble(latitude)) { - deviceAlarm.setLatitude(Double.parseDouble(latitude)); - } else { - deviceAlarm.setLatitude(0.00); - } + Element deviceIdElement = sipMsgInfo.getRootElement().element("DeviceID"); + String channelId = deviceIdElement.getText().toString(); - if (!ObjectUtils.isEmpty(deviceAlarm.getAlarmMethod())) { - if ( deviceAlarm.getAlarmMethod().contains(DeviceAlarmMethod.GPS.getVal() + "")) { - MobilePosition mobilePosition = new MobilePosition(); - mobilePosition.setCreateTime(DateUtil.getNow()); - mobilePosition.setDeviceId(deviceAlarm.getDeviceId()); - mobilePosition.setTime(deviceAlarm.getAlarmTime()); - mobilePosition.setLongitude(deviceAlarm.getLongitude()); - mobilePosition.setLatitude(deviceAlarm.getLatitude()); - mobilePosition.setReportSource("GPS Alarm"); + DeviceAlarm deviceAlarm = new DeviceAlarm(); + deviceAlarm.setCreateTime(DateUtil.getNow()); + deviceAlarm.setDeviceId(sipMsgInfo.getDevice().getDeviceId()); + deviceAlarm.setChannelId(channelId); + deviceAlarm.setAlarmPriority(getText(sipMsgInfo.getRootElement(), "AlarmPriority")); + deviceAlarm.setAlarmMethod(getText(sipMsgInfo.getRootElement(), "AlarmMethod")); + String alarmTime = XmlUtil.getText(sipMsgInfo.getRootElement(), "AlarmTime"); + if (alarmTime == null) { + return; + } + deviceAlarm.setAlarmTime(DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(alarmTime)); + String alarmDescription = getText(sipMsgInfo.getRootElement(), "AlarmDescription"); + if (alarmDescription == null) { + deviceAlarm.setAlarmDescription(""); + } else { + deviceAlarm.setAlarmDescription(alarmDescription); + } + String longitude = getText(sipMsgInfo.getRootElement(), "Longitude"); + if (longitude != null && NumericUtil.isDouble(longitude)) { + deviceAlarm.setLongitude(Double.parseDouble(longitude)); + } else { + deviceAlarm.setLongitude(0.00); + } + String latitude = getText(sipMsgInfo.getRootElement(), "Latitude"); + if (latitude != null && NumericUtil.isDouble(latitude)) { + deviceAlarm.setLatitude(Double.parseDouble(latitude)); + } else { + deviceAlarm.setLatitude(0.00); + } - // 更新device channel 的经纬度 - DeviceChannel deviceChannel = new DeviceChannel(); - deviceChannel.setDeviceId(device.getDeviceId()); - deviceChannel.setChannelId(channelId); - deviceChannel.setLongitude(mobilePosition.getLongitude()); - deviceChannel.setLatitude(mobilePosition.getLatitude()); - deviceChannel.setGpsTime(mobilePosition.getTime()); + if (!ObjectUtils.isEmpty(deviceAlarm.getAlarmMethod())) { + if ( deviceAlarm.getAlarmMethod().contains(DeviceAlarmMethod.GPS.getVal() + "")) { + MobilePosition mobilePosition = new MobilePosition(); + mobilePosition.setCreateTime(DateUtil.getNow()); + mobilePosition.setDeviceId(deviceAlarm.getDeviceId()); + mobilePosition.setTime(deviceAlarm.getAlarmTime()); + mobilePosition.setLongitude(deviceAlarm.getLongitude()); + mobilePosition.setLatitude(deviceAlarm.getLatitude()); + mobilePosition.setReportSource("GPS Alarm"); - deviceChannel = deviceChannelService.updateGps(deviceChannel, device); + // 更新device channel 的经纬度 + DeviceChannel deviceChannel = new DeviceChannel(); + deviceChannel.setDeviceId(sipMsgInfo.getDevice().getDeviceId()); + deviceChannel.setChannelId(channelId); + deviceChannel.setLongitude(mobilePosition.getLongitude()); + deviceChannel.setLatitude(mobilePosition.getLatitude()); + deviceChannel.setGpsTime(mobilePosition.getTime()); - mobilePosition.setLongitudeWgs84(deviceChannel.getLongitudeWgs84()); - mobilePosition.setLatitudeWgs84(deviceChannel.getLatitudeWgs84()); - mobilePosition.setLongitudeGcj02(deviceChannel.getLongitudeGcj02()); - mobilePosition.setLatitudeGcj02(deviceChannel.getLatitudeGcj02()); + deviceChannel = deviceChannelService.updateGps(deviceChannel, sipMsgInfo.getDevice()); - if (userSetting.getSavePositionHistory()) { - storager.insertMobilePosition(mobilePosition); + mobilePosition.setLongitudeWgs84(deviceChannel.getLongitudeWgs84()); + mobilePosition.setLatitudeWgs84(deviceChannel.getLatitudeWgs84()); + mobilePosition.setLongitudeGcj02(deviceChannel.getLongitudeGcj02()); + mobilePosition.setLatitudeGcj02(deviceChannel.getLatitudeGcj02()); + + if (userSetting.getSavePositionHistory()) { + storager.insertMobilePosition(mobilePosition); + } + storager.updateChannelPosition(deviceChannel); + + // 发送redis消息。 通知位置信息的变化 + JSONObject jsonObject = new JSONObject(); + jsonObject.put("time", mobilePosition.getTime()); + jsonObject.put("serial", deviceChannel.getDeviceId()); + jsonObject.put("code", deviceChannel.getChannelId()); + jsonObject.put("longitude", mobilePosition.getLongitude()); + jsonObject.put("latitude", mobilePosition.getLatitude()); + jsonObject.put("altitude", mobilePosition.getAltitude()); + jsonObject.put("direction", mobilePosition.getDirection()); + jsonObject.put("speed", mobilePosition.getSpeed()); + redisCatchStorage.sendMobilePositionMsg(jsonObject); + } + } + if (!ObjectUtils.isEmpty(deviceAlarm.getDeviceId())) { + if (deviceAlarm.getAlarmMethod().contains(DeviceAlarmMethod.Video.getVal() + "")) { + deviceAlarm.setAlarmType(getText(sipMsgInfo.getRootElement().element("Info"), "AlarmType")); + } + } + + if ("7".equals(deviceAlarm.getAlarmMethod()) ) { + // 发送给平台的报警信息。 发送redis通知 + AlarmChannelMessage alarmChannelMessage = new AlarmChannelMessage(); + alarmChannelMessage.setAlarmSn(Integer.parseInt(deviceAlarm.getAlarmMethod())); + alarmChannelMessage.setAlarmDescription(deviceAlarm.getAlarmDescription()); + alarmChannelMessage.setGbId(channelId); + redisCatchStorage.sendAlarmMsg(alarmChannelMessage); + return; + } + + logger.debug("存储报警信息、报警分类"); + // 存储报警信息、报警分类 + if (sipConfig.isAlarm()) { + deviceAlarmService.add(deviceAlarm); + } + logger.info("[收到报警通知]内容：{}", JSONObject.toJSON(deviceAlarm)); + if (redisCatchStorage.deviceIsOnline(sipMsgInfo.getDevice().getDeviceId())) { + publisher.deviceAlarmEventPublish(deviceAlarm); + } } - storager.updateChannelPosition(deviceChannel); - - // 发送redis消息。 通知位置信息的变化 - JSONObject jsonObject = new JSONObject(); - jsonObject.put("time", mobilePosition.getTime()); - jsonObject.put("serial", deviceChannel.getDeviceId()); - jsonObject.put("code", deviceChannel.getChannelId()); - jsonObject.put("longitude", mobilePosition.getLongitude()); - jsonObject.put("latitude", mobilePosition.getLatitude()); - jsonObject.put("altitude", mobilePosition.getAltitude()); - jsonObject.put("direction", mobilePosition.getDirection()); - jsonObject.put("speed", mobilePosition.getSpeed()); - redisCatchStorage.sendMobilePositionMsg(jsonObject); - } - } - if (!ObjectUtils.isEmpty(deviceAlarm.getDeviceId())) { - if (deviceAlarm.getAlarmMethod().contains(DeviceAlarmMethod.Video.getVal() + "")) { - deviceAlarm.setAlarmType(getText(rootElement.element("Info"), "AlarmType")); - } + taskQueueHandlerRun = false; + }); } - if ("7".equals(deviceAlarm.getAlarmMethod()) ) { - // 发送给平台的报警信息。 发送redis通知 - AlarmChannelMessage alarmChannelMessage = new AlarmChannelMessage(); - alarmChannelMessage.setAlarmSn(Integer.parseInt(deviceAlarm.getAlarmMethod())); - alarmChannelMessage.setAlarmDescription(deviceAlarm.getAlarmDescription()); - alarmChannelMessage.setGbId(channelId); - redisCatchStorage.sendAlarmMsg(alarmChannelMessage); - return; - } - logger.debug("存储报警信息、报警分类"); - // 存储报警信息、报警分类 - if (sipConfig.isAlarm()) { - deviceAlarmService.add(deviceAlarm); - } - logger.info("[收到报警通知]内容：{}", JSONObject.toJSON(deviceAlarm)); - if (redisCatchStorage.deviceIsOnline(device.getDeviceId())) { - publisher.deviceAlarmEventPublish(deviceAlarm); - } } @Override diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/MobilePositionNotifyMessageHandler.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/MobilePositionNotifyMessageHandler.java index b5051c0a..67d26d77 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/MobilePositionNotifyMessageHandler.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/MobilePositionNotifyMessageHandler.java @@ -17,6 +17,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.stereotype.Component; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; @@ -26,6 +28,7 @@ import javax.sip.RequestEvent; import javax.sip.SipException; import javax.sip.message.Response; import java.text.ParseException; +import java.util.concurrent.ConcurrentLinkedQueue; import static com.genersoft.iot.vmp.gb28181.utils.XmlUtil.getText; @@ -53,6 +56,14 @@ public class MobilePositionNotifyMessageHandler extends SIPRequestProcessorParen @Autowired private IDeviceChannelService deviceChannelService; + private boolean taskQueueHandlerRun = false; + + private final ConcurrentLinkedQueue taskQueue = new ConcurrentLinkedQueue<>(); + + @Qualifier("taskExecutor") + @Autowired + private ThreadPoolTaskExecutor taskExecutor; + @Override public void afterPropertiesSet() throws Exception { notifyMessageHandler.addHandler(cmdType, this); @@ -61,78 +72,91 @@ public class MobilePositionNotifyMessageHandler extends SIPRequestProcessorParen @Override public void handForDevice(RequestEvent evt, Device device, Element rootElement) { - try { - rootElement = getRootElement(evt, device.getCharset()); - if (rootElement == null) { - logger.warn("[ 移动设备位置数据通知 ] content cannot be null, {}", evt.getRequest()); - responseAck(getServerTransaction(evt), Response.BAD_REQUEST); - return; - } - MobilePosition mobilePosition = new MobilePosition(); - mobilePosition.setCreateTime(DateUtil.getNow()); - if (!ObjectUtils.isEmpty(device.getName())) { - mobilePosition.setDeviceName(device.getName()); - } - mobilePosition.setDeviceId(device.getDeviceId()); - mobilePosition.setChannelId(getText(rootElement, "DeviceID")); - mobilePosition.setTime(getText(rootElement, "Time")); - mobilePosition.setLongitude(Double.parseDouble(getText(rootElement, "Longitude"))); - mobilePosition.setLatitude(Double.parseDouble(getText(rootElement, "Latitude"))); - if (NumericUtil.isDouble(getText(rootElement, "Speed"))) { - mobilePosition.setSpeed(Double.parseDouble(getText(rootElement, "Speed"))); - } else { - mobilePosition.setSpeed(0.0); - } - if (NumericUtil.isDouble(getText(rootElement, "Direction"))) { - mobilePosition.setDirection(Double.parseDouble(getText(rootElement, "Direction"))); - } else { - mobilePosition.setDirection(0.0); - } - if (NumericUtil.isDouble(getText(rootElement, "Altitude"))) { - mobilePosition.setAltitude(Double.parseDouble(getText(rootElement, "Altitude"))); - } else { - mobilePosition.setAltitude(0.0); - } - mobilePosition.setReportSource("Mobile Position"); + taskQueue.offer(new SipMsgInfo(evt, device, rootElement)); + if (!taskQueueHandlerRun) { + taskQueueHandlerRun = true; + taskExecutor.execute(() -> { + while (!taskQueue.isEmpty()) { + SipMsgInfo sipMsgInfo = taskQueue.poll(); + try { + Element rootElementAfterCharset = getRootElement(sipMsgInfo.getEvt(), sipMsgInfo.getDevice().getCharset()); + if (rootElementAfterCharset == null) { + logger.warn("[ 移动设备位置数据通知 ] content cannot be null, {}", sipMsgInfo.getEvt().getRequest()); + responseAck(getServerTransaction(sipMsgInfo.getEvt()), Response.BAD_REQUEST); + return; + } + MobilePosition mobilePosition = new MobilePosition(); + mobilePosition.setCreateTime(DateUtil.getNow()); + if (!ObjectUtils.isEmpty(sipMsgInfo.getDevice().getName())) { + mobilePosition.setDeviceName(sipMsgInfo.getDevice().getName()); + } + mobilePosition.setDeviceId(sipMsgInfo.getDevice().getDeviceId()); + mobilePosition.setChannelId(getText(rootElementAfterCharset, "DeviceID")); + mobilePosition.setTime(getText(rootElementAfterCharset, "Time")); + mobilePosition.setLongitude(Double.parseDouble(getText(rootElementAfterCharset, "Longitude"))); + mobilePosition.setLatitude(Double.parseDouble(getText(rootElementAfterCharset, "Latitude"))); + if (NumericUtil.isDouble(getText(rootElementAfterCharset, "Speed"))) { + mobilePosition.setSpeed(Double.parseDouble(getText(rootElementAfterCharset, "Speed"))); + } else { + mobilePosition.setSpeed(0.0); + } + if (NumericUtil.isDouble(getText(rootElementAfterCharset, "Direction"))) { + mobilePosition.setDirection(Double.parseDouble(getText(rootElementAfterCharset, "Direction"))); + } else { + mobilePosition.setDirection(0.0); + } + if (NumericUtil.isDouble(getText(rootElementAfterCharset, "Altitude"))) { + mobilePosition.setAltitude(Double.parseDouble(getText(rootElementAfterCharset, "Altitude"))); + } else { + mobilePosition.setAltitude(0.0); + } + mobilePosition.setReportSource("Mobile Position"); - // 更新device channel 的经纬度 - DeviceChannel deviceChannel = new DeviceChannel(); - deviceChannel.setDeviceId(device.getDeviceId()); - deviceChannel.setChannelId(mobilePosition.getChannelId()); - deviceChannel.setLongitude(mobilePosition.getLongitude()); - deviceChannel.setLatitude(mobilePosition.getLatitude()); - deviceChannel.setGpsTime(mobilePosition.getTime()); + // 更新device channel 的经纬度 + DeviceChannel deviceChannel = new DeviceChannel(); + deviceChannel.setDeviceId(sipMsgInfo.getDevice().getDeviceId()); + deviceChannel.setChannelId(mobilePosition.getChannelId()); + deviceChannel.setLongitude(mobilePosition.getLongitude()); + deviceChannel.setLatitude(mobilePosition.getLatitude()); + deviceChannel.setGpsTime(mobilePosition.getTime()); - deviceChannel = deviceChannelService.updateGps(deviceChannel, device); + deviceChannel = deviceChannelService.updateGps(deviceChannel, sipMsgInfo.getDevice()); - mobilePosition.setLongitudeWgs84(deviceChannel.getLongitudeWgs84()); - mobilePosition.setLatitudeWgs84(deviceChannel.getLatitudeWgs84()); - mobilePosition.setLongitudeGcj02(deviceChannel.getLongitudeGcj02()); - mobilePosition.setLatitudeGcj02(deviceChannel.getLatitudeGcj02()); + mobilePosition.setLongitudeWgs84(deviceChannel.getLongitudeWgs84()); + mobilePosition.setLatitudeWgs84(deviceChannel.getLatitudeWgs84()); + mobilePosition.setLongitudeGcj02(deviceChannel.getLongitudeGcj02()); + mobilePosition.setLatitudeGcj02(deviceChannel.getLatitudeGcj02()); - if (userSetting.getSavePositionHistory()) { - storager.insertMobilePosition(mobilePosition); - } - storager.updateChannelPosition(deviceChannel); - //回复 200 OK - responseAck(getServerTransaction(evt), Response.OK); + if (userSetting.getSavePositionHistory()) { + storager.insertMobilePosition(mobilePosition); + } + storager.updateChannelPosition(deviceChannel); + //回复 200 OK + responseAck(getServerTransaction(sipMsgInfo.getEvt()), Response.OK); - // 发送redis消息。 通知位置信息的变化 - JSONObject jsonObject = new JSONObject(); - jsonObject.put("time", mobilePosition.getTime()); - jsonObject.put("serial", deviceChannel.getDeviceId()); - jsonObject.put("code", deviceChannel.getChannelId()); - jsonObject.put("longitude", mobilePosition.getLongitude()); - jsonObject.put("latitude", mobilePosition.getLatitude()); - jsonObject.put("altitude", mobilePosition.getAltitude()); - jsonObject.put("direction", mobilePosition.getDirection()); - jsonObject.put("speed", mobilePosition.getSpeed()); - redisCatchStorage.sendMobilePositionMsg(jsonObject); + // 发送redis消息。 通知位置信息的变化 + JSONObject jsonObject = new JSONObject(); + jsonObject.put("time", mobilePosition.getTime()); + jsonObject.put("serial", deviceChannel.getDeviceId()); + jsonObject.put("code", deviceChannel.getChannelId()); + jsonObject.put("longitude", mobilePosition.getLongitude()); + jsonObject.put("latitude", mobilePosition.getLatitude()); + jsonObject.put("altitude", mobilePosition.getAltitude()); + jsonObject.put("direction", mobilePosition.getDirection()); + jsonObject.put("speed", mobilePosition.getSpeed()); + redisCatchStorage.sendMobilePositionMsg(jsonObject); - } catch (DocumentException | SipException | InvalidArgumentException | ParseException e) { - e.printStackTrace(); + } catch (DocumentException | SipException | InvalidArgumentException | ParseException e) { + e.printStackTrace(); + } + + } + taskQueueHandlerRun = false; + }); } + + } @Override diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/PlatformServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/PlatformServiceImpl.java index dedc7c19..3c5644b3 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/PlatformServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/PlatformServiceImpl.java @@ -1,6 +1,7 @@ package com.genersoft.iot.vmp.service.impl; import com.genersoft.iot.vmp.conf.DynamicTask; +import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.gb28181.event.SipSubscribe; import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform; @@ -59,6 +60,9 @@ public class PlatformServiceImpl implements IPlatformService { @Autowired private GbStreamMapper gbStreamMapper; + @Autowired + private UserSetting userSetting; + @Override @@ -241,7 +245,7 @@ public class PlatformServiceImpl implements IPlatformService { if (subscribe != null) { // TODO 暂时只处理视频流的回复,后续增加对国标设备的支持 - List gbStreams = gbStreamMapper.queryGbStreamListInPlatform(platform.getServerGBId()); + List gbStreams = gbStreamMapper.queryGbStreamListInPlatform(platform.getServerGBId(), userSetting.isUsePushingAsStatus()); if (gbStreams.size() == 0) { return; } diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java index c0b9e95b..6662738c 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java @@ -131,23 +131,6 @@ public class PlayServiceImpl implements IPlayService { StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId); playResult.setDevice(device); - result.onCompletion(()->{ - // 点播结束时调用截图接口 - taskExecutor.execute(()->{ - // TODO 应该在上流时调用更好，结束也可能是错误结束 - String path = "snap"; - String fileName = deviceId + "_" + channelId + ".jpg"; - WVPResult wvpResult = (WVPResult)result.getResult(); - if (Objects.requireNonNull(wvpResult).getCode() == 0) { - StreamInfo streamInfoForSuccess = (StreamInfo)wvpResult.getData(); - MediaServerItem mediaInfo = mediaServerService.getOne(streamInfoForSuccess.getMediaServerId()); - String streamUrl = streamInfoForSuccess.getFmp4(); - // 请求截图 - logger.info("[请求截图]: " + fileName); - zlmresTfulUtils.getSnap(mediaInfo, streamUrl, 15, 1, path, fileName); - } - }); - }); if (streamInfo != null) { String streamId = streamInfo.getStream(); if (streamId == null) { @@ -209,6 +192,21 @@ public class PlayServiceImpl implements IPlayService { SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, device.isSsrcCheck(), false); logger.info(JSONObject.toJSONString(ssrcInfo)); play(mediaServerItem, ssrcInfo, device, channelId, (mediaServerItemInUse, response)->{ + // 点播结束时调用截图接口 + taskExecutor.execute(()->{ + // TODO 应该在上流时调用更好，结束也可能是错误结束 + String path = "snap"; + String fileName = deviceId + "_" + channelId + ".jpg"; + WVPResult wvpResult = (WVPResult)result.getResult(); + if (Objects.requireNonNull(wvpResult).getCode() == 0) { + StreamInfo streamInfoForSuccess = (StreamInfo)wvpResult.getData(); + MediaServerItem mediaInfo = mediaServerService.getOne(streamInfoForSuccess.getMediaServerId()); + String streamUrl = streamInfoForSuccess.getFmp4(); + // 请求截图 + logger.info("[请求截图]: " + fileName); + zlmresTfulUtils.getSnap(mediaInfo, streamUrl, 15, 1, path, fileName); + } + }); if (hookEvent != null) { hookEvent.response(mediaServerItem, response); } diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/RedisAlarmMsgListener.java b/src/main/java/com/genersoft/iot/vmp/service/impl/RedisAlarmMsgListener.java deleted file mode 100644 index 1634234a..00000000 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/RedisAlarmMsgListener.java +++ /dev/null @@ -1,77 +0,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.IPlatformChannelService; -import com.genersoft.iot.vmp.storager.IRedisCatchStorage; -import com.genersoft.iot.vmp.storager.IVideoManagerStorage; -import com.genersoft.iot.vmp.utils.DateUtil; -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.stereotype.Component; -import org.springframework.util.ObjectUtils; - -import java.util.List; - - -@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; - - @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(); - - DeviceAlarm deviceAlarm = new DeviceAlarm(); - deviceAlarm.setCreateTime(DateUtil.getNow()); - deviceAlarm.setChannelId(gbId); - deviceAlarm.setAlarmDescription(alarmChannelMessage.getAlarmDescription()); - deviceAlarm.setAlarmMethod("" + alarmChannelMessage.getAlarmSn()); - deviceAlarm.setAlarmPriority("1"); - deviceAlarm.setAlarmTime(DateUtil.getNowForISO8601()); - deviceAlarm.setAlarmType("1"); - deviceAlarm.setLongitude(0); - deviceAlarm.setLatitude(0); - - if (ObjectUtils.isEmpty(gbId)) { - // 发送给所有的上级 - List parentPlatforms = storage.queryEnableParentPlatformList(true); - if (parentPlatforms.size() > 0) { - for (ParentPlatform parentPlatform : parentPlatforms) { - commanderForPlatform.sendAlarmMessage(parentPlatform, deviceAlarm); - } - } - }else { - Device device = storage.queryVideoDevice(gbId); - ParentPlatform platform = storage.queryParentPlatByServerGBId(gbId); - if (device != null && platform == null) { - commander.sendAlarmMessage(device, deviceAlarm); - }else if (device == null && platform != null){ - commanderForPlatform.sendAlarmMessage(platform, deviceAlarm); - }else { - logger.warn("无法确定" + gbId + "是平台还是设备"); - } - } - } -} diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/RedisPushStreamResponseListener.java b/src/main/java/com/genersoft/iot/vmp/service/impl/RedisPushStreamResponseListener.java deleted file mode 100644 index 56c9ff31..00000000 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/RedisPushStreamResponseListener.java +++ /dev/null @@ -1,62 +0,0 @@ -package com.genersoft.iot.vmp.service.impl; - -import com.alibaba.fastjson.JSON; -import com.alibaba.fastjson.JSONObject; -import com.genersoft.iot.vmp.media.zlm.dto.ChannelOnlineEvent; -import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem; -import com.genersoft.iot.vmp.service.IGbStreamService; -import com.genersoft.iot.vmp.service.IMediaServerService; -import com.genersoft.iot.vmp.service.IStreamPushService; -import com.genersoft.iot.vmp.service.bean.MessageForPushChannelResponse; -import com.genersoft.iot.vmp.utils.DateUtil; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.data.redis.connection.Message; -import org.springframework.data.redis.connection.MessageListener; -import org.springframework.stereotype.Component; -import org.springframework.util.ObjectUtils; - -import javax.annotation.Resource; -import java.util.ArrayList; -import java.util.List; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; - -/** - * 接收redis返回的推流结果 - * @author lin - */ -@Component -public class RedisPushStreamResponseListener implements MessageListener { - - private final static Logger logger = LoggerFactory.getLogger(RedisPushStreamResponseListener.class); - - private Map responseEvents = new ConcurrentHashMap<>(); - - public interface PushStreamResponseEvent{ - void run(MessageForPushChannelResponse response); - } - - @Override - public void onMessage(Message message, byte[] bytes) { - // - logger.warn("[REDIS消息-请求推流结果]： {}", new String(message.getBody())); - MessageForPushChannelResponse response = JSON.parseObject(new String(message.getBody()), MessageForPushChannelResponse.class); - if (response == null || ObjectUtils.isEmpty(response.getApp()) || ObjectUtils.isEmpty(response.getStream())){ - logger.info("[REDIS消息-请求推流结果]：参数不全"); - return; - } - // 查看正在等待的invite消息 - if (responseEvents.get(response.getApp() + response.getStream()) != null) { - responseEvents.get(response.getApp() + response.getStream()).run(response); - } - } - - public void addEvent(String app, String stream, PushStreamResponseEvent callback) { - responseEvents.put(app + stream, callback); - } - - public void removeEvent(String app, String stream) { - responseEvents.remove(app + stream); - } -} diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/RedisPushStreamStatusListMsgListener.java b/src/main/java/com/genersoft/iot/vmp/service/impl/RedisPushStreamStatusListMsgListener.java deleted file mode 100644 index bedbf447..00000000 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/RedisPushStreamStatusListMsgListener.java +++ /dev/null @@ -1,81 +0,0 @@ -package com.genersoft.iot.vmp.service.impl; - -import com.alibaba.fastjson.JSON; -import com.alibaba.fastjson.JSONObject; -import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem; -import com.genersoft.iot.vmp.service.IGbStreamService; -import com.genersoft.iot.vmp.service.IMediaServerService; -import com.genersoft.iot.vmp.service.IStreamPushService; -import com.genersoft.iot.vmp.utils.DateUtil; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.data.redis.connection.Message; -import org.springframework.data.redis.connection.MessageListener; -import org.springframework.stereotype.Component; - -import javax.annotation.Resource; -import java.util.*; - -/** - * @Auther: JiangFeng - * @Date: 2022/8/16 11:32 - * @Description: 接收redis发送的推流设备列表更新通知 - */ -@Component -public class RedisPushStreamStatusListMsgListener implements MessageListener { - - private final static Logger logger = LoggerFactory.getLogger(RedisPushStreamStatusListMsgListener.class); - @Resource - private IMediaServerService mediaServerService; - - @Resource - private IStreamPushService streamPushService; - @Resource - private IGbStreamService gbStreamService; - - @Override - public void onMessage(Message message, byte[] bytes) { - // - logger.warn("[REDIS消息-推流设备列表更新]： {}", new String(message.getBody())); - List streamPushItems = JSON.parseArray(new String(message.getBody()), StreamPushItem.class); - //查询全部的app+stream 用于判断是添加还是修改 - List allAppAndStream = streamPushService.getAllAppAndStream(); - - /** - * 用于存储更具APP+Stream过滤后的数据，可以直接存入stream_push表与gb_stream表 - */ - List streamPushItemForSave = new ArrayList<>(); - List streamPushItemForUpdate = new ArrayList<>(); - for (StreamPushItem streamPushItem : streamPushItems) { - String app = streamPushItem.getApp(); - String stream = streamPushItem.getStream(); - boolean contains = allAppAndStream.contains(app + stream); - //不存在就添加 - if (!contains) { - streamPushItem.setStreamType("push"); - streamPushItem.setCreateTime(DateUtil.getNow()); - streamPushItem.setMediaServerId(mediaServerService.getDefaultMediaServer().getId()); - streamPushItem.setOriginType(2); - streamPushItem.setOriginTypeStr("rtsp_push"); - streamPushItem.setTotalReaderCount("0"); - streamPushItemForSave.add(streamPushItem); - } else { - //存在就只修改 name和gbId - streamPushItemForUpdate.add(streamPushItem); - } - } - if (streamPushItemForSave.size() > 0) { - - logger.info("添加{}条",streamPushItemForSave.size()); - logger.info(JSONObject.toJSONString(streamPushItemForSave)); - streamPushService.batchAdd(streamPushItemForSave); - - } - if(streamPushItemForUpdate.size()>0){ - logger.info("修改{}条",streamPushItemForUpdate.size()); - logger.info(JSONObject.toJSONString(streamPushItemForUpdate)); - gbStreamService.updateGbIdOrName(streamPushItemForUpdate); - } - - } -} diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/RedisStreamMsgListener.java b/src/main/java/com/genersoft/iot/vmp/service/impl/RedisStreamMsgListener.java deleted file mode 100644 index 118a2274..00000000 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/RedisStreamMsgListener.java +++ /dev/null @@ -1,72 +0,0 @@ -package com.genersoft.iot.vmp.service.impl; - -import com.alibaba.fastjson.JSON; -import com.alibaba.fastjson.JSONObject; -import com.genersoft.iot.vmp.conf.UserSetting; - -import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander; -import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform; -import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager; -import com.genersoft.iot.vmp.media.zlm.dto.MediaItem; -import com.genersoft.iot.vmp.storager.IVideoManagerStorage; -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.stereotype.Component; - - -/** - * 接收其他wvp发送流变化通知 - * @author lin - */ -@Component -public class RedisStreamMsgListener implements MessageListener { - - private final static Logger logger = LoggerFactory.getLogger(RedisStreamMsgListener.class); - - @Autowired - private UserSetting userSetting; - - @Autowired - private ZLMMediaListManager zlmMediaListManager; - - @Override - public void onMessage(Message message, byte[] bytes) { - - JSONObject steamMsgJson = JSON.parseObject(message.getBody(), JSONObject.class); - if (steamMsgJson == null) { - logger.warn("[收到redis 流变化]消息解析失败"); - return; - } - String serverId = steamMsgJson.getString("serverId"); - - if (userSetting.getServerId().equals(serverId)) { - // 自己发送的消息忽略即可 - return; - } - logger.info("[收到redis 流变化]： {}", new String(message.getBody())); - String app = steamMsgJson.getString("app"); - String stream = steamMsgJson.getString("stream"); - boolean register = steamMsgJson.getBoolean("register"); - String mediaServerId = steamMsgJson.getString("mediaServerId"); - MediaItem mediaItem = new MediaItem(); - mediaItem.setSeverId(serverId); - mediaItem.setApp(app); - mediaItem.setStream(stream); - mediaItem.setRegist(register); - mediaItem.setMediaServerId(mediaServerId); - mediaItem.setCreateStamp(System.currentTimeMillis()/1000); - mediaItem.setAliveSecond(0L); - mediaItem.setTotalReaderCount("0"); - mediaItem.setOriginType(0); - mediaItem.setOriginTypeStr("0"); - mediaItem.setOriginTypeStr("unknown"); - if (register) { - zlmMediaListManager.addPush(mediaItem); - }else { - zlmMediaListManager.removeMedia(app, stream); - } - } -} diff --git a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisAlarmMsgListener.java b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisAlarmMsgListener.java new file mode 100644 index 00000000..84f5ef71 --- /dev/null +++ b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisAlarmMsgListener.java @@ -0,0 +1,100 @@ +package com.genersoft.iot.vmp.service.redisMsg; + +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.storager.IVideoManagerStorage; +import com.genersoft.iot.vmp.utils.DateUtil; +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.data.redis.connection.Message; +import org.springframework.data.redis.connection.MessageListener; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import org.springframework.stereotype.Component; +import org.springframework.util.ObjectUtils; + +import java.util.List; +import java.util.concurrent.ConcurrentLinkedQueue; + + +@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 boolean taskQueueHandlerRun = false; + + private final ConcurrentLinkedQueue taskQueue = new ConcurrentLinkedQueue<>(); + + @Qualifier("taskExecutor") + @Autowired + private ThreadPoolTaskExecutor taskExecutor; + + @Override + public void onMessage(Message message, byte[] bytes) { + logger.info("收到来自REDIS的ALARM通知： {}", new String(message.getBody())); + + taskQueue.offer(message); + if (!taskQueueHandlerRun) { + taskQueueHandlerRun = true; + taskExecutor.execute(() -> { + while (!taskQueue.isEmpty()) { + Message msg = taskQueue.poll(); + + AlarmChannelMessage alarmChannelMessage = JSON.parseObject(msg.getBody(), AlarmChannelMessage.class); + if (alarmChannelMessage == null) { + logger.warn("[REDIS的ALARM通知]消息解析失败"); + return; + } + String gbId = alarmChannelMessage.getGbId(); + + DeviceAlarm deviceAlarm = new DeviceAlarm(); + deviceAlarm.setCreateTime(DateUtil.getNow()); + deviceAlarm.setChannelId(gbId); + deviceAlarm.setAlarmDescription(alarmChannelMessage.getAlarmDescription()); + deviceAlarm.setAlarmMethod("" + alarmChannelMessage.getAlarmSn()); + deviceAlarm.setAlarmPriority("1"); + deviceAlarm.setAlarmTime(DateUtil.getNowForISO8601()); + deviceAlarm.setAlarmType("1"); + deviceAlarm.setLongitude(0); + deviceAlarm.setLatitude(0); + + if (ObjectUtils.isEmpty(gbId)) { + // 发送给所有的上级 + List parentPlatforms = storage.queryEnableParentPlatformList(true); + if (parentPlatforms.size() > 0) { + for (ParentPlatform parentPlatform : parentPlatforms) { + commanderForPlatform.sendAlarmMessage(parentPlatform, deviceAlarm); + } + } + }else { + Device device = storage.queryVideoDevice(gbId); + ParentPlatform platform = storage.queryParentPlatByServerGBId(gbId); + if (device != null && platform == null) { + commander.sendAlarmMessage(device, deviceAlarm); + }else if (device == null && platform != null){ + commanderForPlatform.sendAlarmMessage(platform, deviceAlarm); + }else { + logger.warn("无法确定" + gbId + "是平台还是设备"); + } + } + } + taskQueueHandlerRun = false; + }); + } + + + } +} diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/RedisGbPlayMsgListener.java b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGbPlayMsgListener.java similarity index 67% rename from src/main/java/com/genersoft/iot/vmp/service/impl/RedisGbPlayMsgListener.java rename to src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGbPlayMsgListener.java index e327d0f2..e6ba9b6b 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/RedisGbPlayMsgListener.java +++ b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGbPlayMsgListener.java @@ -1,4 +1,4 @@ -package com.genersoft.iot.vmp.service.impl; +package com.genersoft.iot.vmp.service.redisMsg; import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONObject; @@ -19,8 +19,10 @@ 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.beans.factory.annotation.Qualifier; import org.springframework.data.redis.connection.Message; import org.springframework.data.redis.connection.MessageListener; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.stereotype.Component; import java.text.ParseException; @@ -28,6 +30,7 @@ import java.util.HashMap; import java.util.Map; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; /** @@ -85,6 +88,14 @@ public class RedisGbPlayMsgListener implements MessageListener { @Autowired private ZlmHttpHookSubscribe subscribe; + private boolean taskQueueHandlerRun = false; + + private final ConcurrentLinkedQueue taskQueue = new ConcurrentLinkedQueue<>(); + + @Qualifier("taskExecutor") + @Autowired + private ThreadPoolTaskExecutor taskExecutor; + public interface PlayMsgCallback{ void handler(ResponseSendItemMsg responseSendItemMsg) throws ParseException; @@ -100,94 +111,107 @@ public class RedisGbPlayMsgListener implements MessageListener { @Override public void onMessage(Message message, byte[] bytes) { - JSONObject msgJSON = JSON.parseObject(message.getBody(), JSONObject.class); - WvpRedisMsg wvpRedisMsg = JSON.toJavaObject(msgJSON, WvpRedisMsg.class); - if (!userSetting.getServerId().equals(wvpRedisMsg.getToId())) { - return; - } - if (WvpRedisMsg.isRequest(wvpRedisMsg)) { - logger.info("[收到REDIS通知] 请求： {}", new String(message.getBody())); - switch (wvpRedisMsg.getCmd()){ - case WvpRedisMsgCmd.GET_SEND_ITEM: - RequestSendItemMsg content = JSON.toJavaObject((JSONObject)wvpRedisMsg.getContent(), RequestSendItemMsg.class); - requestSendItemMsgHand(content, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial()); - break; - case WvpRedisMsgCmd.REQUEST_PUSH_STREAM: - RequestPushStreamMsg param = JSON.toJavaObject((JSONObject)wvpRedisMsg.getContent(), RequestPushStreamMsg.class);; - requestPushStreamMsgHand(param, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial()); - break; - default: - break; - } + taskQueue.offer(message); + if (!taskQueueHandlerRun) { + taskQueueHandlerRun = true; + taskExecutor.execute(() -> { + while (!taskQueue.isEmpty()) { + Message msg = taskQueue.poll(); + JSONObject msgJSON = JSON.parseObject(msg.getBody(), JSONObject.class); + WvpRedisMsg wvpRedisMsg = JSON.toJavaObject(msgJSON, WvpRedisMsg.class); + if (!userSetting.getServerId().equals(wvpRedisMsg.getToId())) { + return; + } + if (WvpRedisMsg.isRequest(wvpRedisMsg)) { + logger.info("[收到REDIS通知] 请求： {}", new String(msg.getBody())); - }else { - logger.info("[收到REDIS通知] 回复： {}", new String(message.getBody())); - switch (wvpRedisMsg.getCmd()){ - case WvpRedisMsgCmd.GET_SEND_ITEM: + switch (wvpRedisMsg.getCmd()){ + case WvpRedisMsgCmd.GET_SEND_ITEM: + RequestSendItemMsg content = JSON.toJavaObject((JSONObject)wvpRedisMsg.getContent(), RequestSendItemMsg.class); + requestSendItemMsgHand(content, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial()); + break; + case WvpRedisMsgCmd.REQUEST_PUSH_STREAM: + RequestPushStreamMsg param = JSON.toJavaObject((JSONObject)wvpRedisMsg.getContent(), RequestPushStreamMsg.class);; + requestPushStreamMsgHand(param, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial()); + break; + default: + break; + } - WVPResult content = JSON.toJavaObject((JSONObject)wvpRedisMsg.getContent(), WVPResult.class); + }else { + logger.info("[收到REDIS通知] 回复： {}", new String(msg.getBody())); + switch (wvpRedisMsg.getCmd()){ + case WvpRedisMsgCmd.GET_SEND_ITEM: - String key = wvpRedisMsg.getSerial(); - switch (content.getCode()) { - case 0: - ResponseSendItemMsg responseSendItemMsg =JSON.toJavaObject((JSONObject)content.getData(), ResponseSendItemMsg.class); - PlayMsgCallback playMsgCallback = callbacks.get(key); - if (playMsgCallback != null) { - callbacksForError.remove(key); - try { - playMsgCallback.handler(responseSendItemMsg); - } catch (ParseException e) { - throw new RuntimeException(e); + WVPResult content = JSON.toJavaObject((JSONObject)wvpRedisMsg.getContent(), WVPResult.class); + + String key = wvpRedisMsg.getSerial(); + switch (content.getCode()) { + case 0: + ResponseSendItemMsg responseSendItemMsg =JSON.toJavaObject((JSONObject)content.getData(), ResponseSendItemMsg.class); + PlayMsgCallback playMsgCallback = callbacks.get(key); + if (playMsgCallback != null) { + callbacksForError.remove(key); + try { + playMsgCallback.handler(responseSendItemMsg); + } catch (ParseException e) { + throw new RuntimeException(e); + } + } + break; + case ERROR_CODE_MEDIA_SERVER_NOT_FOUND: + case ERROR_CODE_OFFLINE: + case ERROR_CODE_TIMEOUT: + PlayMsgErrorCallback errorCallback = callbacksForError.get(key); + if (errorCallback != null) { + callbacks.remove(key); + errorCallback.handler(content); + } + break; + default: + break; } - } - break; - case ERROR_CODE_MEDIA_SERVER_NOT_FOUND: - case ERROR_CODE_OFFLINE: - case ERROR_CODE_TIMEOUT: - PlayMsgErrorCallback errorCallback = callbacksForError.get(key); - if (errorCallback != null) { - callbacks.remove(key); - errorCallback.handler(content); - } - break; - default: - break; + break; + case WvpRedisMsgCmd.REQUEST_PUSH_STREAM: + WVPResult wvpResult = JSON.toJavaObject((JSONObject)wvpRedisMsg.getContent(), WVPResult.class); + String serial = wvpRedisMsg.getSerial(); + switch (wvpResult.getCode()) { + case 0: + JSONObject jsonObject = (JSONObject)wvpResult.getData(); + PlayMsgCallbackForStartSendRtpStream playMsgCallback = callbacksForStartSendRtpStream.get(serial); + if (playMsgCallback != null) { + callbacksForError.remove(serial); + playMsgCallback.handler(jsonObject); + } + break; + case ERROR_CODE_MEDIA_SERVER_NOT_FOUND: + case ERROR_CODE_OFFLINE: + case ERROR_CODE_TIMEOUT: + PlayMsgErrorCallback errorCallback = callbacksForError.get(serial); + if (errorCallback != null) { + callbacks.remove(serial); + errorCallback.handler(wvpResult); + } + break; + default: + break; + } + break; + default: + break; + } } - break; - case WvpRedisMsgCmd.REQUEST_PUSH_STREAM: - WVPResult wvpResult = JSON.toJavaObject((JSONObject)wvpRedisMsg.getContent(), WVPResult.class); - String serial = wvpRedisMsg.getSerial(); - switch (wvpResult.getCode()) { - case 0: - JSONObject jsonObject = (JSONObject)wvpResult.getData(); - PlayMsgCallbackForStartSendRtpStream playMsgCallback = callbacksForStartSendRtpStream.get(serial); - if (playMsgCallback != null) { - callbacksForError.remove(serial); - playMsgCallback.handler(jsonObject); - } - break; - case ERROR_CODE_MEDIA_SERVER_NOT_FOUND: - case ERROR_CODE_OFFLINE: - case ERROR_CODE_TIMEOUT: - PlayMsgErrorCallback errorCallback = callbacksForError.get(serial); - if (errorCallback != null) { - callbacks.remove(serial); - errorCallback.handler(wvpResult); - } - break; - default: - break; - } - break; - default: - break; - } + } + taskQueueHandlerRun = false; + }); } + + } /** diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/RedisGpsMsgListener.java b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGpsMsgListener.java similarity index 96% rename from src/main/java/com/genersoft/iot/vmp/service/impl/RedisGpsMsgListener.java rename to src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGpsMsgListener.java index 4e94d68a..bb2f4ad5 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/RedisGpsMsgListener.java +++ b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGpsMsgListener.java @@ -1,7 +1,6 @@ -package com.genersoft.iot.vmp.service.impl; +package com.genersoft.iot.vmp.service.redisMsg; import com.alibaba.fastjson.JSON; -import com.genersoft.iot.vmp.gb28181.bean.HandlerCatchData; import com.genersoft.iot.vmp.service.bean.GPSMsgInfo; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage; diff --git a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamResponseListener.java b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamResponseListener.java new file mode 100644 index 00000000..13e68740 --- /dev/null +++ b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamResponseListener.java @@ -0,0 +1,75 @@ +package com.genersoft.iot.vmp.service.redisMsg; + +import com.alibaba.fastjson.JSON; +import com.genersoft.iot.vmp.service.bean.GPSMsgInfo; +import com.genersoft.iot.vmp.service.bean.MessageForPushChannelResponse; +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.data.redis.connection.Message; +import org.springframework.data.redis.connection.MessageListener; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import org.springframework.stereotype.Component; +import org.springframework.util.ObjectUtils; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; + +/** + * 接收redis返回的推流结果 + * @author lin + */ +@Component +public class RedisPushStreamResponseListener implements MessageListener { + + private final static Logger logger = LoggerFactory.getLogger(RedisPushStreamResponseListener.class); + + private boolean taskQueueHandlerRun = false; + + private final ConcurrentLinkedQueue taskQueue = new ConcurrentLinkedQueue<>(); + + @Qualifier("taskExecutor") + @Autowired + private ThreadPoolTaskExecutor taskExecutor; + + + private Map responseEvents = new ConcurrentHashMap<>(); + + public interface PushStreamResponseEvent{ + void run(MessageForPushChannelResponse response); + } + + @Override + public void onMessage(Message message, byte[] bytes) { + logger.warn("[REDIS消息-请求推流结果]： {}", new String(message.getBody())); + taskQueue.offer(message); + if (!taskQueueHandlerRun) { + taskQueueHandlerRun = true; + taskExecutor.execute(() -> { + while (!taskQueue.isEmpty()) { + Message msg = taskQueue.poll(); + MessageForPushChannelResponse response = JSON.parseObject(new String(msg.getBody()), MessageForPushChannelResponse.class); + if (response == null || ObjectUtils.isEmpty(response.getApp()) || ObjectUtils.isEmpty(response.getStream())){ + logger.info("[REDIS消息-请求推流结果]：参数不全"); + return; + } + // 查看正在等待的invite消息 + if (responseEvents.get(response.getApp() + response.getStream()) != null) { + responseEvents.get(response.getApp() + response.getStream()).run(response); + } + } + taskQueueHandlerRun = false; + }); + } + } + + public void addEvent(String app, String stream, PushStreamResponseEvent callback) { + responseEvents.put(app + stream, callback); + } + + public void removeEvent(String app, String stream) { + responseEvents.remove(app + stream); + } +} diff --git a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamStatusListMsgListener.java b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamStatusListMsgListener.java new file mode 100644 index 00000000..39258364 --- /dev/null +++ b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamStatusListMsgListener.java @@ -0,0 +1,103 @@ +package com.genersoft.iot.vmp.service.redisMsg; + +import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson.JSONObject; +import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem; +import com.genersoft.iot.vmp.service.IGbStreamService; +import com.genersoft.iot.vmp.service.IMediaServerService; +import com.genersoft.iot.vmp.service.IStreamPushService; +import com.genersoft.iot.vmp.service.bean.GPSMsgInfo; +import com.genersoft.iot.vmp.utils.DateUtil; +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.data.redis.connection.Message; +import org.springframework.data.redis.connection.MessageListener; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import org.springframework.stereotype.Component; + +import javax.annotation.Resource; +import java.util.*; +import java.util.concurrent.ConcurrentLinkedQueue; + +/** + * @Auther: JiangFeng + * @Date: 2022/8/16 11:32 + * @Description: 接收redis发送的推流设备列表更新通知 + */ +@Component +public class RedisPushStreamStatusListMsgListener implements MessageListener { + + private final static Logger logger = LoggerFactory.getLogger(RedisPushStreamStatusListMsgListener.class); + @Resource + private IMediaServerService mediaServerService; + + @Resource + private IStreamPushService streamPushService; + @Resource + private IGbStreamService gbStreamService; + + private boolean taskQueueHandlerRun = false; + + private final ConcurrentLinkedQueue taskQueue = new ConcurrentLinkedQueue<>(); + + @Qualifier("taskExecutor") + @Autowired + private ThreadPoolTaskExecutor taskExecutor; + + @Override + public void onMessage(Message message, byte[] bytes) { + logger.info("[REDIS消息-推流设备列表更新]： {}", new String(message.getBody())); + + taskQueue.offer(message); + if (!taskQueueHandlerRun) { + taskQueueHandlerRun = true; + taskExecutor.execute(() -> { + while (!taskQueue.isEmpty()) { + Message msg = taskQueue.poll(); + List streamPushItems = JSON.parseArray(new String(msg.getBody()), StreamPushItem.class); + //查询全部的app+stream 用于判断是添加还是修改 + List allAppAndStream = streamPushService.getAllAppAndStream(); + + /** + * 用于存储更具APP+Stream过滤后的数据，可以直接存入stream_push表与gb_stream表 + */ + List streamPushItemForSave = new ArrayList<>(); + List streamPushItemForUpdate = new ArrayList<>(); + for (StreamPushItem streamPushItem : streamPushItems) { + String app = streamPushItem.getApp(); + String stream = streamPushItem.getStream(); + boolean contains = allAppAndStream.contains(app + stream); + //不存在就添加 + if (!contains) { + streamPushItem.setStreamType("push"); + streamPushItem.setCreateTime(DateUtil.getNow()); + streamPushItem.setMediaServerId(mediaServerService.getDefaultMediaServer().getId()); + streamPushItem.setOriginType(2); + streamPushItem.setOriginTypeStr("rtsp_push"); + streamPushItem.setTotalReaderCount("0"); + streamPushItemForSave.add(streamPushItem); + } else { + //存在就只修改 name和gbId + streamPushItemForUpdate.add(streamPushItem); + } + } + if (streamPushItemForSave.size() > 0) { + + logger.info("添加{}条",streamPushItemForSave.size()); + logger.info(JSONObject.toJSONString(streamPushItemForSave)); + streamPushService.batchAdd(streamPushItemForSave); + + } + if(streamPushItemForUpdate.size()>0){ + logger.info("修改{}条",streamPushItemForUpdate.size()); + logger.info(JSONObject.toJSONString(streamPushItemForUpdate)); + gbStreamService.updateGbIdOrName(streamPushItemForUpdate); + } + } + taskQueueHandlerRun = false; + }); + } + } +} diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/RedisPushStreamStatusMsgListener.java b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamStatusMsgListener.java similarity index 82% rename from src/main/java/com/genersoft/iot/vmp/service/impl/RedisPushStreamStatusMsgListener.java rename to src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamStatusMsgListener.java index 50e894a6..4b1c2d50 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/RedisPushStreamStatusMsgListener.java +++ b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamStatusMsgListener.java @@ -1,24 +1,11 @@ -package com.genersoft.iot.vmp.service.impl; +package com.genersoft.iot.vmp.service.redisMsg; import com.alibaba.fastjson.JSON; -import com.alibaba.fastjson.JSONObject; import com.genersoft.iot.vmp.common.VideoManagerConstants; import com.genersoft.iot.vmp.conf.DynamicTask; -import com.genersoft.iot.vmp.conf.UserSetting; -import com.genersoft.iot.vmp.gb28181.bean.GbStream; -import com.genersoft.iot.vmp.gb28181.event.EventPublisher; -import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent; -import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander; -import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform; -import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager; -import com.genersoft.iot.vmp.media.zlm.dto.MediaItem; -import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem; import com.genersoft.iot.vmp.service.IStreamPushService; -import com.genersoft.iot.vmp.service.bean.GPSMsgInfo; import com.genersoft.iot.vmp.service.bean.PushStreamStatusChangeFromRedisDto; -import com.genersoft.iot.vmp.service.bean.StreamPushItemFromRedis; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; -import com.genersoft.iot.vmp.storager.IVideoManagerStorage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; @@ -30,8 +17,6 @@ import org.springframework.data.redis.connection.MessageListener; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.stereotype.Component; -import java.util.ArrayList; -import java.util.List; import java.util.concurrent.ConcurrentLinkedQueue; @@ -65,7 +50,6 @@ public class RedisPushStreamStatusMsgListener implements MessageListener, Applic @Override public void onMessage(Message message, byte[] bytes) { - // TODO 增加队列 logger.warn("[REDIS消息-推流设备状态变化]： {}", new String(message.getBody())); taskQueue.offer(message); diff --git a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisStreamMsgListener.java b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisStreamMsgListener.java new file mode 100644 index 00000000..1897b6fc --- /dev/null +++ b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisStreamMsgListener.java @@ -0,0 +1,91 @@ +package com.genersoft.iot.vmp.service.redisMsg; + +import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson.JSONObject; +import com.genersoft.iot.vmp.conf.UserSetting; + +import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager; +import com.genersoft.iot.vmp.media.zlm.dto.MediaItem; +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.data.redis.connection.Message; +import org.springframework.data.redis.connection.MessageListener; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import org.springframework.stereotype.Component; + +import java.util.concurrent.ConcurrentLinkedQueue; + + +/** + * 接收其他wvp发送流变化通知 + * @author lin + */ +@Component +public class RedisStreamMsgListener implements MessageListener { + + private final static Logger logger = LoggerFactory.getLogger(RedisStreamMsgListener.class); + + @Autowired + private UserSetting userSetting; + + @Autowired + private ZLMMediaListManager zlmMediaListManager; + + private boolean taskQueueHandlerRun = false; + + private final ConcurrentLinkedQueue taskQueue = new ConcurrentLinkedQueue<>(); + + @Qualifier("taskExecutor") + @Autowired + private ThreadPoolTaskExecutor taskExecutor; + + @Override + public void onMessage(Message message, byte[] bytes) { + + taskQueue.offer(message); + if (!taskQueueHandlerRun) { + taskQueueHandlerRun = true; + taskExecutor.execute(() -> { + while (!taskQueue.isEmpty()) { + Message msg = taskQueue.poll(); + JSONObject steamMsgJson = JSON.parseObject(msg.getBody(), JSONObject.class); + if (steamMsgJson == null) { + logger.warn("[收到redis 流变化]消息解析失败"); + return; + } + String serverId = steamMsgJson.getString("serverId"); + + if (userSetting.getServerId().equals(serverId)) { + // 自己发送的消息忽略即可 + return; + } + logger.info("[收到redis 流变化]： {}", new String(message.getBody())); + String app = steamMsgJson.getString("app"); + String stream = steamMsgJson.getString("stream"); + boolean register = steamMsgJson.getBoolean("register"); + String mediaServerId = steamMsgJson.getString("mediaServerId"); + MediaItem mediaItem = new MediaItem(); + mediaItem.setSeverId(serverId); + mediaItem.setApp(app); + mediaItem.setStream(stream); + mediaItem.setRegist(register); + mediaItem.setMediaServerId(mediaServerId); + mediaItem.setCreateStamp(System.currentTimeMillis()/1000); + mediaItem.setAliveSecond(0L); + mediaItem.setTotalReaderCount("0"); + mediaItem.setOriginType(0); + mediaItem.setOriginTypeStr("0"); + mediaItem.setOriginTypeStr("unknown"); + if (register) { + zlmMediaListManager.addPush(mediaItem); + }else { + zlmMediaListManager.removeMedia(app, stream); + } + } + taskQueueHandlerRun = false; + }); + } + } +} diff --git a/src/main/java/com/genersoft/iot/vmp/storager/dao/GbStreamMapper.java b/src/main/java/com/genersoft/iot/vmp/storager/dao/GbStreamMapper.java index 5e74d99e..8a5d4550 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/dao/GbStreamMapper.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/dao/GbStreamMapper.java @@ -75,18 +75,23 @@ public interface GbStreamMapper { "WHERE gs.gbId = '${gbId}' AND pgs.platformId = '${platformId}'") GbStream queryStreamInPlatform(String platformId, String gbId); - @Select("select gt.gbId as channelId, gt.name, 'wvp-pro' as manufacture, st.status, gt.longitude, gt.latitude, pc.id as parentId," + + @Select("") + List queryGbStreamListInPlatform(String platformId, boolean usPushingAsStatus); @Select("SELECT gs.* FROM gb_stream gs LEFT JOIN platform_gb_stream pgs " + diff --git a/src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStorageImpl.java b/src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStorageImpl.java index 702b5bed..f8a74fef 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStorageImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStorageImpl.java @@ -2,6 +2,7 @@ package com.genersoft.iot.vmp.storager.impl; import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.conf.SipConfig; +import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.gb28181.event.EventPublisher; import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent; @@ -83,6 +84,9 @@ public class VideoManagerStorageImpl implements IVideoManagerStorage { @Autowired private GbStreamMapper gbStreamMapper; + @Autowired + private UserSetting userSetting; + @Autowired private PlatformCatalogMapper catalogMapper; @@ -614,7 +618,7 @@ public class VideoManagerStorageImpl implements IVideoManagerStorage { */ @Override public List queryGbStreamListInPlatform(String platformId) { - return gbStreamMapper.queryGbStreamListInPlatform(platformId); + return gbStreamMapper.queryGbStreamListInPlatform(platformId, userSetting.isUsePushingAsStatus()); } /** diff --git a/src/main/resources/all-application.yml b/src/main/resources/all-application.yml index 3569de2e..ed025b9a 100644 --- a/src/main/resources/all-application.yml +++ b/src/main/resources/all-application.yml @@ -188,6 +188,8 @@ user-settings: record-sip: true # 是否将日志存储进数据库 logInDatebase: true + # 使用推流状态作为推流通道状态 + use-pushing-as-status: true # 关闭在线文档（生产环境建议关闭） springdoc: