From 46bf5b7a398870dff3b50d601e7dbf1ddc697fae Mon Sep 17 00:00:00 2001 From: panlinlin <648540858@qq.com> Date: Tue, 11 Jun 2024 23:49:55 +0800 Subject: [PATCH 1/2] =?UTF-8?q?1078-=E6=94=AF=E6=8C=81=E6=97=A0=E4=BA=BA?= =?UTF-8?q?=E8=A7=82=E7=9C=8B=E8=87=AA=E5=8A=A8=E5=81=9C=E6=B5=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../service/impl/jt1078ServiceImpl.java | 46 ++++++++++++++----- .../vmp/service/impl/MediaServiceImpl.java | 18 ++++++-- 2 files changed, 50 insertions(+), 14 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/jt1078/service/impl/jt1078ServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/jt1078/service/impl/jt1078ServiceImpl.java index ba7b3e7f6..6be1943ee 100644 --- a/src/main/java/com/genersoft/iot/vmp/jt1078/service/impl/jt1078ServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/jt1078/service/impl/jt1078ServiceImpl.java @@ -25,6 +25,9 @@ import com.genersoft.iot.vmp.media.event.hook.Hook; import com.genersoft.iot.vmp.media.event.hook.HookData; import com.genersoft.iot.vmp.media.event.hook.HookSubscribe; import com.genersoft.iot.vmp.media.event.hook.HookType; +import com.genersoft.iot.vmp.media.event.media.MediaArrivalEvent; +import com.genersoft.iot.vmp.media.event.media.MediaDepartureEvent; +import com.genersoft.iot.vmp.media.event.media.MediaNotFoundEvent; import com.genersoft.iot.vmp.media.event.mediaServer.MediaSendRtpStoppedEvent; import com.genersoft.iot.vmp.media.service.IMediaServerService; import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils; @@ -42,6 +45,7 @@ import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.event.EventListener; import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Service; import java.lang.reflect.Field; @@ -71,27 +75,47 @@ public class jt1078ServiceImpl implements Ijt1078Service { @Autowired private IMediaServerService mediaServerService; - @Autowired - private IMediaService mediaService; - @Autowired private DynamicTask dynamicTask; @Autowired private UserSetting userSetting; - @Autowired - private ZLMRESTfulUtils zlmresTfulUtils; - - @Autowired - private CallbackManager callbackManager; - @Autowired private IRedisCatchStorage redisCatchStorage; @Autowired private FtpSetting ftpSetting; + /** + * 流到来的处理 + */ + @Async("taskExecutor") + @org.springframework.context.event.EventListener + public void onApplicationEvent(MediaArrivalEvent event) { + + } + + /** + * 流离开的处理 + */ + @Async("taskExecutor") + @EventListener + public void onApplicationEvent(MediaDepartureEvent event) { + + } + + /** + * 流未找到的处理 + */ + @Async("taskExecutor") + @EventListener + public void onApplicationEvent(MediaNotFoundEvent event) { + + } + + + @Override public JTDevice getDevice(String phoneNumber) { @@ -172,7 +196,7 @@ public class jt1078ServiceImpl implements Ijt1078Service { // 清理数据 redisTemplate.delete(playKey); } - String stream = "jt_play_" + phoneNumber + "_" + channelId; + String stream = "jt_" + phoneNumber + "_" + channelId; MediaServer mediaServer = mediaServerService.getMediaServerForMinimumLoad(null); if (mediaServer == null) { for (GeneralCallback errorCallback : errorCallbacks) { @@ -354,7 +378,7 @@ public class jt1078ServiceImpl implements Ijt1078Service { } String startTimeParam = DateUtil.yyyy_MM_dd_HH_mm_ssTo1078(startTime); String endTimeParam = DateUtil.yyyy_MM_dd_HH_mm_ssTo1078(endTime); - String stream = phoneNumber + "_" + channelId + "_" + startTimeParam + "_" + endTimeParam; + String stream = "jt_" + phoneNumber + "_" + channelId + "_" + startTimeParam + "_" + endTimeParam; MediaServer mediaServer = mediaServerService.getMediaServerForMinimumLoad(null); if (mediaServer == null) { for (GeneralCallback errorCallback : errorCallbacks) { diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServiceImpl.java index efadfc6e4..895228b1c 100755 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServiceImpl.java @@ -10,6 +10,7 @@ import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager; import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander; import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform; +import com.genersoft.iot.vmp.jt1078.service.Ijt1078Service; import com.genersoft.iot.vmp.media.bean.MediaServer; import com.genersoft.iot.vmp.media.bean.ResultForOnPublish; import com.genersoft.iot.vmp.media.zlm.dto.StreamAuthorityInfo; @@ -72,6 +73,9 @@ public class MediaServiceImpl implements IMediaService { @Autowired private ISIPCommander commander; + @Autowired + private Ijt1078Service ijt1078Service; + @Override public boolean authenticatePlay(String app, String stream, String callId) { if (app == null || stream == null) { @@ -267,10 +271,18 @@ public class MediaServiceImpl implements IMediaService { inviteInfo.getChannelId(), inviteInfo.getStream()); storager.stopPlay(inviteInfo.getDeviceId(), inviteInfo.getChannelId()); return result; + }else { + // 判断是否是1078点播 + if (stream.startsWith("jt_")) { + String[] streamParamArray = stream.split("_"); + if (streamParamArray.length == 3) { + ijt1078Service.stopPlay(streamParamArray[1], Integer.parseInt(streamParamArray[2])); + }else if (streamParamArray.length == 5) { + ijt1078Service.stopPlayback(streamParamArray[1], Integer.parseInt(streamParamArray[2])); + } + } } - if (stream.startsWith("1078")) { - return false; - } + SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(null, null, stream, null); if (sendRtpItem != null && "talk".equals(sendRtpItem.getApp() )) { return false; From 291a87e1f7ad015a0f28975440336f7a123c15f5 Mon Sep 17 00:00:00 2001 From: panlinlin <648540858@qq.com> Date: Wed, 12 Jun 2024 00:22:16 +0800 Subject: [PATCH 2/2] =?UTF-8?q?1078-=E6=94=AF=E6=8C=81=E8=87=AA=E5=8A=A8?= =?UTF-8?q?=E7=82=B9=E6=92=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../vmp/jt1078/bean/JTMediaStreamType.java | 5 ++ .../vmp/jt1078/service/Ijt1078Service.java | 2 + .../service/impl/jt1078ServiceImpl.java | 66 +++++++++++++++++-- .../iot/vmp/media/event/media/MediaEvent.java | 10 +++ .../media/event/media/MediaNotFoundEvent.java | 1 + .../media/event/media/MediaPublishEvent.java | 9 --- .../vmp/service/impl/MediaServiceImpl.java | 8 ++- 7 files changed, 82 insertions(+), 19 deletions(-) create mode 100644 src/main/java/com/genersoft/iot/vmp/jt1078/bean/JTMediaStreamType.java diff --git a/src/main/java/com/genersoft/iot/vmp/jt1078/bean/JTMediaStreamType.java b/src/main/java/com/genersoft/iot/vmp/jt1078/bean/JTMediaStreamType.java new file mode 100644 index 000000000..a5833ee70 --- /dev/null +++ b/src/main/java/com/genersoft/iot/vmp/jt1078/bean/JTMediaStreamType.java @@ -0,0 +1,5 @@ +package com.genersoft.iot.vmp.jt1078.bean; + +public enum JTMediaStreamType { + PLAY,PLAYBACK,TALK +} diff --git a/src/main/java/com/genersoft/iot/vmp/jt1078/service/Ijt1078Service.java b/src/main/java/com/genersoft/iot/vmp/jt1078/service/Ijt1078Service.java index 368284755..c1d34ff35 100644 --- a/src/main/java/com/genersoft/iot/vmp/jt1078/service/Ijt1078Service.java +++ b/src/main/java/com/genersoft/iot/vmp/jt1078/service/Ijt1078Service.java @@ -9,6 +9,8 @@ import com.github.pagehelper.PageInfo; import java.util.List; public interface Ijt1078Service { + JTMediaStreamType checkStreamFromJt(String stream); + JTDevice getDevice(String phoneNumber); JTChannel getChannel(Integer terminalDbId, Integer channelId); diff --git a/src/main/java/com/genersoft/iot/vmp/jt1078/service/impl/jt1078ServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/jt1078/service/impl/jt1078ServiceImpl.java index 6be1943ee..7dbec911d 100644 --- a/src/main/java/com/genersoft/iot/vmp/jt1078/service/impl/jt1078ServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/jt1078/service/impl/jt1078ServiceImpl.java @@ -1,6 +1,5 @@ package com.genersoft.iot.vmp.jt1078.service.impl; -import com.alibaba.fastjson2.JSONObject; import com.genersoft.iot.vmp.common.GeneralCallback; import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.common.VideoManagerConstants; @@ -14,7 +13,6 @@ import com.genersoft.iot.vmp.jt1078.bean.common.ConfigAttribute; import com.genersoft.iot.vmp.jt1078.cmd.JT1078Template; import com.genersoft.iot.vmp.jt1078.dao.JTChannelMapper; import com.genersoft.iot.vmp.jt1078.dao.JTTerminalMapper; -import com.genersoft.iot.vmp.jt1078.event.CallbackManager; import com.genersoft.iot.vmp.jt1078.event.FtpUploadEvent; import com.genersoft.iot.vmp.jt1078.proc.request.J1205; import com.genersoft.iot.vmp.jt1078.proc.response.*; @@ -30,12 +28,11 @@ import com.genersoft.iot.vmp.media.event.media.MediaDepartureEvent; import com.genersoft.iot.vmp.media.event.media.MediaNotFoundEvent; import com.genersoft.iot.vmp.media.event.mediaServer.MediaSendRtpStoppedEvent; import com.genersoft.iot.vmp.media.service.IMediaServerService; -import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils; -import com.genersoft.iot.vmp.service.IMediaService; import com.genersoft.iot.vmp.service.bean.InviteErrorCode; import com.genersoft.iot.vmp.service.bean.SSRCInfo; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.utils.DateUtil; +import com.genersoft.iot.vmp.utils.MediaServerUtils; import com.genersoft.iot.vmp.vmanager.bean.ErrorCode; import com.github.pagehelper.PageHelper; import com.github.pagehelper.PageInfo; @@ -111,11 +108,59 @@ public class jt1078ServiceImpl implements Ijt1078Service { @Async("taskExecutor") @EventListener public void onApplicationEvent(MediaNotFoundEvent event) { - + if (!userSetting.isAutoApplyPlay()) { + return; + } + JTMediaStreamType jtMediaStreamType = checkStreamFromJt(event.getStream()); + if (jtMediaStreamType == null){ + return; + } + String[] streamParamArray = event.getStream().split("_"); + String phoneNumber = streamParamArray[1]; + int channelId = Integer.parseInt(streamParamArray[2]); + String params = event.getParams(); + Map paramMap = MediaServerUtils.urlParamToMap(params); + int type = 0; + try { + type = Integer.parseInt(paramMap.get("type")); + }catch (NumberFormatException ignored) {} + if (jtMediaStreamType.equals(JTMediaStreamType.PLAY)) { + play(phoneNumber, channelId, 0, null); + }else if (jtMediaStreamType.equals(JTMediaStreamType.PLAYBACK)) { + String startTimeParam = DateUtil.jt1078Toyyyy_MM_dd_HH_mm_ss(streamParamArray[3]); + String endTimeParam = DateUtil.jt1078Toyyyy_MM_dd_HH_mm_ss(streamParamArray[4]); + int rate = 0; + int playbackType = 0; + int playbackSpeed = 0; + try { + rate = Integer.parseInt(paramMap.get("rate")); + playbackType = Integer.parseInt(paramMap.get("playbackType")); + playbackSpeed = Integer.parseInt(paramMap.get("playbackSpeed")); + }catch (NumberFormatException ignored) {} + playback(phoneNumber, channelId, startTimeParam, endTimeParam, type, rate, playbackType, playbackSpeed, null); + } } - + /** + * 校验流是否是属于部标的 + */ + @Override + public JTMediaStreamType checkStreamFromJt(String stream) { + if (!stream.startsWith("jt_")) { + return null; + } + String[] streamParamArray = stream.split("_"); + if (streamParamArray.length == 3) { + return JTMediaStreamType.PLAY; + }else if (streamParamArray.length == 5) { + return JTMediaStreamType.PLAYBACK; + }else if (streamParamArray.length == 4) { + return JTMediaStreamType.TALK; + }else { + return null; + } + } @Override public JTDevice getDevice(String phoneNumber) { @@ -213,6 +258,9 @@ public class jt1078ServiceImpl implements Ijt1078Service { StreamInfo info = onPublishHandler(mediaServer, hookData, phoneNumber, channelId); for (GeneralCallback errorCallback : errorCallbacks) { + if (errorCallback == null) { + continue; + } errorCallback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), info); } subscribe.removeSubscribe(hook); @@ -232,6 +280,10 @@ public class jt1078ServiceImpl implements Ijt1078Service { }); // 开启收流端口 SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServer, stream, "000", false, false, 0, false, !channel.getHasAudio(), false, 1); + if (ssrcInfo == null) { + stopPlay(phoneNumber, channelId); + return; + } // 设置超时监听 dynamicTask.startDelay(playKey, () -> { logger.info("[1078-点播] 超时， phoneNumber： {}， channelId： {}", phoneNumber, channelId); @@ -928,7 +980,7 @@ public class jt1078ServiceImpl implements Ijt1078Service { throw new ControllerException(ErrorCode.ERROR100.getCode(), "对讲进行中"); } - String receiveStream = "1078" + "_" + phoneNumber + "_" + channelId; + String receiveStream = "jt_" + phoneNumber + "_" + channelId + "_talk"; MediaServer mediaServer = mediaServerService.getOne(mediaServerId); if (mediaServer == null) { for (GeneralCallback errorCallback : errorCallbacks) { diff --git a/src/main/java/com/genersoft/iot/vmp/media/event/media/MediaEvent.java b/src/main/java/com/genersoft/iot/vmp/media/event/media/MediaEvent.java index b6fb89022..1ddcdd459 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/event/media/MediaEvent.java +++ b/src/main/java/com/genersoft/iot/vmp/media/event/media/MediaEvent.java @@ -20,6 +20,16 @@ public class MediaEvent extends ApplicationEvent { private String schema; + private String params; + + public String getParams() { + return params; + } + + public void setParams(String params) { + this.params = params; + } + public String getApp() { return app; diff --git a/src/main/java/com/genersoft/iot/vmp/media/event/media/MediaNotFoundEvent.java b/src/main/java/com/genersoft/iot/vmp/media/event/media/MediaNotFoundEvent.java index 2415566a9..803e44103 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/event/media/MediaNotFoundEvent.java +++ b/src/main/java/com/genersoft/iot/vmp/media/event/media/MediaNotFoundEvent.java @@ -17,6 +17,7 @@ public class MediaNotFoundEvent extends MediaEvent { mediaDepartureEven.setStream(hookParam.getStream()); mediaDepartureEven.setSchema(hookParam.getSchema()); mediaDepartureEven.setMediaServer(mediaServer); + mediaDepartureEven.setParams(hookParam.getParams()); return mediaDepartureEven; } } diff --git a/src/main/java/com/genersoft/iot/vmp/media/event/media/MediaPublishEvent.java b/src/main/java/com/genersoft/iot/vmp/media/event/media/MediaPublishEvent.java index 0d9f032c7..4b5fa7052 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/event/media/MediaPublishEvent.java +++ b/src/main/java/com/genersoft/iot/vmp/media/event/media/MediaPublishEvent.java @@ -21,13 +21,4 @@ public class MediaPublishEvent extends MediaEvent { return mediaPublishEvent; } - private String params; - - public String getParams() { - return params; - } - - public void setParams(String params) { - this.params = params; - } } diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServiceImpl.java index 895228b1c..87312a465 100755 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServiceImpl.java @@ -10,6 +10,7 @@ import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager; import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander; import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform; +import com.genersoft.iot.vmp.jt1078.bean.JTMediaStreamType; import com.genersoft.iot.vmp.jt1078.service.Ijt1078Service; import com.genersoft.iot.vmp.media.bean.MediaServer; import com.genersoft.iot.vmp.media.bean.ResultForOnPublish; @@ -273,11 +274,12 @@ public class MediaServiceImpl implements IMediaService { return result; }else { // 判断是否是1078点播 - if (stream.startsWith("jt_")) { + JTMediaStreamType jtMediaStreamType = ijt1078Service.checkStreamFromJt(stream); + if (jtMediaStreamType != null) { String[] streamParamArray = stream.split("_"); - if (streamParamArray.length == 3) { + if (jtMediaStreamType.equals(JTMediaStreamType.PLAY)) { ijt1078Service.stopPlay(streamParamArray[1], Integer.parseInt(streamParamArray[2])); - }else if (streamParamArray.length == 5) { + }else if (jtMediaStreamType.equals(JTMediaStreamType.PLAYBACK)) { ijt1078Service.stopPlayback(streamParamArray[1], Integer.parseInt(streamParamArray[2])); } }