From 874a57387adb506905693b3de6cb70d8f0719530 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: Tue, 8 Aug 2023 09:49:27 +0800 Subject: [PATCH 01/26] =?UTF-8?q?=E6=94=AF=E6=8C=81=E4=B8=8E=E7=AC=AC?= =?UTF-8?q?=E4=B8=89=E6=96=B9=E5=AF=B9=E6=8E=A5PS=E6=B5=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../iot/vmp/common/VideoManagerConstants.java | 2 + .../vmp/media/zlm/ZLMHttpHookListener.java | 5 +- .../vmp/vmanager/bean/OtherPsSendInfo.java | 137 ++++++++ .../iot/vmp/vmanager/ps/PsController.java | 303 ++++++++++++++++++ 4 files changed, 446 insertions(+), 1 deletion(-) create mode 100644 src/main/java/com/genersoft/iot/vmp/vmanager/bean/OtherPsSendInfo.java create mode 100644 src/main/java/com/genersoft/iot/vmp/vmanager/ps/PsController.java diff --git a/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java b/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java index ec25ea2e5..a6132ca0f 100644 --- a/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java +++ b/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java @@ -154,7 +154,9 @@ public class VideoManagerConstants { public static final String WVP_STREAM_GB_ID_PREFIX = "memberNo_"; public static final String WVP_STREAM_GPS_MSG_PREFIX = "WVP_STREAM_GPS_MSG_"; public static final String WVP_OTHER_SEND_RTP_INFO = "VMP_OTHER_SEND_RTP_INFO_"; + public static final String WVP_OTHER_SEND_PS_INFO = "VMP_OTHER_SEND_PS_INFO_"; public static final String WVP_OTHER_RECEIVE_RTP_INFO = "VMP_OTHER_RECEIVE_RTP_INFO_"; + public static final String WVP_OTHER_RECEIVE_PS_INFO = "VMP_OTHER_RECEIVE_PS_INFO_"; /** * Redis Const 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 52b3664e7..acc9548e0 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 @@ -275,7 +275,10 @@ public class ZLMHttpHookListener { if (param.getApp().equalsIgnoreCase("rtp")) { String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_RTP_INFO + userSetting.getServerId() + "_" + param.getStream(); OtherRtpSendInfo otherRtpSendInfo = (OtherRtpSendInfo)redisTemplate.opsForValue().get(receiveKey); - if (otherRtpSendInfo != null) { + + String receiveKeyForPS = VideoManagerConstants.WVP_OTHER_RECEIVE_PS_INFO + userSetting.getServerId() + "_" + param.getStream(); + OtherPsSendInfo otherPsSendInfo = (OtherPsSendInfo)redisTemplate.opsForValue().get(receiveKeyForPS); + if (otherRtpSendInfo != null || otherPsSendInfo != null) { result.setEnable_mp4(true); } } diff --git a/src/main/java/com/genersoft/iot/vmp/vmanager/bean/OtherPsSendInfo.java b/src/main/java/com/genersoft/iot/vmp/vmanager/bean/OtherPsSendInfo.java new file mode 100644 index 000000000..ac9840921 --- /dev/null +++ b/src/main/java/com/genersoft/iot/vmp/vmanager/bean/OtherPsSendInfo.java @@ -0,0 +1,137 @@ +package com.genersoft.iot.vmp.vmanager.bean; + +public class OtherPsSendInfo { + + /** + * 发流IP + */ + private String sendLocalIp; + + /** + * 发流端口 + */ + private int sendLocalPort; + + /** + * 收流IP + */ + private String receiveIp; + + /** + * 收流端口 + */ + private int receivePort; + + + /** + * 会话ID + */ + private String callId; + + /** + * 流ID + */ + private String stream; + + /** + * 推流应用名 + */ + private String pushApp; + + /** + * 推流流ID + */ + private String pushStream; + + /** + * 推流SSRC + */ + private String pushSSRC; + + public String getSendLocalIp() { + return sendLocalIp; + } + + public void setSendLocalIp(String sendLocalIp) { + this.sendLocalIp = sendLocalIp; + } + + public int getSendLocalPort() { + return sendLocalPort; + } + + public void setSendLocalPort(int sendLocalPort) { + this.sendLocalPort = sendLocalPort; + } + + public String getReceiveIp() { + return receiveIp; + } + + public void setReceiveIp(String receiveIp) { + this.receiveIp = receiveIp; + } + + public int getReceivePort() { + return receivePort; + } + + public void setReceivePort(int receivePort) { + this.receivePort = receivePort; + } + + public String getCallId() { + return callId; + } + + public void setCallId(String callId) { + this.callId = callId; + } + + public String getStream() { + return stream; + } + + public void setStream(String stream) { + this.stream = stream; + } + + public String getPushApp() { + return pushApp; + } + + public void setPushApp(String pushApp) { + this.pushApp = pushApp; + } + + public String getPushStream() { + return pushStream; + } + + public void setPushStream(String pushStream) { + this.pushStream = pushStream; + } + + public String getPushSSRC() { + return pushSSRC; + } + + public void setPushSSRC(String pushSSRC) { + this.pushSSRC = pushSSRC; + } + + @Override + public String toString() { + return "OtherPsSendInfo{" + + "sendLocalIp='" + sendLocalIp + '\'' + + ", sendLocalPort=" + sendLocalPort + + ", receiveIp='" + receiveIp + '\'' + + ", receivePort=" + receivePort + + ", callId='" + callId + '\'' + + ", stream='" + stream + '\'' + + ", pushApp='" + pushApp + '\'' + + ", pushStream='" + pushStream + '\'' + + ", pushSSRC='" + pushSSRC + '\'' + + '}'; + } +} diff --git a/src/main/java/com/genersoft/iot/vmp/vmanager/ps/PsController.java b/src/main/java/com/genersoft/iot/vmp/vmanager/ps/PsController.java new file mode 100644 index 000000000..a2fd81b29 --- /dev/null +++ b/src/main/java/com/genersoft/iot/vmp/vmanager/ps/PsController.java @@ -0,0 +1,303 @@ +package com.genersoft.iot.vmp.vmanager.ps; + +import com.alibaba.fastjson2.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.conf.exception.ControllerException; +import com.genersoft.iot.vmp.media.zlm.SendRtpPortManager; +import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory; +import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe; +import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory; +import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForRtpServerTimeout; +import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange; +import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; +import com.genersoft.iot.vmp.service.IMediaServerService; +import com.genersoft.iot.vmp.utils.redis.RedisUtil; +import com.genersoft.iot.vmp.vmanager.bean.ErrorCode; +import com.genersoft.iot.vmp.vmanager.bean.OtherPsSendInfo; +import com.genersoft.iot.vmp.vmanager.bean.OtherRtpSendInfo; +import io.swagger.v3.oas.annotations.Operation; +import io.swagger.v3.oas.annotations.Parameter; +import io.swagger.v3.oas.annotations.tags.Tag; +import okhttp3.OkHttpClient; +import okhttp3.Request; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.web.bind.annotation.*; + +import java.io.IOException; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.TimeUnit; + +@SuppressWarnings("rawtypes") +@Tag(name = "第三方PS服务对接") + +@RestController +@RequestMapping("/api/ps") +public class PsController { + + private final static Logger logger = LoggerFactory.getLogger(PsController.class); + + @Autowired + private ZLMRTPServerFactory zlmServerFactory; + + @Autowired + private ZlmHttpHookSubscribe hookSubscribe; + + @Autowired + private IMediaServerService mediaServerService; + + @Autowired + private SendRtpPortManager sendRtpPortManager; + + @Autowired + private UserSetting userSetting; + + @Autowired + private DynamicTask dynamicTask; + + + @Autowired + private RedisTemplate redisTemplate; + + + @GetMapping(value = "/receive/open") + @ResponseBody + @Operation(summary = "开启收流和获取发流信息") + @Parameter(name = "isSend", description = "是否发送，false时只开启收流， true同时返回推流信息", required = true) + @Parameter(name = "callId", description = "整个过程的唯一标识，为了与后续接口关联", required = true) + @Parameter(name = "ssrc", description = "来源流的SSRC，不传则不校验来源ssrc", required = false) + @Parameter(name = "stream", description = "形成的流的ID", required = true) + @Parameter(name = "tcpMode", description = "收流模式， 0为UDP， 1为TCP被动", required = true) + @Parameter(name = "callBack", description = "回调地址，如果收流超时会通道回调通知，回调为get请求，参数为callId", required = true) + public OtherPsSendInfo openRtpServer(Boolean isSend, @RequestParam(required = false)String ssrc, String callId, String stream, Integer tcpMode, String callBack) { + + logger.info("[第三方PS服务对接->开启收流和获取发流信息] isSend->{}, ssrc->{}, callId->{}, stream->{}, tcpMode->{}, callBack->{}", + isSend, ssrc, callId, stream, tcpMode==0?"UDP":"TCP被动", callBack); + + MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer(); + if (mediaServerItem == null) { + throw new ControllerException(ErrorCode.ERROR100.getCode(),"没有可用的MediaServer"); + } + if (stream == null) { + throw new ControllerException(ErrorCode.ERROR100.getCode(),"stream参数不可为空"); + } + if (isSend != null && isSend && callId == null) { + throw new ControllerException(ErrorCode.ERROR100.getCode(),"isSend为true时，CallID不能为空"); + } + int ssrcInt = 0; + if (ssrc != null) { + try { + ssrcInt = Integer.parseInt(ssrc); + }catch (NumberFormatException e) { + throw new ControllerException(ErrorCode.ERROR100.getCode(),"ssrc格式错误"); + } + } + String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_PS_INFO + userSetting.getServerId() + "_" + callId + "_" + stream; + int localPort = zlmServerFactory.createRTPServer(mediaServerItem, stream, ssrcInt, null, false, tcpMode); + if (localPort == 0) { + throw new ControllerException(ErrorCode.ERROR100.getCode(), "获取端口失败"); + } + // 注册回调如果rtp收流超时则通过回调发送通知 + if (callBack != null) { + HookSubscribeForRtpServerTimeout hookSubscribeForRtpServerTimeout = HookSubscribeFactory.on_rtp_server_timeout(stream, String.valueOf(ssrcInt), mediaServerItem.getId()); + // 订阅 zlm启动事件, 新的zlm也会从这里进入系统 + hookSubscribe.addSubscribe(hookSubscribeForRtpServerTimeout, + (mediaServerItemInUse, response)->{ + if (stream.equals(response.getString("stream_id"))) { + logger.info("[第三方PS服务对接->开启收流和获取发流信息] 等待收流超时 callId->{}, 发送回调", callId); + // 将信息写入redis中，以备后用 + redisTemplate.delete(receiveKey); + OkHttpClient.Builder httpClientBuilder = new OkHttpClient.Builder(); + OkHttpClient client = httpClientBuilder.build(); + String url = callBack + "?callId=" + callId; + Request request = new Request.Builder().get().url(url).build(); + try { + client.newCall(request).execute(); + } catch (IOException e) { + logger.error("[第三方PS服务对接->开启收流和获取发流信息] 等待收流超时 callId->{}, 发送回调失败", callId, e); + } + hookSubscribe.removeSubscribe(hookSubscribeForRtpServerTimeout); + } + }); + } + OtherPsSendInfo otherPsSendInfo = new OtherPsSendInfo(); + otherPsSendInfo.setReceiveIp(mediaServerItem.getSdpIp()); + otherPsSendInfo.setReceivePort(localPort); + otherPsSendInfo.setCallId(callId); + otherPsSendInfo.setStream(stream); + + // 将信息写入redis中，以备后用 + redisTemplate.opsForValue().set(receiveKey, otherPsSendInfo); + if (isSend != null && isSend) { + String key = VideoManagerConstants.WVP_OTHER_SEND_PS_INFO + userSetting.getServerId() + "_" + callId; + // 预创建发流信息 + int port = sendRtpPortManager.getNextPort(mediaServerItem); + + otherPsSendInfo.setSendLocalIp(mediaServerItem.getSdpIp()); + otherPsSendInfo.setSendLocalPort(port); + // 将信息写入redis中，以备后用 + redisTemplate.opsForValue().set(key, otherPsSendInfo, 300, TimeUnit.SECONDS); + logger.info("[第三方PS服务对接->开启收流和获取发流信息] 结果，callId->{}， {}", callId, otherPsSendInfo); + } + return otherPsSendInfo; + } + + @GetMapping(value = "/receive/close") + @ResponseBody + @Operation(summary = "关闭收流") + @Parameter(name = "stream", description = "流的ID", required = true) + public void closeRtpServer(String stream) { + logger.info("[第三方PS服务对接->关闭收流] stream->{}", stream); + MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer(); + zlmServerFactory.closeRtpServer(mediaServerItem,stream); + String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_PS_INFO + userSetting.getServerId() + "_*_" + stream; + List scan = RedisUtil.scan(redisTemplate, receiveKey); + if (scan.size() > 0) { + for (Object key : scan) { + // 将信息写入redis中，以备后用 + redisTemplate.delete(key); + } + } + } + + @GetMapping(value = "/send/start") + @ResponseBody + @Operation(summary = "发送流") + @Parameter(name = "ssrc", description = "发送流的SSRC", required = true) + @Parameter(name = "dstIp", description = "目标收流IP", required = true) + @Parameter(name = "dstPort", description = "目标收流端口", required = true) + @Parameter(name = "app", description = "待发送应用名", required = true) + @Parameter(name = "stream", description = "待发送流Id", required = true) + @Parameter(name = "callId", description = "整个过程的唯一标识，不传则使用随机端口发流", required = true) + @Parameter(name = "isUdp", description = "是否为UDP", required = true) + public void sendRTP(String ssrc, + String dstIp, + Integer dstPort, + String app, + String stream, + String callId, + Boolean isUdp + ) { + logger.info("[第三方PS服务对接->发送流] " + + "ssrc->{}, \r\n" + + "dstIp->{}, \n" + + "dstPort->{}, \n" + + "app->{}, \n" + + "stream->{}, \n" + + "callId->{} \n", + ssrc, + dstIp, + dstPort, + app, + stream, + callId); + MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer(); + String key = VideoManagerConstants.WVP_OTHER_SEND_PS_INFO + userSetting.getServerId() + "_" + callId; + OtherRtpSendInfo sendInfo = (OtherRtpSendInfo)redisTemplate.opsForValue().get(key); + if (sendInfo == null) { + sendInfo = new OtherRtpSendInfo(); + } + sendInfo.setPushApp(app); + sendInfo.setPushStream(stream); + sendInfo.setPushSSRC(ssrc); + + Map param; + + + param = new HashMap<>(); + param.put("vhost","__defaultVhost__"); + param.put("app",app); + param.put("stream",stream); + param.put("ssrc", ssrc); + + param.put("dst_url", dstIp); + param.put("dst_port", dstPort); + String is_Udp = isUdp ? "1" : "0"; + param.put("is_udp", is_Udp); + param.put("src_port", sendInfo.getSendLocalPortForAudio()); + param.put("use_ps", "0"); + param.put("only_audio", "1"); + + + Boolean streamReady = zlmServerFactory.isStreamReady(mediaServerItem, app, stream); + if (streamReady) { + JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServerItem, param); + if (jsonObject.getInteger("code") == 0) { + logger.info("[第三方PS服务对接->发送流] 视频流发流成功，callId->{}，param->{}", callId, param); + redisTemplate.opsForValue().set(key, sendInfo); + }else { + redisTemplate.delete(key); + logger.info("[第三方PS服务对接->发送流] 视频流发流失败，callId->{}, {}", callId, jsonObject.getString("msg")); + throw new ControllerException(ErrorCode.ERROR100.getCode(), "[视频流发流失败] " + jsonObject.getString("msg")); + } + }else { + logger.info("[第三方PS服务对接->发送流] 流不存在，等待流上线，callId->{}", callId); + String uuid = UUID.randomUUID().toString(); + HookSubscribeForStreamChange hookSubscribeForStreamChange = HookSubscribeFactory.on_stream_changed(app, stream, true, "rtsp", mediaServerItem.getId()); + dynamicTask.startDelay(uuid, ()->{ + logger.info("[第三方PS服务对接->发送流] 等待流上线超时 callId->{}", callId); + redisTemplate.delete(key); + hookSubscribe.removeSubscribe(hookSubscribeForStreamChange); + }, 10000); + + // 订阅 zlm启动事件, 新的zlm也会从这里进入系统 + OtherRtpSendInfo finalSendInfo = sendInfo; + hookSubscribe.removeSubscribe(hookSubscribeForStreamChange); + hookSubscribe.addSubscribe(hookSubscribeForStreamChange, + (mediaServerItemInUse, response)->{ + dynamicTask.stop(uuid); + logger.info("[第三方PS服务对接->发送流] 流上线，开始发流 callId->{}", callId); + try { + Thread.sleep(400); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaServerItem, param); + if (jsonObject.getInteger("code") == 0) { + logger.info("[第三方PS服务对接->发送流] 视频流发流成功，callId->{}，param->{}", callId, param); + redisTemplate.opsForValue().set(key, finalSendInfo); + }else { + redisTemplate.delete(key); + logger.info("[第三方PS服务对接->发送流] 视频流发流失败，callId->{}, {}", callId, jsonObject.getString("msg")); + throw new ControllerException(ErrorCode.ERROR100.getCode(), "[视频流发流失败] " + jsonObject.getString("msg")); + } + hookSubscribe.removeSubscribe(hookSubscribeForStreamChange); + }); + } + } + + @GetMapping(value = "/send/stop") + @ResponseBody + @Operation(summary = "关闭发送流") + @Parameter(name = "callId", description = "整个过程的唯一标识，不传则使用随机端口发流", required = true) + public void closeSendRTP(String callId) { + logger.info("[第三方PS服务对接->关闭发送流] callId->{}", callId); + String key = VideoManagerConstants.WVP_OTHER_SEND_PS_INFO + userSetting.getServerId() + "_" + callId; + OtherRtpSendInfo sendInfo = (OtherRtpSendInfo)redisTemplate.opsForValue().get(key); + if (sendInfo == null){ + throw new ControllerException(ErrorCode.ERROR100.getCode(), "未开启发流"); + } + Map param = new HashMap<>(); + param.put("vhost","__defaultVhost__"); + param.put("app",sendInfo.getPushApp()); + param.put("stream",sendInfo.getPushStream()); + param.put("ssrc",sendInfo.getPushSSRC()); + MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer(); + Boolean result = zlmServerFactory.stopSendRtpStream(mediaServerItem, param); + if (!result) { + logger.info("[第三方PS服务对接->关闭发送流] 失败 callId->{}", callId); + throw new ControllerException(ErrorCode.ERROR100.getCode(), "停止发流失败"); + }else { + logger.info("[第三方PS服务对接->关闭发送流] 成功 callId->{}", callId); + } + redisTemplate.delete(key); + } + +} From 5e34039deffa57fc50e93e45b6fa6879ce70e552 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: Tue, 8 Aug 2023 10:01:51 +0800 Subject: [PATCH 02/26] =?UTF-8?q?=E6=94=AF=E6=8C=81=E5=AF=B9=E6=8E=A5?= =?UTF-8?q?=E7=AC=AC=E4=B8=89=E6=96=B9PS=E6=B5=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../iot/vmp/media/zlm/ZLMHttpHookListener.java | 1 + .../genersoft/iot/vmp/vmanager/ps/PsController.java | 12 +++++++----- 2 files changed, 8 insertions(+), 5 deletions(-) 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 0d6e69ee3..3c8f2de21 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 @@ -29,6 +29,7 @@ import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage; import com.genersoft.iot.vmp.utils.DateUtil; import com.genersoft.iot.vmp.vmanager.bean.ErrorCode; +import com.genersoft.iot.vmp.vmanager.bean.OtherPsSendInfo; import com.genersoft.iot.vmp.vmanager.bean.OtherRtpSendInfo; import com.genersoft.iot.vmp.vmanager.bean.StreamContent; import org.slf4j.Logger; diff --git a/src/main/java/com/genersoft/iot/vmp/vmanager/ps/PsController.java b/src/main/java/com/genersoft/iot/vmp/vmanager/ps/PsController.java index a2fd81b29..3ea265f9c 100644 --- a/src/main/java/com/genersoft/iot/vmp/vmanager/ps/PsController.java +++ b/src/main/java/com/genersoft/iot/vmp/vmanager/ps/PsController.java @@ -6,12 +6,13 @@ import com.genersoft.iot.vmp.conf.DynamicTask; import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.exception.ControllerException; import com.genersoft.iot.vmp.media.zlm.SendRtpPortManager; -import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory; +import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory; import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe; import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory; import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForRtpServerTimeout; import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; +import com.genersoft.iot.vmp.media.zlm.dto.hook.OnRtpServerTimeoutHookParam; import com.genersoft.iot.vmp.service.IMediaServerService; import com.genersoft.iot.vmp.utils.redis.RedisUtil; import com.genersoft.iot.vmp.vmanager.bean.ErrorCode; @@ -45,7 +46,7 @@ public class PsController { private final static Logger logger = LoggerFactory.getLogger(PsController.class); @Autowired - private ZLMRTPServerFactory zlmServerFactory; + private ZLMServerFactory zlmServerFactory; @Autowired private ZlmHttpHookSubscribe hookSubscribe; @@ -109,8 +110,9 @@ public class PsController { HookSubscribeForRtpServerTimeout hookSubscribeForRtpServerTimeout = HookSubscribeFactory.on_rtp_server_timeout(stream, String.valueOf(ssrcInt), mediaServerItem.getId()); // 订阅 zlm启动事件, 新的zlm也会从这里进入系统 hookSubscribe.addSubscribe(hookSubscribeForRtpServerTimeout, - (mediaServerItemInUse, response)->{ - if (stream.equals(response.getString("stream_id"))) { + (mediaServerItemInUse, hookParam)->{ + OnRtpServerTimeoutHookParam serverTimeoutHookParam = (OnRtpServerTimeoutHookParam) hookParam; + if (stream.equals(serverTimeoutHookParam.getStream_id())) { logger.info("[第三方PS服务对接->开启收流和获取发流信息] 等待收流超时 callId->{}, 发送回调", callId); // 将信息写入redis中，以备后用 redisTemplate.delete(receiveKey); @@ -159,7 +161,7 @@ public class PsController { zlmServerFactory.closeRtpServer(mediaServerItem,stream); String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_PS_INFO + userSetting.getServerId() + "_*_" + stream; List scan = RedisUtil.scan(redisTemplate, receiveKey); - if (scan.size() > 0) { + if (!scan.isEmpty()) { for (Object key : scan) { // 将信息写入redis中，以备后用 redisTemplate.delete(key); From d67f0a1ea403fbebfff70e7eaf4fe4870be2a1b8 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: Tue, 8 Aug 2023 16:06:02 +0800 Subject: [PATCH 03/26] =?UTF-8?q?=E8=A7=A3=E5=86=B3=E4=BD=BF=E7=94=A8redis?= =?UTF-8?q?=E9=9B=86=E7=BE=A4=E6=97=B6=E8=8E=B7=E5=8F=96=E5=8F=91=E9=80=81?= =?UTF-8?q?=E7=AB=AF=E5=8F=A3=E5=A4=B1=E8=B4=A5=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../iot/vmp/media/zlm/SendRtpPortManager.java | 42 +++++++++++++++---- .../iot/vmp/vmanager/ps/PsController.java | 31 +++++++++++--- 2 files changed, 58 insertions(+), 15 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/SendRtpPortManager.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/SendRtpPortManager.java index f960c7dc8..3f28d02ad 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/SendRtpPortManager.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/SendRtpPortManager.java @@ -30,7 +30,7 @@ public class SendRtpPortManager { private final String KEY = "VM_MEDIA_SEND_RTP_PORT_"; - public int getNextPort(MediaServerItem mediaServer) { + public synchronized int getNextPort(MediaServerItem mediaServer) { if (mediaServer == null) { logger.warn("[发送端口管理] 参数错误，mediaServer为NULL"); return -1; @@ -50,17 +50,15 @@ public class SendRtpPortManager { String sendRtpPortRange = mediaServer.getSendRtpPortRange(); int startPort; int endPort; - if (sendRtpPortRange == null) { - logger.warn("{}未设置发送端口默认值，自动使用40000-50000作为端口范围", mediaServer.getId()); + if (sendRtpPortRange != null) { String[] portArray = sendRtpPortRange.split(","); if (portArray.length != 2 || !NumberUtils.isParsable(portArray[0]) || !NumberUtils.isParsable(portArray[1])) { - logger.warn("{}发送端口配置格式错误，自动使用40000-50000作为端口范围", mediaServer.getId()); + logger.warn("{}发送端口配置格式错误，自动使用50000-60000作为端口范围", mediaServer.getId()); startPort = 50000; endPort = 60000; }else { - if ( Integer.parseInt(portArray[1]) - Integer.parseInt(portArray[0]) < 1) { - logger.warn("{}发送端口配置错误,结束端口至少比开始端口大一，自动使用40000-50000作为端口范围", mediaServer.getId()); + logger.warn("{}发送端口配置错误,结束端口至少比开始端口大一，自动使用50000-60000作为端口范围", mediaServer.getId()); startPort = 50000; endPort = 60000; }else { @@ -69,6 +67,7 @@ public class SendRtpPortManager { } } }else { + logger.warn("{}未设置发送端口默认值，自动使用50000-60000作为端口范围", mediaServer.getId()); startPort = 50000; endPort = 60000; } @@ -76,10 +75,35 @@ public class SendRtpPortManager { logger.warn("{}获取redis连接信息失败", mediaServer.getId()); return -1; } +// RedisAtomicInteger redisAtomicInteger = new RedisAtomicInteger(sendIndexKey , redisTemplate.getConnectionFactory()); +// return redisAtomicInteger.getAndUpdate((current)->{ +// return getPort(current, startPort, endPort, checkPort-> !sendRtpItemMap.containsKey(checkPort)); +// }); + return getSendPort(startPort, endPort, sendIndexKey, sendRtpItemMap); + } + + private synchronized int getSendPort(int startPort, int endPort, String sendIndexKey, Map sendRtpItemMap){ RedisAtomicInteger redisAtomicInteger = new RedisAtomicInteger(sendIndexKey , redisTemplate.getConnectionFactory()); - return redisAtomicInteger.getAndUpdate((current)->{ - return getPort(current, startPort, endPort, checkPort-> !sendRtpItemMap.containsKey(checkPort)); - }); + if (redisAtomicInteger.get() < startPort) { + redisAtomicInteger.set(startPort); + return startPort; + }else { + int port = redisAtomicInteger.getAndIncrement(); + if (port > endPort) { + redisAtomicInteger.set(startPort); + if (sendRtpItemMap.containsKey(startPort)) { + return getSendPort(startPort, endPort, sendIndexKey, sendRtpItemMap); + }else { + return startPort; + } + } + if (sendRtpItemMap.containsKey(port)) { + return getSendPort(startPort, endPort, sendIndexKey, sendRtpItemMap); + }else { + return port; + } + } + } interface CheckPortCallback{ diff --git a/src/main/java/com/genersoft/iot/vmp/vmanager/ps/PsController.java b/src/main/java/com/genersoft/iot/vmp/vmanager/ps/PsController.java index a2fd81b29..045480ac5 100644 --- a/src/main/java/com/genersoft/iot/vmp/vmanager/ps/PsController.java +++ b/src/main/java/com/genersoft/iot/vmp/vmanager/ps/PsController.java @@ -16,7 +16,6 @@ import com.genersoft.iot.vmp.service.IMediaServerService; import com.genersoft.iot.vmp.utils.redis.RedisUtil; import com.genersoft.iot.vmp.vmanager.bean.ErrorCode; import com.genersoft.iot.vmp.vmanager.bean.OtherPsSendInfo; -import com.genersoft.iot.vmp.vmanager.bean.OtherRtpSendInfo; import io.swagger.v3.oas.annotations.Operation; import io.swagger.v3.oas.annotations.Parameter; import io.swagger.v3.oas.annotations.tags.Tag; @@ -200,9 +199,9 @@ public class PsController { callId); MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer(); String key = VideoManagerConstants.WVP_OTHER_SEND_PS_INFO + userSetting.getServerId() + "_" + callId; - OtherRtpSendInfo sendInfo = (OtherRtpSendInfo)redisTemplate.opsForValue().get(key); + OtherPsSendInfo sendInfo = (OtherPsSendInfo)redisTemplate.opsForValue().get(key); if (sendInfo == null) { - sendInfo = new OtherRtpSendInfo(); + sendInfo = new OtherPsSendInfo(); } sendInfo.setPushApp(app); sendInfo.setPushStream(stream); @@ -221,7 +220,7 @@ public class PsController { param.put("dst_port", dstPort); String is_Udp = isUdp ? "1" : "0"; param.put("is_udp", is_Udp); - param.put("src_port", sendInfo.getSendLocalPortForAudio()); + param.put("src_port", sendInfo.getSendLocalPort()); param.put("use_ps", "0"); param.put("only_audio", "1"); @@ -248,7 +247,7 @@ public class PsController { }, 10000); // 订阅 zlm启动事件, 新的zlm也会从这里进入系统 - OtherRtpSendInfo finalSendInfo = sendInfo; + OtherPsSendInfo finalSendInfo = sendInfo; hookSubscribe.removeSubscribe(hookSubscribeForStreamChange); hookSubscribe.addSubscribe(hookSubscribeForStreamChange, (mediaServerItemInUse, response)->{ @@ -280,7 +279,7 @@ public class PsController { public void closeSendRTP(String callId) { logger.info("[第三方PS服务对接->关闭发送流] callId->{}", callId); String key = VideoManagerConstants.WVP_OTHER_SEND_PS_INFO + userSetting.getServerId() + "_" + callId; - OtherRtpSendInfo sendInfo = (OtherRtpSendInfo)redisTemplate.opsForValue().get(key); + OtherPsSendInfo sendInfo = (OtherPsSendInfo)redisTemplate.opsForValue().get(key); if (sendInfo == null){ throw new ControllerException(ErrorCode.ERROR100.getCode(), "未开启发流"); } @@ -300,4 +299,24 @@ public class PsController { redisTemplate.delete(key); } + + @GetMapping(value = "/getTestPort") + @ResponseBody + public int getTestPort() { + MediaServerItem defaultMediaServer = mediaServerService.getDefaultMediaServer(); + +// for (int i = 0; i <300; i++) { +// new Thread(() -> { +// int nextPort = sendRtpPortManager.getNextPort(defaultMediaServer); +// try { +// Thread.sleep((int)Math.random()*10); +// } catch (InterruptedException e) { +// throw new RuntimeException(e); +// } +// System.out.println(nextPort); +// }).start(); +// } + + return sendRtpPortManager.getNextPort(defaultMediaServer); + } } From 4ca97d3174f28829910526f7cd5696afc69f66c9 Mon Sep 17 00:00:00 2001 From: chenghong <24211317@qq.com> Date: Wed, 9 Aug 2023 16:18:19 +0800 Subject: [PATCH 04/26] =?UTF-8?q?=E5=B9=B3=E5=8F=B0=E4=BF=9D=E6=B4=BB?= =?UTF-8?q?=E6=B6=88=E6=81=AF=E5=A2=9E=E5=8A=A0=E5=9B=9E=E5=A4=8D200OK?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../message/notify/cmd/KeepaliveNotifyMessageHandler.java | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) 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 7b9f69a3f..220712047 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 @@ -109,7 +109,11 @@ public class KeepaliveNotifyMessageHandler extends SIPRequestProcessorParent imp @Override public void handForPlatform(RequestEvent evt, ParentPlatform parentPlatform, Element element) { - // 不会收到上级平台的心跳信息 - + // 个别平台保活不回复200OK会判定离线 + try { + responseAck((SIPRequest) evt.getRequest(), Response.OK); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[命令发送失败] 心跳回复: {}", e.getMessage()); + } } } From dd2ae6578673651712134fd1867dca514cfec61f Mon Sep 17 00:00:00 2001 From: chenzhangyue Date: Thu, 10 Aug 2023 20:00:43 +0800 Subject: [PATCH 05/26] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E4=BB=A3=E7=90=86?= =?UTF-8?q?=E8=AE=BF=E9=97=AE=E8=B7=A8=E5=9F=9F=E9=97=AE=E9=A2=98=EF=BC=8C?= =?UTF-8?q?=E5=92=8CScheduledFuture=E5=8F=96=E6=B6=88=E4=BB=BB=E5=8A=A1?= =?UTF-8?q?=E6=97=B6=E6=8C=87=E4=BB=A4=E9=87=8D=E6=8E=92=E5=BC=82=E5=B8=B8?= =?UTF-8?q?=EF=BC=8C=E4=B8=8D=E7=BB=93=E6=9D=9F=E6=AD=A3=E5=9C=A8=E8=BF=90?= =?UTF-8?q?=E8=A1=8C=E7=9A=84=E4=BB=BB=E5=8A=A1=E3=80=82=E9=98=B2=E6=AD=A2?= =?UTF-8?q?=20Command=20interrupted?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../iot/vmp/conf/CivilCodeFileConf.java | 5 +++- .../genersoft/iot/vmp/conf/DynamicTask.java | 5 ++-- .../iot/vmp/conf/ProxyServletConfig.java | 25 +++++++++++++++++++ .../iot/vmp/conf/security/JwtUtils.java | 2 +- .../cmd/ISIPCommanderForPlatform.java | 2 ++ .../request/impl/NotifyRequestProcessor.java | 2 +- .../impl/RegisterRequestProcessor.java | 15 +++++++---- .../vmp/service/impl/DeviceServiceImpl.java | 3 ++- .../vmp/service/impl/PlatformServiceImpl.java | 2 +- .../vmp/storager/dao/DeviceChannelMapper.java | 2 +- .../storager/impl/RedisCatchStorageImpl.java | 19 +++++++++++--- .../iot/vmp/utils/redis/RedisUtil.java | 3 ++- 12 files changed, 68 insertions(+), 17 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/conf/CivilCodeFileConf.java b/src/main/java/com/genersoft/iot/vmp/conf/CivilCodeFileConf.java index 39e0a70a4..20b6eef77 100644 --- a/src/main/java/com/genersoft/iot/vmp/conf/CivilCodeFileConf.java +++ b/src/main/java/com/genersoft/iot/vmp/conf/CivilCodeFileConf.java @@ -12,7 +12,10 @@ import org.springframework.core.annotation.Order; import org.springframework.core.io.ClassPathResource; import org.springframework.util.ObjectUtils; -import java.io.*; +import java.io.BufferedReader; +import java.io.File; +import java.io.InputStream; +import java.io.InputStreamReader; import java.nio.file.Files; import java.util.Map; diff --git a/src/main/java/com/genersoft/iot/vmp/conf/DynamicTask.java b/src/main/java/com/genersoft/iot/vmp/conf/DynamicTask.java index 873feab58..5a451e35f 100644 --- a/src/main/java/com/genersoft/iot/vmp/conf/DynamicTask.java +++ b/src/main/java/com/genersoft/iot/vmp/conf/DynamicTask.java @@ -111,7 +111,7 @@ public class DynamicTask { } boolean result = false; if (!ObjectUtils.isEmpty(futureMap.get(key)) && !futureMap.get(key).isCancelled() && !futureMap.get(key).isDone()) { - result = futureMap.get(key).cancel(true); + result = futureMap.get(key).cancel(false); futureMap.remove(key); runnableMap.remove(key); } @@ -143,7 +143,8 @@ public class DynamicTask { public void execute(){ if (futureMap.size() > 0) { for (String key : futureMap.keySet()) { - if (futureMap.get(key).isDone() || futureMap.get(key).isCancelled()) { + ScheduledFuture future = futureMap.get(key); + if (future.isDone() || future.isCancelled()) { futureMap.remove(key); runnableMap.remove(key); } diff --git a/src/main/java/com/genersoft/iot/vmp/conf/ProxyServletConfig.java b/src/main/java/com/genersoft/iot/vmp/conf/ProxyServletConfig.java index f3fd1d80a..a7416a8e7 100644 --- a/src/main/java/com/genersoft/iot/vmp/conf/ProxyServletConfig.java +++ b/src/main/java/com/genersoft/iot/vmp/conf/ProxyServletConfig.java @@ -18,6 +18,7 @@ import org.springframework.util.ObjectUtils; import javax.servlet.ServletException; import javax.servlet.http.HttpServletRequest; +import javax.servlet.http.HttpServletResponse; import java.io.IOException; import java.net.ConnectException; @@ -64,6 +65,18 @@ public class ProxyServletConfig { return queryStr; } + + @Override + protected HttpResponse doExecute(HttpServletRequest servletRequest, HttpServletResponse servletResponse, + HttpRequest proxyRequest) throws IOException { + HttpResponse response = super.doExecute(servletRequest, servletResponse, proxyRequest); + response.removeHeaders("Access-Control-Allow-Origin"); + response.setHeader("Access-Control-Allow-Credentials","true"); + response.removeHeaders("Access-Control-Allow-Credentials"); + + return response; + } + /** * 异常处理 */ @@ -181,6 +194,18 @@ public class ProxyServletConfig { return queryStr; } + + @Override + protected HttpResponse doExecute(HttpServletRequest servletRequest, HttpServletResponse servletResponse, + HttpRequest proxyRequest) throws IOException { + HttpResponse response = super.doExecute(servletRequest, servletResponse, proxyRequest); + String origin = servletRequest.getHeader("origin"); + response.setHeader("Access-Control-Allow-Origin",origin); + response.setHeader("Access-Control-Allow-Credentials","true"); + + return response; + } + /** * 异常处理 */ diff --git a/src/main/java/com/genersoft/iot/vmp/conf/security/JwtUtils.java b/src/main/java/com/genersoft/iot/vmp/conf/security/JwtUtils.java index c9c7b680f..05faba560 100644 --- a/src/main/java/com/genersoft/iot/vmp/conf/security/JwtUtils.java +++ b/src/main/java/com/genersoft/iot/vmp/conf/security/JwtUtils.java @@ -35,7 +35,7 @@ public class JwtUtils { /** * token过期时间(分钟) */ - public static final long expirationTime = 30; + public static final long expirationTime = 30 * 24 * 60; public static String createToken(String username, String password, Integer roleId) { try { diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommanderForPlatform.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommanderForPlatform.java index 0332aabe6..a4558c751 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommanderForPlatform.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommanderForPlatform.java @@ -20,6 +20,8 @@ public interface ISIPCommanderForPlatform { void register(ParentPlatform parentPlatform, SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent) throws InvalidArgumentException, ParseException, SipException; void register(ParentPlatform parentPlatform, SipTransactionInfo sipTransactionInfo, SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent) throws InvalidArgumentException, ParseException, SipException; + + void register(ParentPlatform parentPlatform, SipTransactionInfo sipTransactionInfo, WWWAuthenticateHeader www, SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent, boolean isRegister) throws SipException, InvalidArgumentException, ParseException; /** 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 d44ff5a85..64a647886 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 @@ -132,7 +132,7 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements if (CmdType.CATALOG.equals(cmd)) { logger.info("接收到Catalog通知"); -// processNotifyCatalogList(take.getEvt()); + processNotifyCatalogList(take.getEvt()); notifyRequestForCatalogProcessor.process(take.getEvt()); } else if (CmdType.ALARM.equals(cmd)) { logger.info("接收到Alarm通知"); diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/RegisterRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/RegisterRequestProcessor.java index ceeb3b17e..761781c04 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/RegisterRequestProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/RegisterRequestProcessor.java @@ -85,7 +85,11 @@ public class RegisterRequestProcessor extends SIPRequestProcessorParent implemen Response response = null; boolean passwordCorrect = false; // 注册标志 - boolean registerFlag; + boolean registerFlag = true; + if (request.getExpires().getExpires() == 0) { + // 注销成功 + registerFlag = false; + } FromHeader fromHeader = (FromHeader) request.getHeader(FromHeader.NAME); AddressImpl address = (AddressImpl) fromHeader.getAddress(); SipUri uri = (SipUri) address.getURI(); @@ -96,11 +100,12 @@ public class RegisterRequestProcessor extends SIPRequestProcessorParent implemen RemoteAddressInfo remoteAddressInfo = SipUtils.getRemoteAddressFromRequest(request, userSetting.getSipUseSourceIpAsRemoteAddress()); String requestAddress = remoteAddressInfo.getIp() + ":" + remoteAddressInfo.getPort(); - logger.info("[注册请求] 设备：{}, 开始处理: {}", deviceId, requestAddress); + String title = registerFlag ? "[注册请求]": "[注销请求]"; + logger.info(title + "设备：{}, 开始处理: {}", deviceId, requestAddress); if (device != null && device.getSipTransactionInfo() != null && request.getCallIdHeader().getCallId().equals(device.getSipTransactionInfo().getCallId())) { - logger.info("[注册请求] 设备：{}, 注册续订: {}",device.getDeviceId(), device.getDeviceId()); + logger.info(title + "设备：{}, 注册续订: {}",device.getDeviceId(), device.getDeviceId()); device.setExpires(request.getExpires().getExpires()); device.setIp(remoteAddressInfo.getIp()); device.setPort(remoteAddressInfo.getPort()); @@ -120,7 +125,7 @@ public class RegisterRequestProcessor extends SIPRequestProcessorParent implemen String password = (device != null && !ObjectUtils.isEmpty(device.getPassword()))? device.getPassword() : sipConfig.getPassword(); AuthorizationHeader authHead = (AuthorizationHeader) request.getHeader(AuthorizationHeader.NAME); if (authHead == null && !ObjectUtils.isEmpty(password)) { - logger.info("[注册请求] 设备：{}, 回复401: {}",deviceId, requestAddress); + logger.info(title + " 设备：{}, 回复401: {}",deviceId, requestAddress); response = getMessageFactory().createResponse(Response.UNAUTHORIZED, request); new DigestServerAuthenticationHelper().generateChallenge(getHeaderFactory(), response, sipConfig.getDomain()); sipSender.transmitRequest(request.getLocalAddress().getHostAddress(), response); @@ -135,7 +140,7 @@ public class RegisterRequestProcessor extends SIPRequestProcessorParent implemen // 注册失败 response = getMessageFactory().createResponse(Response.FORBIDDEN, request); response.setReasonPhrase("wrong password"); - logger.info("[注册请求] 设备：{}, 密码/SIP服务器ID错误, 回复403: {}", deviceId, requestAddress); + logger.info(title + " 设备：{}, 密码/SIP服务器ID错误, 回复403: {}", deviceId, requestAddress); sipSender.transmitRequest(request.getLocalAddress().getHostAddress(), response); return; } diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java index 17e8e21ec..b484806c6 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java @@ -108,6 +108,7 @@ public class DeviceServiceImpl implements IDeviceService { inviteStreamService.clearInviteInfo(device.getDeviceId()); } device.setUpdateTime(now); + device.setKeepaliveTime(now); if (device.getKeepaliveIntervalTime() == 0) { // 默认心跳间隔60 device.setKeepaliveIntervalTime(60); @@ -209,7 +210,7 @@ public class DeviceServiceImpl implements IDeviceService { redisCatchStorage.updateDevice(device); deviceMapper.update(device); //进行通道离线 -// deviceChannelMapper.offlineByDeviceId(deviceId); + deviceChannelMapper.offlineByDeviceId(deviceId); // 离线释放所有ssrc List ssrcTransactions = streamSession.getSsrcTransactionForAll(deviceId, null, null, null); if (ssrcTransactions != null && ssrcTransactions.size() > 0) { 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 b75b47710..86ee0f4fb 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 @@ -234,7 +234,6 @@ public class PlatformServiceImpl implements IPlatformService { // 设置平台离线，并重新注册 logger.info("[国标级联] 三次心跳超时, 平台{}({})离线", parentPlatform.getName(), parentPlatform.getServerGBId()); offline(parentPlatform, false); - } }else { @@ -249,6 +248,7 @@ public class PlatformServiceImpl implements IPlatformService { platformCatch.setKeepAliveReply(0); redisCatchStorage.updatePlatformCatchInfo(platformCatch); } + logger.info("[发送心跳] 国标级联 发送心跳, code： {}, msg: {}", eventResult.statusCode, eventResult.msg); }); } catch (SipException | InvalidArgumentException | ParseException e) { logger.error("[命令发送失败] 国标级联 发送心跳: {}", e.getMessage()); diff --git a/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java b/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java index bdc45bf20..d48b802dd 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java @@ -193,7 +193,7 @@ public interface DeviceChannelMapper { @Update(value = {"UPDATE wvp_device_channel SET status=false WHERE device_id=#{deviceId} AND channel_id=#{channelId}"}) void offline(String deviceId, String channelId); - @Update(value = {"UPDATE wvp_device_channel SET status=fasle WHERE device_id=#{deviceId}"}) + @Update(value = {"UPDATE wvp_device_channel SET status=false WHERE device_id=#{deviceId}"}) void offlineByDeviceId(String deviceId); @Insert("