From 764d04b497356ba6bcbb75fd42b51eca750f7223 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期三, 29 五月 2024 15:02:51 +0800
Subject: [PATCH] 调整上级观看消息的发送

---
 src/main/java/com/genersoft/iot/vmp/service/impl/MediaServiceImpl.java |  323 +++++++++++++++++++++++++++++++++++++++++++----------
 1 files changed, 259 insertions(+), 64 deletions(-)

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 9c54068..144fb59 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
@@ -1,22 +1,38 @@
 package com.genersoft.iot.vmp.service.impl;
 
-import com.genersoft.iot.vmp.common.StreamInfo;
-import com.genersoft.iot.vmp.conf.MediaConfig;
-import com.genersoft.iot.vmp.media.bean.MediaInfo;
-import com.genersoft.iot.vmp.media.service.IMediaServerService;
-import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookListener;
-import com.genersoft.iot.vmp.media.zlm.dto.MediaServer;
+import com.genersoft.iot.vmp.common.InviteInfo;
+import com.genersoft.iot.vmp.common.InviteSessionType;
+import com.genersoft.iot.vmp.common.VideoManagerConstants;
+import com.genersoft.iot.vmp.conf.UserSetting;
+import com.genersoft.iot.vmp.conf.exception.ControllerException;
+import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;
+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.media.bean.MediaServer;
+import com.genersoft.iot.vmp.media.bean.ResultForOnPublish;
 import com.genersoft.iot.vmp.media.zlm.dto.StreamAuthorityInfo;
-import com.genersoft.iot.vmp.media.zlm.dto.hook.HookResultForOnPublish;
-import com.genersoft.iot.vmp.service.IMediaService;
+import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
+import com.genersoft.iot.vmp.service.*;
 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.utils.MediaServerUtils;
+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 org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Service;
-import org.springframework.util.ObjectUtils;
 
+import javax.sip.InvalidArgumentException;
+import javax.sip.SipException;
+import java.text.ParseException;
 import java.util.List;
+import java.util.Map;
 
 @Service
 public class MediaServiceImpl implements IMediaService {
@@ -27,70 +43,34 @@
     private IRedisCatchStorage redisCatchStorage;
 
     @Autowired
-    private IMediaServerService mediaServerService;
+    private IStreamProxyService streamProxyService;
 
     @Autowired
-    private MediaConfig mediaConfig;
+    private UserSetting userSetting;
 
+    @Autowired
+    private RedisTemplate<Object, Object> redisTemplate;
 
-    @Override
-    public StreamInfo getStreamInfoByAppAndStream(MediaServer mediaServerItem, String app, String stream, MediaInfo mediaInfo, String callId) {
-        return getStreamInfoByAppAndStream(mediaServerItem, app, stream, mediaInfo, null, callId, true);
-    }
+    @Autowired
+    private IUserService userService;
 
-    @Override
-    public StreamInfo getStreamInfoByAppAndStreamWithCheck(String app, String stream, String mediaServerId, String addr, boolean authority) {
-        StreamInfo streamInfo = null;
-        if (mediaServerId == null) {
-            mediaServerId = mediaConfig.getId();
-        }
-        MediaServer mediaInfo = mediaServerService.getOne(mediaServerId);
-        if (mediaInfo == null) {
-            return null;
-        }
-        String calld = null;
-        StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(app, stream);
-        if (streamAuthorityInfo != null) {
-            calld = streamAuthorityInfo.getCallId();
-        }
-        List<StreamInfo> streamInfoList = mediaServerService.getMediaList(mediaInfo, app, stream, calld);
-        if (streamInfoList.isEmpty()) {
-            return null;
-        }else {
-            return streamInfoList.get(0);
-        }
-    }
+    @Autowired
+    private IInviteStreamService inviteStreamService;
 
+    @Autowired
+    private VideoStreamSessionManager sessionManager;
 
+    @Autowired
+    private IVideoManagerStorage storager;
 
-    @Override
-    public StreamInfo getStreamInfoByAppAndStreamWithCheck(String app, String stream, String mediaServerId, boolean authority) {
-        return getStreamInfoByAppAndStreamWithCheck(app, stream, mediaServerId, null, authority);
-    }
+    @Autowired
+    private IDeviceService deviceService;
 
-    @Override
-    public StreamInfo getStreamInfoByAppAndStream(MediaServer mediaServer, String app, String stream, MediaInfo mediaInfo, String addr, String callId, boolean isPlay) {
-        StreamInfo streamInfoResult = new StreamInfo();
-        streamInfoResult.setStream(stream);
-        streamInfoResult.setApp(app);
-        if (addr == null) {
-            addr = mediaServer.getStreamIp();
-        }
+    @Autowired
+    private ISIPCommanderForPlatform commanderForPlatform;
 
-        streamInfoResult.setIp(addr);
-        streamInfoResult.setMediaServerId(mediaServer.getId());
-        String callIdParam = ObjectUtils.isEmpty(callId)?"":"?callId=" + callId;
-        streamInfoResult.setRtmp(addr, mediaServer.getRtmpPort(),mediaServer.getRtmpSSlPort(), app,  stream, callIdParam);
-        streamInfoResult.setRtsp(addr, mediaServer.getRtspPort(),mediaServer.getRtspSSLPort(), app,  stream, callIdParam);
-        streamInfoResult.setFlv(addr, mediaServer.getHttpPort(),mediaServer.getHttpSSlPort(), app,  stream, callIdParam);
-        streamInfoResult.setFmp4(addr, mediaServer.getHttpPort(),mediaServer.getHttpSSlPort(), app,  stream, callIdParam);
-        streamInfoResult.setHls(addr, mediaServer.getHttpPort(),mediaServer.getHttpSSlPort(), app,  stream, callIdParam);
-        streamInfoResult.setTs(addr, mediaServer.getHttpPort(),mediaServer.getHttpSSlPort(), app,  stream, callIdParam);
-        streamInfoResult.setRtc(addr, mediaServer.getHttpPort(),mediaServer.getHttpSSlPort(), app,  stream, callIdParam, isPlay);
-
-        streamInfoResult.setMediaInfo(mediaInfo);
-        return streamInfoResult;
-    }
+    @Autowired
+    private ISIPCommander commander;
 
     @Override
     public boolean authenticatePlay(String app, String stream, String callId) {
@@ -101,6 +81,221 @@
             return true;
         }
         StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(app, stream);
-        return (streamAuthorityInfo != null && streamAuthorityInfo.getCallId() != null && !streamAuthorityInfo.getCallId().equals(callId));
+        if (streamAuthorityInfo == null || streamAuthorityInfo.getCallId() == null) {
+            return true;
+        }
+        return streamAuthorityInfo.getCallId().equals(callId);
+    }
+
+    @Override
+    public ResultForOnPublish authenticatePublish(MediaServer mediaServer, String app, String stream, String params) {
+        // 鎺ㄦ祦閴存潈鐨勫鐞�
+        if (!"rtp".equals(app)) {
+            StreamProxyItem streamProxyItem = streamProxyService.getStreamProxyByAppAndStream(app, stream);
+            if (streamProxyItem != null) {
+                ResultForOnPublish result = new ResultForOnPublish();
+                result.setEnable_audio(streamProxyItem.isEnableAudio());
+                result.setEnable_mp4(streamProxyItem.isEnableMp4());
+                return result;
+            }
+            if (userSetting.getPushAuthority()) {
+                // 瀵逛簬鎺ㄦ祦杩涜閴存潈
+                Map<String, String> paramMap = MediaServerUtils.urlParamToMap(params);
+                // 鎺ㄦ祦閴存潈
+                if (params == null) {
+                    logger.info("鎺ㄦ祦閴存潈澶辫触锛� 缂哄皯蹇呰鍙傛暟锛歴ign=md5(user琛ㄧ殑pushKey)");
+                    throw new ControllerException(ErrorCode.ERROR401.getCode(), "Unauthorized");
+                }
+
+                String sign = paramMap.get("sign");
+                if (sign == null) {
+                    logger.info("鎺ㄦ祦閴存潈澶辫触锛� 缂哄皯蹇呰鍙傛暟锛歴ign=md5(user琛ㄧ殑pushKey)");
+                    throw new ControllerException(ErrorCode.ERROR401.getCode(), "Unauthorized");
+                }
+                // 鎺ㄦ祦鑷畾涔夋挱鏀鹃壌鏉冪爜
+                String callId = paramMap.get("callId");
+                // 閴存潈閰嶇疆
+                boolean hasAuthority = userService.checkPushAuthority(callId, sign);
+                if (!hasAuthority) {
+                    logger.info("鎺ㄦ祦閴存潈澶辫触锛� sign 鏃犳潈闄�: callId={}. sign={}", callId, sign);
+                    throw new ControllerException(ErrorCode.ERROR401.getCode(), "Unauthorized");
+                }
+                StreamAuthorityInfo streamAuthorityInfo = StreamAuthorityInfo.getInstanceByHook(app, stream, mediaServer.getId());
+                streamAuthorityInfo.setCallId(callId);
+                streamAuthorityInfo.setSign(sign);
+                // 閴存潈閫氳繃
+                redisCatchStorage.updateStreamAuthorityInfo(app, stream, streamAuthorityInfo);
+            }
+        }
+
+
+        ResultForOnPublish result = new ResultForOnPublish();
+        result.setEnable_audio(true);
+
+        // 鏄惁褰曞儚
+        if ("rtp".equals(app)) {
+            result.setEnable_mp4(userSetting.getRecordSip());
+        } else {
+            result.setEnable_mp4(userSetting.isRecordPushLive());
+        }
+        // 鍥芥爣娴�
+        if ("rtp".equals(app)) {
+
+            InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(null, stream);
+
+            // 鍗曠鍙fā寮忎笅淇敼娴� ID
+            if (!mediaServer.isRtpEnable() && inviteInfo == null) {
+                String ssrc = String.format("%010d", Long.parseLong(stream, 16));
+                inviteInfo = inviteStreamService.getInviteInfoBySSRC(ssrc);
+                if (inviteInfo != null) {
+                    result.setStream_replace(inviteInfo.getStream());
+                    logger.info("[ZLM HOOK]鎺ㄦ祦閴存潈 stream: {} 鏇挎崲涓� {}", stream, inviteInfo.getStream());
+                    stream = inviteInfo.getStream();
+                }
+            }
+
+            // 璁剧疆闊抽淇℃伅鍙婂綍鍒朵俊鎭�
+            List<SsrcTransaction> ssrcTransactionForAll = sessionManager.getSsrcTransactionForAll(null, null, null, stream);
+            if (ssrcTransactionForAll != null && ssrcTransactionForAll.size() == 1) {
+
+                // 涓哄綍鍒跺浗鏍囨ā鎷熶竴涓壌鏉冧俊鎭�, 鏂逛究鍚庣画鍐欏叆褰曞儚鏂囦欢鏃朵娇鐢�
+                StreamAuthorityInfo streamAuthorityInfo = StreamAuthorityInfo.getInstanceByHook(app, stream, mediaServer.getId());
+                streamAuthorityInfo.setApp(app);
+                streamAuthorityInfo.setStream(ssrcTransactionForAll.get(0).getStream());
+                streamAuthorityInfo.setCallId(ssrcTransactionForAll.get(0).getSipTransactionInfo().getCallId());
+
+                redisCatchStorage.updateStreamAuthorityInfo(app, ssrcTransactionForAll.get(0).getStream(), streamAuthorityInfo);
+
+                String deviceId = ssrcTransactionForAll.get(0).getDeviceId();
+                String channelId = ssrcTransactionForAll.get(0).getChannelId();
+                DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
+                if (deviceChannel != null) {
+                    result.setEnable_audio(deviceChannel.getHasAudio());
+                }
+                // 濡傛灉鏄綍鍍忎笅杞藉氨璁剧疆瑙嗛闂撮殧鍗佺
+                if (ssrcTransactionForAll.get(0).getType() == InviteSessionType.DOWNLOAD) {
+                    // 鑾峰彇褰曞儚鐨勬�绘椂闀匡紝鐒跺悗璁剧疆涓鸿繖涓棰戠殑鏃堕暱
+                    InviteInfo inviteInfoForDownload = inviteStreamService.getInviteInfo(InviteSessionType.DOWNLOAD, deviceId, channelId, stream);
+                    if (inviteInfoForDownload != null && inviteInfoForDownload.getStreamInfo() != null) {
+                        String startTime = inviteInfoForDownload.getStreamInfo().getStartTime();
+                        String endTime = inviteInfoForDownload.getStreamInfo().getEndTime();
+                        long difference = DateUtil.getDifference(startTime, endTime) / 1000;
+                        result.setMp4_max_second((int) difference);
+                        result.setEnable_mp4(true);
+                        // 璁剧疆涓�2淇濊瘉寰楀埌鐨刴p4鐨勬椂闀挎槸姝e父鐨�
+                        result.setModify_stamp(2);
+                    }
+                }
+                // 濡傛灉鏄痶alk瀵硅锛屽垯榛樿鑾峰彇澹伴煶
+                if (ssrcTransactionForAll.get(0).getType() == InviteSessionType.TALK) {
+                    result.setEnable_audio(true);
+                }
+            }
+        } else if (app.equals("broadcast")) {
+            result.setEnable_audio(true);
+        } else if (app.equals("talk")) {
+            result.setEnable_audio(true);
+        }
+        if (app.equalsIgnoreCase("rtp")) {
+            String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_RTP_INFO + userSetting.getServerId() + "_" + stream;
+            OtherRtpSendInfo otherRtpSendInfo = (OtherRtpSendInfo) redisTemplate.opsForValue().get(receiveKey);
+
+            String receiveKeyForPS = VideoManagerConstants.WVP_OTHER_RECEIVE_PS_INFO + userSetting.getServerId() + "_" + stream;
+            OtherPsSendInfo otherPsSendInfo = (OtherPsSendInfo) redisTemplate.opsForValue().get(receiveKeyForPS);
+            if (otherRtpSendInfo != null || otherPsSendInfo != null) {
+                result.setEnable_mp4(true);
+            }
+        }
+        return result;
+    }
+
+    @Override
+    public boolean closeStreamOnNoneReader(String mediaServerId, String app, String stream, String schema) {
+        boolean result = false;
+        // 鍥芥爣绫诲瀷鐨勬祦
+        if ("rtp".equals(app)) {
+            result = userSetting.getStreamOnDemand();
+            // 鍥芥爣娴侊紝 鐐规挱/褰曞儚鍥炴斁/褰曞儚涓嬭浇
+            InviteInfo inviteInfo = inviteStreamService.getInviteInfoByStream(null, stream);
+            // 鐐规挱
+            if (inviteInfo != null) {
+                // 褰曞儚涓嬭浇
+                if (inviteInfo.getType() == InviteSessionType.DOWNLOAD) {
+                    return false;
+                }
+                // 鏀跺埌鏃犱汉瑙傜湅璇存槑娴佷篃娌℃湁鍦ㄥ線涓婄骇鎺ㄩ��
+                if (redisCatchStorage.isChannelSendingRTP(inviteInfo.getChannelId())) {
+                    List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByChannelId(
+                            inviteInfo.getChannelId());
+                    if (!sendRtpItems.isEmpty()) {
+                        for (SendRtpItem sendRtpItem : sendRtpItems) {
+                            ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
+                            try {
+                                commanderForPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId());
+                            } catch (SipException | InvalidArgumentException | ParseException e) {
+                                logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 鍙戦�丅YE: {}", e.getMessage());
+                            }
+                            redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), sendRtpItem.getChannelId(),
+                                    sendRtpItem.getCallId(), sendRtpItem.getStream());
+                            if (InviteStreamType.PUSH == sendRtpItem.getPlayType()) {
+                                redisCatchStorage.sendPlatformStopPlayMsg(sendRtpItem,parentPlatform);
+                            }
+                        }
+                    }
+                }
+                Device device = deviceService.getDevice(inviteInfo.getDeviceId());
+                if (device != null) {
+                    try {
+                        // 澶氭煡璇竴娆¢槻姝㈠凡缁忚澶勭悊浜�
+                        InviteInfo info = inviteStreamService.getInviteInfo(inviteInfo.getType(),
+                                inviteInfo.getDeviceId(), inviteInfo.getChannelId(), inviteInfo.getStream());
+                        if (info != null) {
+                            commander.streamByeCmd(device, inviteInfo.getChannelId(),
+                                    inviteInfo.getStream(), null);
+                        } else {
+                            logger.info("[鏃犱汉瑙傜湅] 鏈壘鍒拌澶囩殑鐐规挱淇℃伅锛� {}锛� 娴侊細{}", inviteInfo.getDeviceId(), stream);
+                        }
+                    } catch (InvalidArgumentException | ParseException | SipException |
+                             SsrcTransactionNotFoundException e) {
+                        logger.error("[鏃犱汉瑙傜湅]鐐规挱锛� 鍙戦�丅YE澶辫触 {}", e.getMessage());
+                    }
+                } else {
+                    logger.info("[鏃犱汉瑙傜湅] 鏈壘鍒拌澶囷細 {}锛屾祦锛歿}", inviteInfo.getDeviceId(), stream);
+                }
+
+                inviteStreamService.removeInviteInfo(inviteInfo.getType(), inviteInfo.getDeviceId(),
+                        inviteInfo.getChannelId(), inviteInfo.getStream());
+                storager.stopPlay(inviteInfo.getDeviceId(), inviteInfo.getChannelId());
+                return result;
+            }
+            SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(null, null, stream, null);
+            if (sendRtpItem != null && "talk".equals(sendRtpItem.getApp())) {
+                return false;
+            }
+        } else if ("talk".equals(app) || "broadcast".equals(app)) {
+            return false;
+        } else {
+            // 闈炲浗鏍囨祦 鎺ㄦ祦/鎷夋祦浠g悊
+            // 鎷夋祦浠g悊
+            StreamProxyItem streamProxyItem = streamProxyService.getStreamProxyByAppAndStream(app, stream);
+            if (streamProxyItem != null) {
+                if (streamProxyItem.isEnableRemoveNoneReader()) {
+                    // 鏃犱汉瑙傜湅鑷姩绉婚櫎
+                    result = true;
+                    streamProxyService.del(app, stream);
+                    String url = streamProxyItem.getUrl() != null ? streamProxyItem.getUrl() : streamProxyItem.getSrcUrl();
+                    logger.info("[{}/{}]<-[{}] 鎷夋祦浠g悊鏃犱汉瑙傜湅宸茬粡绉婚櫎", app, stream, url);
+                } else if (streamProxyItem.isEnableDisableNoneReader()) {
+                    // 鏃犱汉瑙傜湅鍋滅敤
+                    result = true;
+                    // 淇敼鏁版嵁
+                    streamProxyService.stop(app, stream);
+                } else {
+                    // 鏃犱汉瑙傜湅涓嶅仛澶勭悊
+                    result = false;
+                }
+            }
+        }
+        return result;
     }
 }

--
Gitblit v1.8.0