From 0ff4ed217de599e6c6336f0330787b7593641dc4 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: Mon, 17 Oct 2022 12:39:58 +0800 Subject: [PATCH 1/4] =?UTF-8?q?=E4=BC=98=E5=8C=96=E6=B6=88=E6=81=AF?= =?UTF-8?q?=E5=A4=84=E7=90=86=E4=B8=AD=E5=AD=98=E5=9C=A8=E5=8F=AF=E8=83=BD?= =?UTF-8?q?=E5=BC=82=E5=B8=B8=E7=9A=84=E5=A4=84=E7=90=86=E6=B5=81=E7=A8=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../request/impl/InviteRequestProcessor.java | 256 +++++++++++------- .../request/impl/NotifyRequestProcessor.java | 72 +++-- .../cmd/DeviceControlQueryMessageHandler.java | 2 +- .../cmd/KeepaliveNotifyMessageHandler.java | 49 ++-- .../MobilePositionNotifyMessageHandler.java | 16 +- .../query/cmd/CatalogQueryMessageHandler.java | 52 ++-- .../ConfigDownloadResponseMessageHandler.java | 23 +- .../DeviceControlResponseMessageHandler.java | 23 +- .../cmd/DeviceInfoResponseMessageHandler.java | 24 +- .../MobilePositionResponseMessageHandler.java | 15 +- .../PresetQueryResponseMessageHandler.java | 20 +- .../cmd/RecordInfoResponseMessageHandler.java | 168 ++++++------ 12 files changed, 414 insertions(+), 306 deletions(-) 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 cfca8e1a..42ae826c 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 @@ -132,7 +132,11 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements if (requesterId == null || channelId == null) { logger.info("无法从FromHeader的Address中获取到平台id，返回400"); // 参数不全， 发400，请求错误 - responseAck(serverTransaction, Response.BAD_REQUEST); + try { + responseAck(serverTransaction, Response.BAD_REQUEST); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[命令发送失败] invite BAD_REQUEST: {}", e.getMessage()); + } return; } @@ -141,6 +145,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements ParentPlatform platform = storager.queryParentPlatByServerGBId(requesterId); if (platform == null) { inviteFromDeviceHandle(serverTransaction, requesterId); + } else { // 查询平台下是否有该通道 DeviceChannel channel = storager.queryChannelInParentPlatform(requesterId, channelId); @@ -158,7 +163,11 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements // return; // } // 通道存在，发100，TRYING - responseAck(serverTransaction, Response.TRYING); + try { + responseAck(serverTransaction, Response.TRYING); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[命令发送失败] invite TRYING: {}", e.getMessage()); + } } else if (channel == null && gbStream != null) { String mediaServerId = gbStream.getMediaServerId(); @@ -166,13 +175,21 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements if (mediaServerItem == null) { if ("proxy".equals(gbStream.getStreamType())) { logger.info("[ app={}, stream={} ]找不到zlm {}，返回410", gbStream.getApp(), gbStream.getStream(), mediaServerId); - responseAck(serverTransaction, Response.GONE); + try { + responseAck(serverTransaction, Response.GONE); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[命令发送失败] invite GONE: {}", e.getMessage()); + } return; } else { streamPushItem = streamPushService.getPush(gbStream.getApp(), gbStream.getStream()); if (streamPushItem == null || streamPushItem.getServerId().equals(userSetting.getServerId())) { logger.info("[ app={}, stream={} ]找不到zlm {}，返回410", gbStream.getApp(), gbStream.getStream(), mediaServerId); - responseAck(serverTransaction, Response.GONE); + try { + responseAck(serverTransaction, Response.GONE); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[命令发送失败] invite GONE: {}", e.getMessage()); + } return; } } @@ -181,25 +198,47 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements streamPushItem = streamPushService.getPush(gbStream.getApp(), gbStream.getStream()); if (streamPushItem == null) { logger.info("[ app={}, stream={} ]找不到zlm {}，返回410", gbStream.getApp(), gbStream.getStream(), mediaServerId); - responseAck(serverTransaction, Response.GONE); + try { + responseAck(serverTransaction, Response.GONE); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[命令发送失败] invite GONE: {}", e.getMessage()); + } return; } }else if("proxy".equals(gbStream.getStreamType())){ proxyByAppAndStream = streamProxyService.getStreamProxyByAppAndStream(gbStream.getApp(), gbStream.getStream()); if (proxyByAppAndStream == null) { logger.info("[ app={}, stream={} ]找不到zlm {}，返回410", gbStream.getApp(), gbStream.getStream(), mediaServerId); - responseAck(serverTransaction, Response.GONE); + try { + responseAck(serverTransaction, Response.GONE); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[命令发送失败] invite GONE: {}", e.getMessage()); + } return; } } } - responseAck(serverTransaction, Response.CALL_IS_BEING_FORWARDED); // 通道存在，发181，呼叫转接中 + try { + responseAck(serverTransaction, Response.CALL_IS_BEING_FORWARDED); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[命令发送失败] invite CALL_IS_BEING_FORWARDED: {}", e.getMessage()); + } } else if (catalog != null) { - responseAck(serverTransaction, Response.BAD_REQUEST, "catalog channel can not play"); // 目录不支持点播 + try { + // 目录不支持点播 + responseAck(serverTransaction, Response.BAD_REQUEST, "catalog channel can not play"); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[命令发送失败] invite 目录不支持点播: {}", e.getMessage()); + } return; } else { logger.info("通道不存在，返回404"); - responseAck(serverTransaction, Response.NOT_FOUND); // 通道不存在，发404，资源不存在 + try { + // 通道不存在，发404，资源不存在 + responseAck(serverTransaction, Response.NOT_FOUND); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[命令发送失败] invite 通道不存在: {}", e.getMessage()); + } return; } // 解析sdp消息, 使用jainsip 自带的sdp解析方式 @@ -270,7 +309,12 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements if (port == -1) { logger.info("不支持的媒体格式，返回415"); // 回复不支持的格式 - responseAck(serverTransaction, Response.UNSUPPORTED_MEDIA_TYPE); // 不支持的格式，发415 + try { + // 不支持的格式，发415 + responseAck(serverTransaction, Response.UNSUPPORTED_MEDIA_TYPE); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[命令发送失败] invite 不支持的格式: {}", e.getMessage()); + } return; } String username = sdp.getOrigin().getUsername(); @@ -283,13 +327,21 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements device = storager.queryVideoDeviceByPlatformIdAndChannelId(requesterId, channelId); if (device == null) { logger.warn("点播平台{}的通道{}时未找到设备信息", requesterId, channel); - responseAck(serverTransaction, Response.SERVER_INTERNAL_ERROR); + try { + responseAck(serverTransaction, Response.SERVER_INTERNAL_ERROR); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[命令发送失败] invite 未找到设备信息: {}", e.getMessage()); + } return; } mediaServerItem = playService.getNewMediaServerItem(device); if (mediaServerItem == null) { logger.warn("未找到可用的zlm"); - responseAck(serverTransaction, Response.BUSY_HERE); + try { + responseAck(serverTransaction, Response.BUSY_HERE); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[命令发送失败] invite BUSY_HERE: {}", e.getMessage()); + } return; } SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId, @@ -301,7 +353,11 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements } if (sendRtpItem == null) { logger.warn("服务器端口资源不足"); - responseAck(serverTransaction, Response.BUSY_HERE); + try { + responseAck(serverTransaction, Response.BUSY_HERE); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[命令发送失败] invite 服务器端口资源不足: {}", e.getMessage()); + } return; } sendRtpItem.setCallId(callIdHeader.getCallId()); @@ -474,13 +530,8 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements } } } - - } catch (SipException | InvalidArgumentException | ParseException e) { - e.printStackTrace(); - logger.warn("sdp解析错误"); - e.printStackTrace(); } catch (SdpParseException e) { - e.printStackTrace(); + logger.error("sdp解析错误", e); } catch (SdpException e) { e.printStackTrace(); } @@ -492,7 +543,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements private void pushProxyStream(RequestEvent evt, ServerTransaction serverTransaction, GbStream gbStream, ParentPlatform platform, CallIdHeader callIdHeader, MediaServerItem mediaServerItem, int port, Boolean tcpActive, boolean mediaTransmissionTCP, - String channelId, String addressStr, String ssrc, String requesterId) throws InvalidArgumentException, ParseException, SipException { + String channelId, String addressStr, String ssrc, String requesterId) { Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, gbStream.getApp(), gbStream.getStream()); if (streamReady) { // 自平台内容 @@ -502,7 +553,11 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements if (sendRtpItem == null) { logger.warn("服务器端口资源不足"); - responseAck(serverTransaction, Response.BUSY_HERE); + try { + responseAck(serverTransaction, Response.BUSY_HERE); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[命令发送失败] invite 服务器端口资源不足: {}", e.getMessage()); + } return; } if (tcpActive != null) { @@ -527,7 +582,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements private void pushStream(RequestEvent evt, ServerTransaction serverTransaction, GbStream gbStream, StreamPushItem streamPushItem, ParentPlatform platform, CallIdHeader callIdHeader, MediaServerItem mediaServerItem, int port, Boolean tcpActive, boolean mediaTransmissionTCP, - String channelId, String addressStr, String ssrc, String requesterId) throws InvalidArgumentException, ParseException, SipException { + String channelId, String addressStr, String ssrc, String requesterId) { // 推流 if (streamPushItem.isSelf()) { Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, gbStream.getApp(), gbStream.getStream()); @@ -539,7 +594,11 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements if (sendRtpItem == null) { logger.warn("服务器端口资源不足"); - responseAck(serverTransaction, Response.BUSY_HERE); + try { + responseAck(serverTransaction, Response.BUSY_HERE); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[命令发送失败] invite 服务器端口资源不足: {}", e.getMessage()); + } return; } if (tcpActive != null) { @@ -577,15 +636,23 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements private void notifyStreamOnline(RequestEvent evt, ServerTransaction serverTransaction, GbStream gbStream, StreamPushItem streamPushItem, ParentPlatform platform, CallIdHeader callIdHeader, MediaServerItem mediaServerItem, int port, Boolean tcpActive, boolean mediaTransmissionTCP, - String channelId, String addressStr, String ssrc, String requesterId) throws InvalidArgumentException, ParseException, SipException { + String channelId, String addressStr, String ssrc, String requesterId) { if ("proxy".equals(gbStream.getStreamType())) { // TODO 控制启用以使设备上线 logger.info("[ app={}, stream={} ]通道未推流，启用流后开始推流", gbStream.getApp(), gbStream.getStream()); - responseAck(serverTransaction, Response.BAD_REQUEST, "channel [" + gbStream.getGbId() + "] offline"); + try { + responseAck(serverTransaction, Response.BAD_REQUEST, "channel [" + gbStream.getGbId() + "] offline"); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[命令发送失败] invite 通道未推流: {}", e.getMessage()); + } } else if ("push".equals(gbStream.getStreamType())) { if (!platform.isStartOfflinePush()) { // 平台设置中关闭了拉起离线的推流则直接回复 - responseAck(serverTransaction, Response.TEMPORARILY_UNAVAILABLE, "channel stream not pushing"); + try { + responseAck(serverTransaction, Response.TEMPORARILY_UNAVAILABLE, "channel stream not pushing"); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[命令发送失败] invite 通道未推流: {}", e.getMessage()); + } return; } // 发送redis消息以使设备上线 @@ -713,38 +780,28 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements } redisCatchStorage.updateSendRTPSever(sendRtpItem); }, (wvpResult) -> { - try { - // 错误 - if (wvpResult.getCode() == RedisGbPlayMsgListener.ERROR_CODE_OFFLINE) { - // 离线 - // 查询是否在本机上线了 - StreamPushItem currentStreamPushItem = streamPushService.getPush(streamPushItem.getApp(), streamPushItem.getStream()); - if (currentStreamPushItem.isPushIng()) { - // 在线状态 - pushStream(evt, serverTransaction, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive, - mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId); - } else { - // 不在线 拉起 - notifyStreamOnline(evt, serverTransaction, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive, - mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId); - } + // 错误 + if (wvpResult.getCode() == RedisGbPlayMsgListener.ERROR_CODE_OFFLINE) { + // 离线 + // 查询是否在本机上线了 + StreamPushItem currentStreamPushItem = streamPushService.getPush(streamPushItem.getApp(), streamPushItem.getStream()); + if (currentStreamPushItem.isPushIng()) { + // 在线状态 + pushStream(evt, serverTransaction, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive, + mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId); + + } else { + // 不在线 拉起 + notifyStreamOnline(evt, serverTransaction, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive, + mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId); } - } catch (InvalidArgumentException | ParseException | SipException e) { - logger.error("[命令发送失败] 国标级联 点播回复: {}", e.getMessage()); } - - try { responseAck(serverTransaction, Response.BUSY_HERE); - } catch (SipException e) { - e.printStackTrace(); - } catch (InvalidArgumentException e) { - e.printStackTrace(); - } catch (ParseException e) { - e.printStackTrace(); + } catch (InvalidArgumentException | ParseException | SipException e) { + logger.error("[命令发送失败] 国标级联 点播回复 BUSY_HERE: {}", e.getMessage()); } - return; }); } @@ -782,14 +839,17 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements return null; } - public void inviteFromDeviceHandle(ServerTransaction serverTransaction, String requesterId) throws InvalidArgumentException, ParseException, SipException, SdpException { + public void inviteFromDeviceHandle(ServerTransaction serverTransaction, String requesterId) { // 非上级平台请求，查询是否设备请求（通常为接收语音广播的设备） Device device = redisCatchStorage.getDevice(requesterId); if (device != null) { logger.info("收到设备" + requesterId + "的语音广播Invite请求"); - responseAck(serverTransaction, Response.TRYING); - + try { + responseAck(serverTransaction, Response.TRYING); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[命令发送失败] invite BAD_REQUEST: {}", e.getMessage()); + } String contentString = new String(serverTransaction.getRequest().getRawContent()); // jainSip不支持y=字段， 移除移除以解析。 String substring = contentString; @@ -803,51 +863,65 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements if (ssrcIndex > 0) { substring = contentString.substring(0, ssrcIndex); } - SessionDescription sdp = SdpFactory.getInstance().createSessionDescription(substring); + SessionDescription sdp = null; + try { + sdp = SdpFactory.getInstance().createSessionDescription(substring); + // 获取支持的格式 + Vector mediaDescriptions = sdp.getMediaDescriptions(true); + // 查看是否支持PS 负载96 + int port = -1; + //boolean recvonly = false; + boolean mediaTransmissionTCP = false; + Boolean tcpActive = null; + for (int i = 0; i < mediaDescriptions.size(); i++) { + MediaDescription mediaDescription = (MediaDescription) mediaDescriptions.get(i); + Media media = mediaDescription.getMedia(); - // 获取支持的格式 - Vector mediaDescriptions = sdp.getMediaDescriptions(true); - // 查看是否支持PS 负载96 - int port = -1; - //boolean recvonly = false; - boolean mediaTransmissionTCP = false; - Boolean tcpActive = null; - for (int i = 0; i < mediaDescriptions.size(); i++) { - MediaDescription mediaDescription = (MediaDescription) mediaDescriptions.get(i); - Media media = mediaDescription.getMedia(); - - Vector mediaFormats = media.getMediaFormats(false); - if (mediaFormats.contains("8")) { - port = media.getMediaPort(); - String protocol = media.getProtocol(); - // 区分TCP发流还是udp， 当前默认udp - if ("TCP/RTP/AVP".equals(protocol)) { - String setup = mediaDescription.getAttribute("setup"); - if (setup != null) { - mediaTransmissionTCP = true; - if ("active".equals(setup)) { - tcpActive = true; - } else if ("passive".equals(setup)) { - tcpActive = false; + Vector mediaFormats = media.getMediaFormats(false); + if (mediaFormats.contains("8")) { + port = media.getMediaPort(); + String protocol = media.getProtocol(); + // 区分TCP发流还是udp， 当前默认udp + if ("TCP/RTP/AVP".equals(protocol)) { + String setup = mediaDescription.getAttribute("setup"); + if (setup != null) { + mediaTransmissionTCP = true; + if ("active".equals(setup)) { + tcpActive = true; + } else if ("passive".equals(setup)) { + tcpActive = false; + } } } + break; } - break; } + if (port == -1) { + logger.info("不支持的媒体格式，返回415"); + // 回复不支持的格式 + try { + responseAck(serverTransaction, Response.UNSUPPORTED_MEDIA_TYPE); // 不支持的格式，发415 + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[命令发送失败] invite 不支持的媒体格式，返回415， {}", e.getMessage()); + } + return; + } + String username = sdp.getOrigin().getUsername(); + String addressStr = sdp.getOrigin().getAddress(); + logger.info("设备{}请求语音流，地址：{}:{}，ssrc：{}", username, addressStr, port, ssrc); + } catch (SdpException e) { + logger.error("[SDP解析异常]", e); } - if (port == -1) { - logger.info("不支持的媒体格式，返回415"); - // 回复不支持的格式 - responseAck(serverTransaction, Response.UNSUPPORTED_MEDIA_TYPE); // 不支持的格式，发415 - return; - } - String username = sdp.getOrigin().getUsername(); - String addressStr = sdp.getOrigin().getAddress(); - logger.info("设备{}请求语音流，地址：{}:{}，ssrc：{}", username, addressStr, port, ssrc); + + } else { logger.warn("来自无效设备/平台的请求"); - responseAck(serverTransaction, Response.BAD_REQUEST); + try { + responseAck(serverTransaction, Response.BAD_REQUEST);; // 不支持的格式，发415 + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[命令发送失败] invite 来自无效设备/平台的请求， {}", e.getMessage()); + } } } } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestProcessor.java index f78ee181..930ddb59 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestProcessor.java @@ -93,46 +93,44 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements @Override public void process(RequestEvent evt) { + ServerTransaction serverTransaction = getServerTransaction(evt); try { - taskQueue.offer(new HandlerCatchData(evt, null, null)); - ServerTransaction serverTransaction = getServerTransaction(evt); responseAck(serverTransaction, Response.OK); - if (!taskQueueHandlerRun) { - taskQueueHandlerRun = true; - taskExecutor.execute(()-> { - while (!taskQueue.isEmpty()) { - try { - HandlerCatchData take = taskQueue.poll(); - Element rootElement = getRootElement(take.getEvt()); - if (rootElement == null) { - logger.error("处理NOTIFY消息时未获取到消息体,{}", take.getEvt().getRequest()); - continue; - } - String cmd = XmlUtil.getText(rootElement, "CmdType"); - - if (CmdType.CATALOG.equals(cmd)) { - logger.info("接收到Catalog通知"); - processNotifyCatalogList(take.getEvt()); - } else if (CmdType.ALARM.equals(cmd)) { - logger.info("接收到Alarm通知"); - processNotifyAlarm(take.getEvt()); - } else if (CmdType.MOBILE_POSITION.equals(cmd)) { - logger.info("接收到MobilePosition通知"); - processNotifyMobilePosition(take.getEvt()); - } else { - logger.info("接收到消息：" + cmd); - } - } catch (DocumentException e) { - logger.error("处理NOTIFY消息时错误", e); - } - } - taskQueueHandlerRun = false; - }); - } - } catch (SipException | InvalidArgumentException | ParseException e) { + }catch (SipException | InvalidArgumentException | ParseException e) { e.printStackTrace(); - } finally { - taskQueueHandlerRun = false; + } + taskQueue.offer(new HandlerCatchData(evt, null, null)); + if (!taskQueueHandlerRun) { + taskQueueHandlerRun = true; + taskExecutor.execute(()-> { + while (!taskQueue.isEmpty()) { + try { + HandlerCatchData take = taskQueue.poll(); + Element rootElement = getRootElement(take.getEvt()); + if (rootElement == null) { + logger.error("处理NOTIFY消息时未获取到消息体,{}", take.getEvt().getRequest()); + continue; + } + String cmd = XmlUtil.getText(rootElement, "CmdType"); + + if (CmdType.CATALOG.equals(cmd)) { + logger.info("接收到Catalog通知"); + processNotifyCatalogList(take.getEvt()); + } else if (CmdType.ALARM.equals(cmd)) { + logger.info("接收到Alarm通知"); + processNotifyAlarm(take.getEvt()); + } else if (CmdType.MOBILE_POSITION.equals(cmd)) { + logger.info("接收到MobilePosition通知"); + processNotifyMobilePosition(take.getEvt()); + } else { + logger.info("接收到消息：" + cmd); + } + } catch (DocumentException e) { + logger.error("处理NOTIFY消息时错误", e); + } + } + taskQueueHandlerRun = false; + }); } } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/control/cmd/DeviceControlQueryMessageHandler.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/control/cmd/DeviceControlQueryMessageHandler.java index 855ad3b2..cd70dd0d 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/control/cmd/DeviceControlQueryMessageHandler.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/control/cmd/DeviceControlQueryMessageHandler.java @@ -112,10 +112,10 @@ public class DeviceControlQueryMessageHandler extends SIPRequestProcessorParent if (deviceForPlatform == null) { try { responseAck(serverTransaction, Response.NOT_FOUND); - return; } catch (SipException | InvalidArgumentException | ParseException e) { logger.error("[命令发送失败] 错误信息: {}", e.getMessage()); } + return; } try { cmder.fronEndCmd(deviceForPlatform, channelId, cmdString, eventResult -> { diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/KeepaliveNotifyMessageHandler.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/KeepaliveNotifyMessageHandler.java index 289f1628..cd0dcd13 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/KeepaliveNotifyMessageHandler.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/KeepaliveNotifyMessageHandler.java @@ -52,35 +52,36 @@ public class KeepaliveNotifyMessageHandler extends SIPRequestProcessorParent imp // 未注册的设备不做处理 return; } + // 回复200 OK try { - // 判断RPort是否改变，改变则说明路由nat信息变化，修改设备信息 - // 获取到通信地址等信息 - ViaHeader viaHeader = (ViaHeader) evt.getRequest().getHeader(ViaHeader.NAME); - String received = viaHeader.getReceived(); - int rPort = viaHeader.getRPort(); - // 解析本地地址替代 - if (ObjectUtils.isEmpty(received) || rPort == -1) { - received = viaHeader.getHost(); - rPort = viaHeader.getPort(); - } - if (device.getPort() != rPort) { - device.setPort(rPort); - device.setHostAddress(received.concat(":").concat(String.valueOf(rPort))); - } - device.setKeepaliveTime(DateUtil.getNow()); - // 回复200 OK responseAck(getServerTransaction(evt), Response.OK); - if (device.getOnline() == 1) { - deviceService.updateDevice(device); - }else { - // 对于已经离线的设备判断他的注册是否已经过期 - if (!deviceService.expire(device)){ - deviceService.online(device); - } - } } catch (SipException | InvalidArgumentException | ParseException e) { logger.error("[命令发送失败] 国标级联 心跳回复: {}", e.getMessage()); } + // 判断RPort是否改变，改变则说明路由nat信息变化，修改设备信息 + // 获取到通信地址等信息 + ViaHeader viaHeader = (ViaHeader) evt.getRequest().getHeader(ViaHeader.NAME); + String received = viaHeader.getReceived(); + int rPort = viaHeader.getRPort(); + // 解析本地地址替代 + if (ObjectUtils.isEmpty(received) || rPort == -1) { + received = viaHeader.getHost(); + rPort = viaHeader.getPort(); + } + if (device.getPort() != rPort) { + device.setPort(rPort); + device.setHostAddress(received.concat(":").concat(String.valueOf(rPort))); + } + device.setKeepaliveTime(DateUtil.getNow()); + + if (device.getOnline() == 1) { + deviceService.updateDevice(device); + }else { + // 对于已经离线的设备判断他的注册是否已经过期 + if (!deviceService.expire(device)){ + deviceService.online(device); + } + } } @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 652cd830..82cd5528 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 @@ -81,8 +81,12 @@ public class MobilePositionNotifyMessageHandler extends SIPRequestProcessorParen 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); + try { + logger.warn("[ 移动设备位置数据通知 ] content cannot be null, {}", sipMsgInfo.getEvt().getRequest()); + responseAck(getServerTransaction(sipMsgInfo.getEvt()), Response.BAD_REQUEST); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[命令发送失败] 移动设备位置数据通知 内容为空: {}", e.getMessage()); + } continue; } MobilePosition mobilePosition = new MobilePosition(); @@ -133,7 +137,11 @@ public class MobilePositionNotifyMessageHandler extends SIPRequestProcessorParen } storager.updateChannelPosition(deviceChannel); //回复 200 OK - responseAck(getServerTransaction(sipMsgInfo.getEvt()), Response.OK); + try { + responseAck(getServerTransaction(sipMsgInfo.getEvt()), Response.OK); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[命令发送失败] 移动设备位置数据回复200: {}", e.getMessage()); + } // 发送redis消息。 通知位置信息的变化 JSONObject jsonObject = new JSONObject(); @@ -147,7 +155,7 @@ public class MobilePositionNotifyMessageHandler extends SIPRequestProcessorParen jsonObject.put("speed", mobilePosition.getSpeed()); redisCatchStorage.sendMobilePositionMsg(jsonObject); - } catch (DocumentException | SipException | InvalidArgumentException | ParseException e) { + } catch (DocumentException e) { e.printStackTrace(); } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/query/cmd/CatalogQueryMessageHandler.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/query/cmd/CatalogQueryMessageHandler.java index 82f4a257..9e79f3f0 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/query/cmd/CatalogQueryMessageHandler.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/query/cmd/CatalogQueryMessageHandler.java @@ -67,33 +67,37 @@ public class CatalogQueryMessageHandler extends SIPRequestProcessorParent implem try { // 回复200 OK responseAck(getServerTransaction(evt), Response.OK); - Element snElement = rootElement.element("SN"); - String sn = snElement.getText(); - // 准备回复通道信息 - List deviceChannelInPlatforms = storager.queryChannelWithCatalog(parentPlatform.getServerGBId()); - // 查询关联的直播通道 - List gbStreams = storager.queryGbStreamListInPlatform(parentPlatform.getServerGBId()); - // 回复目录信息 - List catalogs = storager.queryCatalogInPlatform(parentPlatform.getServerGBId()); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[命令发送失败] 国标级联 目录查询回复200OK: {}", e.getMessage()); + } + Element snElement = rootElement.element("SN"); + String sn = snElement.getText(); + // 准备回复通道信息 + List deviceChannelInPlatforms = storager.queryChannelWithCatalog(parentPlatform.getServerGBId()); + // 查询关联的直播通道 + List gbStreams = storager.queryGbStreamListInPlatform(parentPlatform.getServerGBId()); + // 回复目录信息 + List catalogs = storager.queryCatalogInPlatform(parentPlatform.getServerGBId()); - List allChannels = new ArrayList<>(); + List allChannels = new ArrayList<>(); - // 回复平台 + // 回复平台 // DeviceChannel deviceChannel = getChannelForPlatform(parentPlatform); // allChannels.add(deviceChannel); - // 回复目录 - if (catalogs.size() > 0) { - allChannels.addAll(catalogs); - } - // 回复级联的通道 - if (deviceChannelInPlatforms.size() > 0) { - allChannels.addAll(deviceChannelInPlatforms); - } - // 回复直播的通道 - if (gbStreams.size() > 0) { - allChannels.addAll(gbStreams); - } + // 回复目录 + if (catalogs.size() > 0) { + allChannels.addAll(catalogs); + } + // 回复级联的通道 + if (deviceChannelInPlatforms.size() > 0) { + allChannels.addAll(deviceChannelInPlatforms); + } + // 回复直播的通道 + if (gbStreams.size() > 0) { + allChannels.addAll(gbStreams); + } + try { if (allChannels.size() > 0) { cmderFroPlatform.catalogQuery(allChannels, parentPlatform, sn, fromHeader.getTag()); }else { @@ -101,9 +105,11 @@ public class CatalogQueryMessageHandler extends SIPRequestProcessorParent implem cmderFroPlatform.catalogQuery(null, parentPlatform, sn, fromHeader.getTag(), 0); } } catch (SipException | InvalidArgumentException | ParseException e) { - logger.error("[命令发送失败] 国标级联 目录查询: {}", e.getMessage()); + logger.error("[命令发送失败] 国标级联 目录查询回复: {}", e.getMessage()); } + + } private DeviceChannel getChannelForPlatform(ParentPlatform platform) { diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/ConfigDownloadResponseMessageHandler.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/ConfigDownloadResponseMessageHandler.java index 1c4aa8a8..175b89bb 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/ConfigDownloadResponseMessageHandler.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/ConfigDownloadResponseMessageHandler.java @@ -53,19 +53,20 @@ public class ConfigDownloadResponseMessageHandler extends SIPRequestProcessorPar try { // 回复200 OK responseAck(getServerTransaction(evt), Response.OK); - // 此处是对本平台发出DeviceControl指令的应答 - JSONObject json = new JSONObject(); - XmlUtil.node2Json(element, json); - if (logger.isDebugEnabled()) { - logger.debug(json.toJSONString()); - } - RequestMessage msg = new RequestMessage(); - msg.setKey(key); - msg.setData(json); - deferredResultHolder.invokeAllResult(msg); } catch (SipException | InvalidArgumentException | ParseException e) { - logger.error("[命令发送失败] 国标级联 设备配置查询: {}", e.getMessage()); + logger.error("[命令发送失败] 设备配置查询: {}", e.getMessage()); } + // 此处是对本平台发出DeviceControl指令的应答 + JSONObject json = new JSONObject(); + XmlUtil.node2Json(element, json); + if (logger.isDebugEnabled()) { + logger.debug(json.toJSONString()); + } + RequestMessage msg = new RequestMessage(); + msg.setKey(key); + msg.setData(json); + deferredResultHolder.invokeAllResult(msg); + } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/DeviceControlResponseMessageHandler.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/DeviceControlResponseMessageHandler.java index 12c84686..d068aeff 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/DeviceControlResponseMessageHandler.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/DeviceControlResponseMessageHandler.java @@ -47,20 +47,21 @@ public class DeviceControlResponseMessageHandler extends SIPRequestProcessorPare // 此处是对本平台发出DeviceControl指令的应答 try { responseAck(getServerTransaction(evt), Response.OK); - JSONObject json = new JSONObject(); - String channelId = getText(element, "DeviceID"); - XmlUtil.node2Json(element, json); - if (logger.isDebugEnabled()) { - logger.debug(json.toJSONString()); - } - RequestMessage msg = new RequestMessage(); - String key = DeferredResultHolder.CALLBACK_CMD_DEVICECONTROL + device.getDeviceId() + channelId; - msg.setKey(key); - msg.setData(json); - deferredResultHolder.invokeAllResult(msg); } catch (SipException | InvalidArgumentException | ParseException e) { logger.error("[命令发送失败] 国标级联 设备控制: {}", e.getMessage()); } + JSONObject json = new JSONObject(); + String channelId = getText(element, "DeviceID"); + XmlUtil.node2Json(element, json); + if (logger.isDebugEnabled()) { + logger.debug(json.toJSONString()); + } + RequestMessage msg = new RequestMessage(); + String key = DeferredResultHolder.CALLBACK_CMD_DEVICECONTROL + device.getDeviceId() + channelId; + msg.setKey(key); + msg.setData(json); + deferredResultHolder.invokeAllResult(msg); + } @Override diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/DeviceInfoResponseMessageHandler.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/DeviceInfoResponseMessageHandler.java index a0109edd..1544289b 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/DeviceInfoResponseMessageHandler.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/DeviceInfoResponseMessageHandler.java @@ -78,9 +78,14 @@ public class DeviceInfoResponseMessageHandler extends SIPRequestProcessorParent ServerTransaction serverTransaction = getServerTransaction(evt); try { rootElement = getRootElement(evt, device.getCharset()); - if (rootElement == null) { + + if (rootElement == null) { logger.warn("[ 接收到DeviceInfo应答消息 ] content cannot be null, {}", evt.getRequest()); - responseAck(serverTransaction, Response.BAD_REQUEST); + try { + responseAck(serverTransaction, Response.BAD_REQUEST); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[命令发送失败] DeviceInfo应答消息 BAD_REQUEST: {}", e.getMessage()); + } return; } Element deviceIdElement = rootElement.element("DeviceID"); @@ -100,17 +105,16 @@ public class DeviceInfoResponseMessageHandler extends SIPRequestProcessorParent msg.setKey(key); msg.setData(device); deferredResultHolder.invokeAllResult(msg); + } catch (DocumentException e) { + throw new RuntimeException(e); + } + try { // 回复200 OK responseAck(serverTransaction, Response.OK); - } catch (DocumentException e) { - e.printStackTrace(); - } catch (InvalidArgumentException e) { - e.printStackTrace(); - } catch (ParseException e) { - e.printStackTrace(); - } catch (SipException e) { - e.printStackTrace(); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[命令发送失败] DeviceInfo应答消息 200: {}", e.getMessage()); } + } @Override diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/MobilePositionResponseMessageHandler.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/MobilePositionResponseMessageHandler.java index 158f5b71..5e6c39da 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/MobilePositionResponseMessageHandler.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/MobilePositionResponseMessageHandler.java @@ -71,7 +71,11 @@ public class MobilePositionResponseMessageHandler extends SIPRequestProcessorPar rootElement = getRootElement(evt, device.getCharset()); if (rootElement == null) { logger.warn("[ 移动设备位置数据查询回复 ] content cannot be null, {}", evt.getRequest()); - responseAck(serverTransaction, Response.BAD_REQUEST); + try { + responseAck(serverTransaction, Response.BAD_REQUEST); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[命令发送失败] 移动设备位置数据查询 BAD_REQUEST: {}", e.getMessage()); + } return; } MobilePosition mobilePosition = new MobilePosition(); @@ -133,8 +137,13 @@ public class MobilePositionResponseMessageHandler extends SIPRequestProcessorPar jsonObject.put("speed", mobilePosition.getSpeed()); redisCatchStorage.sendMobilePositionMsg(jsonObject); //回复 200 OK - responseAck(serverTransaction, Response.OK); - } catch (DocumentException | SipException | InvalidArgumentException | ParseException e) { + try { + responseAck(serverTransaction, Response.OK); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[命令发送失败] 移动设备位置数据查询 200: {}", e.getMessage()); + } + + } catch (DocumentException e) { e.printStackTrace(); } } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/PresetQueryResponseMessageHandler.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/PresetQueryResponseMessageHandler.java index 366c7add..92091833 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/PresetQueryResponseMessageHandler.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/PresetQueryResponseMessageHandler.java @@ -58,7 +58,11 @@ public class PresetQueryResponseMessageHandler extends SIPRequestProcessorParent if (rootElement == null) { logger.warn("[ 设备预置位查询应答 ] content cannot be null, {}", evt.getRequest()); - responseAck(serverTransaction, Response.BAD_REQUEST); + try { + responseAck(serverTransaction, Response.BAD_REQUEST); + } catch (InvalidArgumentException | ParseException | SipException e) { + logger.error("[命令发送失败] 设备预置位查询应答处理: {}", e.getMessage()); + } return; } Element presetListNumElement = rootElement.element("PresetList"); @@ -67,7 +71,11 @@ public class PresetQueryResponseMessageHandler extends SIPRequestProcessorParent String deviceId = getText(rootElement, "DeviceID"); String key = DeferredResultHolder.CALLBACK_CMD_PRESETQUERY + deviceId; if (snElement == null || presetListNumElement == null) { - responseAck(serverTransaction, Response.BAD_REQUEST, "xml error"); + try { + responseAck(serverTransaction, Response.BAD_REQUEST, "xml error"); + } catch (InvalidArgumentException | ParseException | SipException e) { + logger.error("[命令发送失败] 设备预置位查询应答处理: {}", e.getMessage()); + } return; } int sumNum = Integer.parseInt(presetListNumElement.attributeValue("Num")); @@ -94,11 +102,13 @@ public class PresetQueryResponseMessageHandler extends SIPRequestProcessorParent requestMessage.setKey(key); requestMessage.setData(presetQuerySipReqList); deferredResultHolder.invokeAllResult(requestMessage); - responseAck(serverTransaction, Response.OK); + try { + responseAck(serverTransaction, Response.OK); + } catch (InvalidArgumentException | ParseException | SipException e) { + logger.error("[命令发送失败] 设备预置位查询应答处理: {}", e.getMessage()); + } } catch (DocumentException e) { logger.error("[解析xml]失败: ", e); - } catch (InvalidArgumentException | ParseException | SipException e) { - logger.error("[命令发送失败] 设备预置位查询应答处理: {}", e.getMessage()); } } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/RecordInfoResponseMessageHandler.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/RecordInfoResponseMessageHandler.java index 50a4ed8d..3c755367 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/RecordInfoResponseMessageHandler.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/RecordInfoResponseMessageHandler.java @@ -69,95 +69,91 @@ public class RecordInfoResponseMessageHandler extends SIPRequestProcessorParent @Override public void handForDevice(RequestEvent evt, Device device, Element rootElement) { - - // 回复200 OK try { + // 回复200 OK responseAck(getServerTransaction(evt), Response.OK); - taskQueue.offer(new HandlerCatchData(evt, device, rootElement)); - if (!taskQueueHandlerRun) { - taskQueueHandlerRun = true; - taskExecutor.execute(()->{ - while (!taskQueue.isEmpty()) { - try { - HandlerCatchData take = taskQueue.poll(); - Element rootElementForCharset = getRootElement(take.getEvt(), take.getDevice().getCharset()); - if (rootElement == null) { - logger.warn("[ 国标录像 ] content cannot be null, {}", evt.getRequest()); - continue; - } - String sn = getText(rootElementForCharset, "SN"); - String channelId = getText(rootElementForCharset, "DeviceID"); - RecordInfo recordInfo = new RecordInfo(); - recordInfo.setChannelId(channelId); - recordInfo.setDeviceId(take.getDevice().getDeviceId()); - recordInfo.setSn(sn); - recordInfo.setName(getText(rootElementForCharset, "Name")); - String sumNumStr = getText(rootElementForCharset, "SumNum"); - int sumNum = 0; - if (!ObjectUtils.isEmpty(sumNumStr)) { - sumNum = Integer.parseInt(sumNumStr); - } - recordInfo.setSumNum(sumNum); - Element recordListElement = rootElementForCharset.element("RecordList"); - if (recordListElement == null || sumNum == 0) { - logger.info("无录像数据"); - eventPublisher.recordEndEventPush(recordInfo); - recordDataCatch.put(take.getDevice().getDeviceId(), sn, sumNum, new ArrayList<>()); - releaseRequest(take.getDevice().getDeviceId(), sn); - } else { - Iterator recordListIterator = recordListElement.elementIterator(); - if (recordListIterator != null) { - List recordList = new ArrayList<>(); - // 遍历DeviceList - while (recordListIterator.hasNext()) { - Element itemRecord = recordListIterator.next(); - Element recordElement = itemRecord.element("DeviceID"); - if (recordElement == null) { - logger.info("记录为空，下一个..."); - continue; - } - RecordItem record = new RecordItem(); - record.setDeviceId(getText(itemRecord, "DeviceID")); - record.setName(getText(itemRecord, "Name")); - record.setFilePath(getText(itemRecord, "FilePath")); - record.setFileSize(getText(itemRecord, "FileSize")); - record.setAddress(getText(itemRecord, "Address")); - - String startTimeStr = getText(itemRecord, "StartTime"); - record.setStartTime(DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(startTimeStr)); - - String endTimeStr = getText(itemRecord, "EndTime"); - record.setEndTime(DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(endTimeStr)); - - record.setSecrecy(itemRecord.element("Secrecy") == null ? 0 - : Integer.parseInt(getText(itemRecord, "Secrecy"))); - record.setType(getText(itemRecord, "Type")); - record.setRecorderId(getText(itemRecord, "RecorderID")); - recordList.add(record); - } - recordInfo.setRecordList(recordList); - // 发送消息，如果是上级查询此录像，则会通过这里通知给上级 - eventPublisher.recordEndEventPush(recordInfo); - int count = recordDataCatch.put(take.getDevice().getDeviceId(), sn, sumNum, recordList); - logger.info("[国标录像]， {}->{}: {}/{}", take.getDevice().getDeviceId(), sn, count, sumNum); - } - - if (recordDataCatch.isComplete(take.getDevice().getDeviceId(), sn)){ - releaseRequest(take.getDevice().getDeviceId(), sn); - } - } - } catch (DocumentException e) { - logger.error("xml解析异常： ", e); - } - } - taskQueueHandlerRun = false; - }); - } - - } catch (SipException | InvalidArgumentException | ParseException e) { + }catch (SipException | InvalidArgumentException | ParseException e) { logger.error("[命令发送失败] 国标级联 国标录像: {}", e.getMessage()); - } finally { - taskQueueHandlerRun = false; + } + taskQueue.offer(new HandlerCatchData(evt, device, rootElement)); + if (!taskQueueHandlerRun) { + taskQueueHandlerRun = true; + taskExecutor.execute(()->{ + while (!taskQueue.isEmpty()) { + try { + HandlerCatchData take = taskQueue.poll(); + Element rootElementForCharset = getRootElement(take.getEvt(), take.getDevice().getCharset()); + if (rootElement == null) { + logger.warn("[ 国标录像 ] content cannot be null, {}", evt.getRequest()); + continue; + } + String sn = getText(rootElementForCharset, "SN"); + String channelId = getText(rootElementForCharset, "DeviceID"); + RecordInfo recordInfo = new RecordInfo(); + recordInfo.setChannelId(channelId); + recordInfo.setDeviceId(take.getDevice().getDeviceId()); + recordInfo.setSn(sn); + recordInfo.setName(getText(rootElementForCharset, "Name")); + String sumNumStr = getText(rootElementForCharset, "SumNum"); + int sumNum = 0; + if (!ObjectUtils.isEmpty(sumNumStr)) { + sumNum = Integer.parseInt(sumNumStr); + } + recordInfo.setSumNum(sumNum); + Element recordListElement = rootElementForCharset.element("RecordList"); + if (recordListElement == null || sumNum == 0) { + logger.info("无录像数据"); + eventPublisher.recordEndEventPush(recordInfo); + recordDataCatch.put(take.getDevice().getDeviceId(), sn, sumNum, new ArrayList<>()); + releaseRequest(take.getDevice().getDeviceId(), sn); + } else { + Iterator recordListIterator = recordListElement.elementIterator(); + if (recordListIterator != null) { + List recordList = new ArrayList<>(); + // 遍历DeviceList + while (recordListIterator.hasNext()) { + Element itemRecord = recordListIterator.next(); + Element recordElement = itemRecord.element("DeviceID"); + if (recordElement == null) { + logger.info("记录为空，下一个..."); + continue; + } + RecordItem record = new RecordItem(); + record.setDeviceId(getText(itemRecord, "DeviceID")); + record.setName(getText(itemRecord, "Name")); + record.setFilePath(getText(itemRecord, "FilePath")); + record.setFileSize(getText(itemRecord, "FileSize")); + record.setAddress(getText(itemRecord, "Address")); + + String startTimeStr = getText(itemRecord, "StartTime"); + record.setStartTime(DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(startTimeStr)); + + String endTimeStr = getText(itemRecord, "EndTime"); + record.setEndTime(DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(endTimeStr)); + + record.setSecrecy(itemRecord.element("Secrecy") == null ? 0 + : Integer.parseInt(getText(itemRecord, "Secrecy"))); + record.setType(getText(itemRecord, "Type")); + record.setRecorderId(getText(itemRecord, "RecorderID")); + recordList.add(record); + } + recordInfo.setRecordList(recordList); + // 发送消息，如果是上级查询此录像，则会通过这里通知给上级 + eventPublisher.recordEndEventPush(recordInfo); + int count = recordDataCatch.put(take.getDevice().getDeviceId(), sn, sumNum, recordList); + logger.info("[国标录像]， {}->{}: {}/{}", take.getDevice().getDeviceId(), sn, count, sumNum); + } + + if (recordDataCatch.isComplete(take.getDevice().getDeviceId(), sn)){ + releaseRequest(take.getDevice().getDeviceId(), sn); + } + } + } catch (DocumentException e) { + logger.error("xml解析异常： ", e); + } + } + taskQueueHandlerRun = false; + }); } } From 2f1161819919214ed3c4eb8f094ef1839671ee97 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: Mon, 17 Oct 2022 14:08:44 +0800 Subject: [PATCH 2/4] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E7=94=A8=E6=88=B7?= =?UTF-8?q?=E5=88=A0=E9=99=A4=E5=A4=B1=E8=B4=A5=EF=BC=8C#636?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../com/genersoft/iot/vmp/vmanager/user/UserController.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/vmanager/user/UserController.java b/src/main/java/com/genersoft/iot/vmp/vmanager/user/UserController.java index 0052fd3b..6ebb7561 100644 --- a/src/main/java/com/genersoft/iot/vmp/vmanager/user/UserController.java +++ b/src/main/java/com/genersoft/iot/vmp/vmanager/user/UserController.java @@ -125,8 +125,8 @@ public class UserController { } } - @DeleteMapping("/删除用户") - @Operation(summary = "停止视频回放") + @DeleteMapping("/delete") + @Operation(summary = "删除用户") @Parameter(name = "id", description = "用户Id", required = true) public void delete(@RequestParam Integer id){ // 获取当前登录用户id From 1e1364e51a66614469b91e396d33a54bdf8f6e98 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: Mon, 17 Oct 2022 14:33:25 +0800 Subject: [PATCH 3/4] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E5=BC=80=E5=A7=8B?= =?UTF-8?q?=E7=82=B9=E6=92=AD=E5=8A=9F=E8=83=BD=E6=8E=A5=E5=8F=A3=E8=B0=83?= =?UTF-8?q?=E7=94=A8=E8=BF=94=E5=9B=9E=E5=8F=82=E6=95=B0=E4=B8=8E=E5=AE=9A?= =?UTF-8?q?=E4=B9=89=E7=9A=84=E4=B8=8D=E7=AC=A6=20#635?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../genersoft/iot/vmp/common/StreamInfo.java | 36 ++++++++++++++++++- .../iot/vmp/service/impl/PlayServiceImpl.java | 2 +- .../vmanager/gb28181/play/PlayController.java | 2 +- .../gb28181/play/bean/PlayResult.java | 7 ++-- 4 files changed, 41 insertions(+), 6 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/common/StreamInfo.java b/src/main/java/com/genersoft/iot/vmp/common/StreamInfo.java index 928c305e..93d837c8 100644 --- a/src/main/java/com/genersoft/iot/vmp/common/StreamInfo.java +++ b/src/main/java/com/genersoft/iot/vmp/common/StreamInfo.java @@ -1,44 +1,78 @@ package com.genersoft.iot.vmp.common; +import io.swagger.v3.oas.annotations.media.Schema; +@Schema(description = "流信息") public class StreamInfo { + @Schema(description = "应用名") private String app; + @Schema(description = "流ID") private String stream; + @Schema(description = "设备编号") private String deviceID; + @Schema(description = "通道编号") private String channelId; + @Schema(description = "HTTP-FLV流地址") private String flv; + @Schema(description = "IP") private String ip; - + @Schema(description = "HTTPS-FLV流地址") private String https_flv; + @Schema(description = "Websocket-FLV流地址") private String ws_flv; + @Schema(description = "Websockets-FLV流地址") private String wss_flv; + @Schema(description = "HTTP-FMP4流地址") private String fmp4; + @Schema(description = "HTTPS-FMP4流地址") private String https_fmp4; + @Schema(description = "Websocket-FMP4流地址") private String ws_fmp4; + @Schema(description = "Websockets-FMP4流地址") private String wss_fmp4; + @Schema(description = "HLS流地址") private String hls; + @Schema(description = "HTTPS-HLS流地址") private String https_hls; + @Schema(description = "Websocket-HLS流地址") private String ws_hls; + @Schema(description = "Websockets-HLS流地址") private String wss_hls; + @Schema(description = "HTTP-TS流地址") private String ts; + @Schema(description = "HTTPS-TS流地址") private String https_ts; + @Schema(description = "Websocket-TS流地址") private String ws_ts; + @Schema(description = "Websockets-TS流地址") private String wss_ts; + @Schema(description = "RTMP流地址") private String rtmp; + @Schema(description = "RTMPS流地址") private String rtmps; + @Schema(description = "RTSP流地址") private String rtsp; + @Schema(description = "RTSPS流地址") private String rtsps; + @Schema(description = "RTC流地址") private String rtc; + @Schema(description = "RTCS流地址") private String rtcs; + @Schema(description = "流媒体ID") private String mediaServerId; + @Schema(description = "流编码信息") private Object tracks; + @Schema(description = "开始时间") private String startTime; + @Schema(description = "结束时间") private String endTime; + @Schema(description = "进度（录像下载使用）") private double progress; + @Schema(description = "是否暂停（录像回放使用）") private boolean pause; public static class TransactionInfo{ 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 667daceb..20a31fec 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 @@ -129,7 +129,7 @@ public class PlayServiceImpl implements IPlayService { String uuid = UUID.randomUUID().toString(); msg.setId(uuid); playResult.setUuid(uuid); - DeferredResult> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue()); + DeferredResult> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue()); playResult.setResult(result); // 录像查询以channelId作为deviceId查询 resultHolder.put(key, uuid, result); diff --git a/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java b/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java index b9652541..d22e56b9 100644 --- a/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java +++ b/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java @@ -82,7 +82,7 @@ public class PlayController { @Parameter(name = "deviceId", description = "设备国标编号", required = true) @Parameter(name = "channelId", description = "通道国标编号", required = true) @GetMapping("/start/{deviceId}/{channelId}") - public DeferredResult> play(@PathVariable String deviceId, + public DeferredResult> play(@PathVariable String deviceId, @PathVariable String channelId) { // 获取可用的zlm diff --git a/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/bean/PlayResult.java b/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/bean/PlayResult.java index bfbf3eb3..58d4ce20 100644 --- a/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/bean/PlayResult.java +++ b/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/bean/PlayResult.java @@ -1,5 +1,6 @@ package com.genersoft.iot.vmp.vmanager.gb28181.play.bean; +import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.gb28181.bean.Device; import com.genersoft.iot.vmp.vmanager.bean.WVPResult; import org.springframework.http.ResponseEntity; @@ -7,16 +8,16 @@ import org.springframework.web.context.request.async.DeferredResult; public class PlayResult { - private DeferredResult> result; + private DeferredResult> result; private String uuid; private Device device; - public DeferredResult> getResult() { + public DeferredResult> getResult() { return result; } - public void setResult(DeferredResult> result) { + public void setResult(DeferredResult> result) { this.result = result; } From 7918f037342e95913fd6af62f32702930742456c Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: Mon, 17 Oct 2022 16:38:28 +0800 Subject: [PATCH 4/4] =?UTF-8?q?=E4=BC=98=E5=8C=96=E6=8C=89=E9=9C=80?= =?UTF-8?q?=E6=8B=89=E6=B5=81=E9=85=8D=E7=BD=AE=EF=BC=8C=E6=8B=89=E6=B5=81?= =?UTF-8?q?=E4=BB=A3=E7=90=86=E6=94=AF=E6=8C=81=E6=8C=89=E9=9C=80=E6=8B=89?= =?UTF-8?q?=E6=B5=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- sql/mysql.sql | 1 - sql/update.sql | 5 + .../genersoft/iot/vmp/conf/MediaConfig.java | 8 -- .../genersoft/iot/vmp/conf/UserSetting.java | 10 ++ .../vmp/media/zlm/ZLMHttpHookListener.java | 93 +++++++++++-------- .../vmp/media/zlm/dto/MediaServerItem.java | 12 --- .../vmp/media/zlm/dto/StreamProxyItem.java | 2 +- .../service/impl/MediaServerServiceImpl.java | 2 - .../vmp/storager/dao/MediaServerMapper.java | 4 - .../vmp/storager/dao/StreamProxyMapper.java | 5 +- src/main/resources/all-application.yml | 4 +- .../src/components/dialog/MediaServerEdit.vue | 13 +-- .../src/components/dialog/StreamProxyEdit.vue | 11 ++- web_src/src/components/setting/Media.vue | 3 - 14 files changed, 91 insertions(+), 82 deletions(-) diff --git a/sql/mysql.sql b/sql/mysql.sql index 8c6c7b0b..a088c25a 100644 --- a/sql/mysql.sql +++ b/sql/mysql.sql @@ -277,7 +277,6 @@ CREATE TABLE `media_server` ( `rtspSSLPort` int NOT NULL, `autoConfig` int NOT NULL, `secret` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL, - `streamNoneReaderDelayMS` int NOT NULL, `rtpEnable` int NOT NULL, `rtpPortRange` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL, `sendRtpPortRange` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL, diff --git a/sql/update.sql b/sql/update.sql index e69de29b..e0d963b0 100644 --- a/sql/update.sql +++ b/sql/update.sql @@ -0,0 +1,5 @@ +alter table wvp.media_server + drop column streamNoneReaderDelayMS; + +alter table stream_proxy + add enable_disable_none_reader bit(1) default null; diff --git a/src/main/java/com/genersoft/iot/vmp/conf/MediaConfig.java b/src/main/java/com/genersoft/iot/vmp/conf/MediaConfig.java index 35ff08ee..708f72be 100644 --- a/src/main/java/com/genersoft/iot/vmp/conf/MediaConfig.java +++ b/src/main/java/com/genersoft/iot/vmp/conf/MediaConfig.java @@ -69,9 +69,6 @@ public class MediaConfig{ @Value("${media.secret}") private String secret; - @Value("${media.stream-none-reader-delay-ms:15000}") - private int streamNoneReaderDelayMS = 15000; - @Value("${media.rtp.enable}") private boolean rtpEnable; @@ -151,10 +148,6 @@ public class MediaConfig{ return secret; } - public int getStreamNoneReaderDelayMS() { - return streamNoneReaderDelayMS; - } - public boolean isRtpEnable() { return rtpEnable; } @@ -219,7 +212,6 @@ public class MediaConfig{ mediaServerItem.setRtspSSLPort(rtspSSLPort); mediaServerItem.setAutoConfig(autoConfig); mediaServerItem.setSecret(secret); - mediaServerItem.setStreamNoneReaderDelayMS(streamNoneReaderDelayMS); mediaServerItem.setRtpEnable(rtpEnable); mediaServerItem.setRtpPortRange(rtpPortRange); mediaServerItem.setSendRtpPortRange(sendRtpPortRange); 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 cad6e69d..ea2655e2 100644 --- a/src/main/java/com/genersoft/iot/vmp/conf/UserSetting.java +++ b/src/main/java/com/genersoft/iot/vmp/conf/UserSetting.java @@ -33,6 +33,8 @@ public class UserSetting { private Boolean usePushingAsStatus = Boolean.TRUE; + private Boolean streamOnDemand = Boolean.TRUE; + private String serverId = "000000"; private String thirdPartyGBIdReg = "[\\s\\S]*"; @@ -146,4 +148,12 @@ public class UserSetting { public void setUsePushingAsStatus(Boolean usePushingAsStatus) { this.usePushingAsStatus = usePushingAsStatus; } + + public Boolean getStreamOnDemand() { + return streamOnDemand; + } + + public void setStreamOnDemand(Boolean streamOnDemand) { + this.streamOnDemand = streamOnDemand; + } } diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java index e1689c94..bde4f827 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java @@ -558,9 +558,12 @@ public class ZLMHttpHookListener { String app = json.getString("app"); JSONObject ret = new JSONObject(); ret.put("code", 0); + // 录像下载 + ret.put("close", userSetting.getStreamOnDemand()); if ("rtp".equals(app)){ - ret.put("close", true); + // 国标流， 点播/录像回放/录像下载 StreamInfo streamInfoForPlayCatch = redisCatchStorage.queryPlayByStreamId(streamId); + // 点播 if (streamInfoForPlayCatch != null) { // 收到无人观看说明流也没有在往上级推送 if (redisCatchStorage.isChannelSendingRTP(streamInfoForPlayCatch.getChannelId())) { @@ -590,40 +593,39 @@ public class ZLMHttpHookListener { redisCatchStorage.stopPlay(streamInfoForPlayCatch); storager.stopPlay(streamInfoForPlayCatch.getDeviceID(), streamInfoForPlayCatch.getChannelId()); - }else{ - StreamInfo streamInfoForPlayBackCatch = redisCatchStorage.queryPlayback(null, null, streamId, null); - if (streamInfoForPlayBackCatch != null ) { - if (streamInfoForPlayBackCatch.isPause()) { - ret.put("close", false); - }else { - Device device = deviceService.queryDevice(streamInfoForPlayBackCatch.getDeviceID()); - if (device != null) { - try { - cmder.streamByeCmd(device,streamInfoForPlayBackCatch.getChannelId(), - streamInfoForPlayBackCatch.getStream(), null); - } catch (InvalidArgumentException | ParseException | SipException | - SsrcTransactionNotFoundException e) { - logger.error("[无人观看]回放， 发送BYE失败 {}", e.getMessage()); - } - } - redisCatchStorage.stopPlayback(streamInfoForPlayBackCatch.getDeviceID(), - streamInfoForPlayBackCatch.getChannelId(), streamInfoForPlayBackCatch.getStream(), null); - } - + return ret; + } + // 录像回放 + StreamInfo streamInfoForPlayBackCatch = redisCatchStorage.queryPlayback(null, null, streamId, null); + if (streamInfoForPlayBackCatch != null ) { + if (streamInfoForPlayBackCatch.isPause()) { + ret.put("close", false); }else { - StreamInfo streamInfoForDownload = redisCatchStorage.queryDownload(null, null, streamId, null); - // 进行录像下载时无人观看不断流 - if (streamInfoForDownload != null) { - ret.put("close", false); + Device device = deviceService.queryDevice(streamInfoForPlayBackCatch.getDeviceID()); + if (device != null) { + try { + cmder.streamByeCmd(device,streamInfoForPlayBackCatch.getChannelId(), + streamInfoForPlayBackCatch.getStream(), null); + } catch (InvalidArgumentException | ParseException | SipException | + SsrcTransactionNotFoundException e) { + logger.error("[无人观看]回放， 发送BYE失败 {}", e.getMessage()); + } } + redisCatchStorage.stopPlayback(streamInfoForPlayBackCatch.getDeviceID(), + streamInfoForPlayBackCatch.getChannelId(), streamInfoForPlayBackCatch.getStream(), null); } + return ret; } - MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId); - if (mediaServerItem != null && mediaServerItem.getStreamNoneReaderDelayMS() == -1) { + // 录像下载 + StreamInfo streamInfoForDownload = redisCatchStorage.queryDownload(null, null, streamId, null); + // 进行录像下载时无人观看不断流 + if (streamInfoForDownload != null) { ret.put("close", false); + return ret; } - return ret; }else { + // 非国标流 推流/拉流代理 + // 拉流代理 StreamProxyItem streamProxyItem = streamProxyService.getStreamProxyByAppAndStream(app, streamId); if (streamProxyItem != null ) { if (streamProxyItem.isEnable_remove_none_reader()) { @@ -635,12 +637,21 @@ public class ZLMHttpHookListener { }else if (streamProxyItem.isEnable_disable_none_reader()) { // 无人观看停用 ret.put("close", true); + // 修改数据 + streamProxyService.stop(app, streamId); }else { ret.put("close", false); } + return ret; } - return ret; + // 推流具有主动性，暂时不做处理 +// StreamPushItem streamPushItem = streamPushService.getPush(app, streamId); +// if (streamPushItem != null) { +// // TODO 发送停止 +// +// } } + return ret; } /** @@ -655,19 +666,27 @@ public class ZLMHttpHookListener { } String mediaServerId = json.getString("mediaServerId"); MediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId); - if (userSetting.isAutoApplyPlay() && mediaInfo != null && mediaInfo.isRtpEnable()) { + if (userSetting.isAutoApplyPlay() && mediaInfo != null) { String app = json.getString("app"); String streamId = json.getString("stream"); if ("rtp".equals(app)) { - String[] s = streamId.split("_"); - if (s.length == 2) { - String deviceId = s[0]; - String channelId = s[1]; - Device device = redisCatchStorage.getDevice(deviceId); - if (device != null) { - playService.play(mediaInfo,deviceId, channelId, null, null, null); + if (mediaInfo.isRtpEnable()) { + String[] s = streamId.split("_"); + if (s.length == 2) { + String deviceId = s[0]; + String channelId = s[1]; + Device device = redisCatchStorage.getDevice(deviceId); + if (device != null) { + playService.play(mediaInfo,deviceId, channelId, null, null, null); + } } } + }else { + // 拉流代理 + StreamProxyItem streamProxyByAppAndStream = streamProxyService.getStreamProxyByAppAndStream(app, streamId); + if (streamProxyByAppAndStream != null && streamProxyByAppAndStream.isEnable_disable_none_reader()) { + streamProxyService.start(app, streamId); + } } } diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/MediaServerItem.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/MediaServerItem.java index ec09ce5a..bf58187e 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/MediaServerItem.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/MediaServerItem.java @@ -54,9 +54,6 @@ public class MediaServerItem{ @Schema(description = "ZLM鉴权参数") private String secret; - @Schema(description = "某个流无人观看时，触发hook.on_stream_none_reader事件的最大等待时间，单位毫秒") - private int streamNoneReaderDelayMS; - @Schema(description = "keepalive hook触发间隔,单位秒") private int hookAliveInterval; @@ -119,7 +116,6 @@ public class MediaServerItem{ rtspSSLPort = zlmServerConfig.getRtspSSlport(); autoConfig = true; // 默认值true; secret = zlmServerConfig.getApiSecret(); - streamNoneReaderDelayMS = zlmServerConfig.getGeneralStreamNoneReaderDelayMS(); hookAliveInterval = zlmServerConfig.getHookAliveInterval(); rtpEnable = false; // 默认使用单端口;直到用户自己设置开启多端口 rtpPortRange = zlmServerConfig.getPortRange().replace("_",","); // 默认使用30000,30500作为级联时发送流的端口号 @@ -240,14 +236,6 @@ public class MediaServerItem{ this.secret = secret; } - public int getStreamNoneReaderDelayMS() { - return streamNoneReaderDelayMS; - } - - public void setStreamNoneReaderDelayMS(int streamNoneReaderDelayMS) { - this.streamNoneReaderDelayMS = streamNoneReaderDelayMS; - } - public boolean isRtpEnable() { return rtpEnable; } diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/StreamProxyItem.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/StreamProxyItem.java index f0d08a14..ea0bdcaa 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/StreamProxyItem.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/dto/StreamProxyItem.java @@ -38,7 +38,7 @@ public class StreamProxyItem extends GbStream { @Schema(description = "是否 无人观看时删除") private boolean enable_remove_none_reader; - @Schema(description = "是否 无人观看时不启用") + @Schema(description = "是否 无人观看时自动停用") private boolean enable_disable_none_reader; @Schema(description = "上级平台国标ID") private String platformGbId; diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java index afa2daad..7c5b1499 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java @@ -541,7 +541,6 @@ public class MediaServerServiceImpl implements IMediaServerService { param.put("hook.on_record_mp4",""); } param.put("hook.timeoutSec","20"); - param.put("general.streamNoneReaderDelayMS",mediaServerItem.getStreamNoneReaderDelayMS()==-1?"3600000":mediaServerItem.getStreamNoneReaderDelayMS() ); // 推流断开后可以在超时时间内重新连接上继续推流，这样播放器会接着播放。 // 置0关闭此特性(推流断开会导致立即断开播放器) // 此参数不应大于播放器超时时间 @@ -606,7 +605,6 @@ public class MediaServerServiceImpl implements IMediaServerService { mediaServerItem.setStreamIp(ip); mediaServerItem.setHookIp(sipConfig.getIp()); mediaServerItem.setSdpIp(ip); - mediaServerItem.setStreamNoneReaderDelayMS(zlmServerConfig.getGeneralStreamNoneReaderDelayMS()); return mediaServerItem; } diff --git a/src/main/java/com/genersoft/iot/vmp/storager/dao/MediaServerMapper.java b/src/main/java/com/genersoft/iot/vmp/storager/dao/MediaServerMapper.java index 0e37bbfa..95b3d150 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/dao/MediaServerMapper.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/dao/MediaServerMapper.java @@ -26,7 +26,6 @@ public interface MediaServerMapper { "rtspSSLPort, " + "autoConfig, " + "secret, " + - "streamNoneReaderDelayMS, " + "rtpEnable, " + "rtpPortRange, " + "sendRtpPortRange, " + @@ -51,7 +50,6 @@ public interface MediaServerMapper { "${rtspSSLPort}, " + "${autoConfig}, " + "'${secret}', " + - "${streamNoneReaderDelayMS}, " + "${rtpEnable}, " + "'${rtpPortRange}', " + "'${sendRtpPortRange}', " + @@ -77,7 +75,6 @@ public interface MediaServerMapper { ", rtspPort=${rtspPort}" + ", rtspSSLPort=${rtspSSLPort}" + ", autoConfig=${autoConfig}" + - ", streamNoneReaderDelayMS=${streamNoneReaderDelayMS}" + ", rtpEnable=${rtpEnable}" + ", rtpPortRange='${rtpPortRange}'" + ", sendRtpPortRange='${sendRtpPortRange}'" + @@ -102,7 +99,6 @@ public interface MediaServerMapper { ", rtspPort=${rtspPort}" + ", rtspSSLPort=${rtspSSLPort}" + ", autoConfig=${autoConfig}" + - ", streamNoneReaderDelayMS=${streamNoneReaderDelayMS}" + ", rtpEnable=${rtpEnable}" + ", rtpPortRange='${rtpPortRange}'" + ", sendRtpPortRange='${sendRtpPortRange}'" + diff --git a/src/main/java/com/genersoft/iot/vmp/storager/dao/StreamProxyMapper.java b/src/main/java/com/genersoft/iot/vmp/storager/dao/StreamProxyMapper.java index c05910e3..a9827ad3 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/dao/StreamProxyMapper.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/dao/StreamProxyMapper.java @@ -11,10 +11,10 @@ import java.util.List; public interface StreamProxyMapper { @Insert("INSERT INTO stream_proxy (type, name, app, stream,mediaServerId, url, src_url, dst_url, " + - "timeout_ms, ffmpeg_cmd_key, rtp_type, enable_hls, enable_mp4, enable, status, enable_remove_none_reader, createTime) VALUES" + + "timeout_ms, ffmpeg_cmd_key, rtp_type, enable_hls, enable_mp4, enable, status, enable_remove_none_reader, enable_disable_none_reader, createTime) VALUES" + "('${type}','${name}', '${app}', '${stream}', '${mediaServerId}','${url}', '${src_url}', '${dst_url}', " + "'${timeout_ms}', '${ffmpeg_cmd_key}', '${rtp_type}', ${enable_hls}, ${enable_mp4}, ${enable}, ${status}, " + - "${enable_remove_none_reader}, '${createTime}' )") + "${enable_remove_none_reader}, ${enable_disable_none_reader}, '${createTime}' )") int add(StreamProxyItem streamProxyDto); @Update("UPDATE stream_proxy " + @@ -33,6 +33,7 @@ public interface StreamProxyMapper { "enable=#{enable}, " + "status=#{status}, " + "enable_remove_none_reader=#{enable_remove_none_reader}, " + + "enable_disable_none_reader=#{enable_disable_none_reader}, " + "enable_mp4=#{enable_mp4} " + "WHERE app=#{app} AND stream=#{stream}") int update(StreamProxyItem streamProxyDto); diff --git a/src/main/resources/all-application.yml b/src/main/resources/all-application.yml index ed025b9a..49d5aca1 100644 --- a/src/main/resources/all-application.yml +++ b/src/main/resources/all-application.yml @@ -146,8 +146,6 @@ media: auto-config: true # [可选] zlm服务器的hook.admin_params=secret secret: 035c73f7-bb6b-4889-a715-d9eb2d1925cc - # [可选] zlm服务器的general.streamNoneReaderDelayMS - stream-none-reader-delay-ms: 18000 # 无人观看多久自动关闭流, -1表示永不自动关闭,即 关闭按需拉流 # 启用多端口模式, 多端口模式使用端口区分每路流，兼容性更好。 单端口使用流的ssrc区分， 点播超时建议使用多端口测试 rtp: # [可选] 是否启用多端口模式, 开启后会在portRange范围内选择端口用于媒体流传输 @@ -190,6 +188,8 @@ user-settings: logInDatebase: true # 使用推流状态作为推流通道状态 use-pushing-as-status: true + # 按需拉流, true：有人观看拉流，无人观看释放， false：拉起后不自动释放 + stream-on-demand: true # 关闭在线文档（生产环境建议关闭） springdoc: diff --git a/web_src/src/components/dialog/MediaServerEdit.vue b/web_src/src/components/dialog/MediaServerEdit.vue index 1754461d..24f8c85f 100644 --- a/web_src/src/components/dialog/MediaServerEdit.vue +++ b/web_src/src/components/dialog/MediaServerEdit.vue @@ -41,10 +41,6 @@ - - - - @@ -74,6 +70,10 @@ + + + + @@ -94,9 +94,6 @@ - - - - @@ -172,7 +169,6 @@ export default { hookIp: "", sdpIp: "", streamIp: "", - streamNoneReaderDelayMS: "", secret: "035c73f7-bb6b-4889-a715-d9eb2d1925cc", httpPort: "", httpSSlPort: "", @@ -332,7 +328,6 @@ export default { hookIp: "", sdpIp: "", streamIp: "", - streamNoneReaderDelayMS: "", secret: "035c73f7-bb6b-4889-a715-d9eb2d1925cc", httpPort: "", httpSSlPort: "", diff --git a/web_src/src/components/dialog/StreamProxyEdit.vue b/web_src/src/components/dialog/StreamProxyEdit.vue index 936bc53d..1270999f 100644 --- a/web_src/src/components/dialog/StreamProxyEdit.vue +++ b/web_src/src/components/dialog/StreamProxyEdit.vue @@ -105,7 +105,9 @@ - + + + @@ -170,6 +172,7 @@ export default { enable_hls: true, enable_mp4: false, enable_remove_none_reader: false, + enable_disable_none_reader: true, platformGbId: null, mediaServerId: null, }, @@ -276,6 +279,12 @@ export default { if (this.platform.enable && this.platform.expires == "0") { this.platform.expires = "300"; } + }, + removeNoneReader: function(checked) { + this.proxyParam.enable_disable_none_reader = !checked; + }, + disableNoneReaderHandType: function(checked) { + this.proxyParam.enable_remove_none_reader = !checked; } }, }; diff --git a/web_src/src/components/setting/Media.vue b/web_src/src/components/setting/Media.vue index 19465b96..66426010 100644 --- a/web_src/src/components/setting/Media.vue +++ b/web_src/src/components/setting/Media.vue @@ -42,9 +42,6 @@ - - -