From 922a9e8b3216ee5b6fd36ce607eaa81f6adf70ca Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期二, 07 二月 2023 14:52:33 +0800
Subject: [PATCH] 修复语音对讲返回的地址

---
 src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java |  316 +++++++++++++++++++++++++++++++++++-----------------
 1 files changed, 210 insertions(+), 106 deletions(-)

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 dc3f5d8..56efa72 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
@@ -24,24 +24,25 @@
 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.IDeviceService;
 import com.genersoft.iot.vmp.service.IMediaServerService;
 import com.genersoft.iot.vmp.service.IMediaService;
 import com.genersoft.iot.vmp.service.IPlayService;
-import com.genersoft.iot.vmp.service.bean.InviteTimeOutCallback;
-import com.genersoft.iot.vmp.service.bean.PlayBackCallback;
-import com.genersoft.iot.vmp.service.bean.PlayBackResult;
-import com.genersoft.iot.vmp.service.bean.SSRCInfo;
+import com.genersoft.iot.vmp.service.bean.*;
+import com.genersoft.iot.vmp.service.redisMsg.RedisGbPlayMsgListener;
 import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
 import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
 import com.genersoft.iot.vmp.utils.DateUtil;
 import com.genersoft.iot.vmp.utils.redis.RedisUtil;
 import com.genersoft.iot.vmp.vmanager.bean.AudioBroadcastResult;
 import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
+import com.genersoft.iot.vmp.vmanager.bean.StreamContent;
 import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
 import com.genersoft.iot.vmp.vmanager.gb28181.play.bean.AudioBroadcastEvent;
+import gov.nist.javax.sip.message.SIPResponse;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -49,18 +50,15 @@
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.stereotype.Service;
 import org.springframework.util.ObjectUtils;
-import org.springframework.web.context.request.async.DeferredResult;
 
 import javax.sip.InvalidArgumentException;
 import javax.sip.ResponseEvent;
 import javax.sip.SipException;
+import javax.sip.header.CallIdHeader;
 import java.math.BigDecimal;
 import java.math.RoundingMode;
 import java.text.ParseException;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.UUID;
+import java.util.*;
 
 @SuppressWarnings(value = {"rawtypes", "unchecked"})
 @Service
@@ -119,10 +117,19 @@
     @Autowired
     private ZlmHttpHookSubscribe subscribe;
 
+    @Autowired
+    private ISIPCommanderForPlatform commanderForPlatform;
+
 
     @Qualifier("taskExecutor")
     @Autowired
     private ThreadPoolTaskExecutor taskExecutor;
+
+    @Autowired
+    private RedisGbPlayMsgListener redisGbPlayMsgListener;
+
+    @Autowired
+    private ZlmHttpHookSubscribe hookSubscribe;
 
 
     @Override
@@ -347,7 +354,7 @@
 //                        }
                         SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, ip, port, ssrcInfo.getSsrc(), device.getDeviceId(),
                                 device.getDeviceId(), channelId,
-                                false);
+                                false, false);
 
 
 //                        if (sendRtpItem.getLocalPort() == 0) {
@@ -375,6 +382,7 @@
                         sendRtpItem.setStreamId("1000");
                         sendRtpItem.setSsrc(ssrc);
                         sendRtpItem.setOnlyAudio(true);
+                        sendRtpItem.setRtcp(false);
                         redisCatchStorage.updateSendRTPSever(sendRtpItem);
 
                         Map<String, Object> param = new HashMap<>(12);
@@ -446,6 +454,9 @@
                     mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream());
                     streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
                     mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream());
+                    // 鍙栨秷璁㈤槄娑堟伅鐩戝惉
+                    HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId());
+                    subscribe.removeSubscribe(hookSubscribe);
                 }
             }
         }, userSetting.getPlayTimeout());
@@ -455,7 +466,6 @@
             dynamicTask.stop(timeOutTaskKey);
             // 閲婃斁ssrc
             mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
-
             streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
 
             RequestMessage msg = new RequestMessage();
@@ -473,7 +483,7 @@
                 onPublishHandlerForPlay(mediaServerItemInuse, response, device.getDeviceId(), channelId);
                 hookEvent.response(mediaServerItemInuse, response);
                 logger.info("[鐐规挱鎴愬姛] deviceId: {}, channelId: {}", device.getDeviceId(), channelId);
-                String streamUrl = String.format("rtsp://127.0.0.1:%s/%s/%s", mediaServerItemInuse.getRtspPort(), "rtp",  ssrcInfo.getStream());
+                String streamUrl = String.format("http://127.0.0.1:%s/%s/%s.live.flv", mediaServerItemInuse.getHttpPort(), "rtp",  ssrcInfo.getStream());
                 String path = "snap";
                 String fileName = device.getDeviceId() + "_" + channelId + ".jpg";
                 // 璇锋眰鎴浘
@@ -581,14 +591,10 @@
         }
     }
 
-    private void onPublishHandlerForPlayback(MediaServerItem mediaServerItem, JSONObject response, String deviceId, String channelId, String uuid) {
-        RequestMessage msg = new RequestMessage();
-        msg.setKey(DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId);
-        if (!ObjectUtils.isEmpty(uuid)) {
-            msg.setId(uuid);
-        }
-        StreamInfo streamInfo = onPublishHandler(mediaServerItem, response, deviceId, channelId);
+    private void onPublishHandlerForPlayback(MediaServerItem mediaServerItem, JSONObject response, String deviceId, String channelId, PlayBackCallback playBackCallback) {
 
+        StreamInfo streamInfo = onPublishHandler(mediaServerItem, response, deviceId, channelId);
+        PlayBackResult<StreamInfo> playBackResult = new PlayBackResult<>();
         if (streamInfo != null) {
             DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
             if (deviceChannel != null) {
@@ -597,17 +603,16 @@
             }
             redisCatchStorage.startPlay(streamInfo);
 
-            WVPResult wvpResult = new WVPResult();
-            wvpResult.setCode(ErrorCode.SUCCESS.getCode());
-            wvpResult.setMsg(ErrorCode.SUCCESS.getMsg());
-            wvpResult.setData(streamInfo);
-            msg.setData(wvpResult);
 
-            resultHolder.invokeAllResult(msg);
+            playBackResult.setCode(ErrorCode.SUCCESS.getCode());
+            playBackResult.setMsg(ErrorCode.SUCCESS.getMsg());
+            playBackResult.setData(streamInfo);
+            playBackCallback.call(playBackResult);
         } else {
             logger.warn("褰曞儚鍥炴斁璋冪敤澶辫触锛�");
-            msg.setData(WVPResult.fail(ErrorCode.ERROR100.getCode(), "褰曞儚鍥炴斁璋冪敤澶辫触锛�"));
-            resultHolder.invokeAllResult(msg);
+            playBackResult.setCode(ErrorCode.ERROR100.getCode());
+            playBackResult.setMsg("褰曞儚鍥炴斁璋冪敤澶辫触锛�");
+            playBackCallback.call(playBackResult);
         }
     }
 
@@ -618,7 +623,7 @@
         }
         MediaServerItem mediaServerItem;
         if (ObjectUtils.isEmpty(device.getMediaServerId()) || "auto".equals(device.getMediaServerId())) {
-            mediaServerItem = mediaServerService.getMediaServerForMinimumLoad();
+            mediaServerItem = mediaServerService.getMediaServerForMinimumLoad(null);
         } else {
             mediaServerItem = mediaServerService.getOne(device.getMediaServerId());
         }
@@ -629,45 +634,56 @@
     }
 
     @Override
-    public DeferredResult<WVPResult<StreamInfo>> playBack(String deviceId, String channelId, String startTime,
+    public MediaServerItem getNewMediaServerItemHasAssist(Device device) {
+        if (device == null) {
+            return null;
+        }
+        MediaServerItem mediaServerItem;
+        if (ObjectUtils.isEmpty(device.getMediaServerId()) || "auto".equals(device.getMediaServerId())) {
+            mediaServerItem = mediaServerService.getMediaServerForMinimumLoad(true);
+        } else {
+            mediaServerItem = mediaServerService.getOne(device.getMediaServerId());
+        }
+        if (mediaServerItem == null) {
+            logger.warn("[鑾峰彇鍙敤鐨刏LM鑺傜偣]鏈壘鍒板彲浣跨敤鐨刏LM...");
+        }
+        return mediaServerItem;
+    }
+
+    @Override
+    public void playBack(String deviceId, String channelId, String startTime,
                                                           String endTime, InviteStreamCallback inviteStreamCallback,
                                                           PlayBackCallback callback) {
         Device device = storager.queryVideoDevice(deviceId);
         if (device == null) {
-            return null;
+            return;
         }
         MediaServerItem newMediaServerItem = getNewMediaServerItem(device);
         SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, device.isSsrcCheck(), true);
 
-        return playBack(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, inviteStreamCallback, callback);
+        playBack(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, inviteStreamCallback, callback);
     }
 
     @Override
-    public DeferredResult<WVPResult<StreamInfo>> playBack(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo,
+    public void playBack(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo,
                                                           String deviceId, String channelId, String startTime,
                                                           String endTime, InviteStreamCallback infoCallBack,
                                                           PlayBackCallback playBackCallback) {
         if (mediaServerItem == null || ssrcInfo == null) {
-            return null;
+            return;
         }
-        String uuid = UUID.randomUUID().toString();
-        String key = DeferredResultHolder.CALLBACK_CMD_PLAYBACK + deviceId + channelId;
+
         Device device = storager.queryVideoDevice(deviceId);
         if (device == null) {
             throw new ControllerException(ErrorCode.ERROR100.getCode(), "璁惧锛� " + deviceId + "涓嶅瓨鍦�");
         }
-        DeferredResult<WVPResult<StreamInfo>> result = new DeferredResult<>(30000L);
-        resultHolder.put(DeferredResultHolder.CALLBACK_CMD_PLAYBACK + deviceId + channelId, uuid, result);
-        RequestMessage requestMessage = new RequestMessage();
-        requestMessage.setId(uuid);
-        requestMessage.setKey(key);
-        PlayBackResult<RequestMessage> playBackResult = new PlayBackResult<>();
+
+        PlayBackResult<StreamInfo> playBackResult = new PlayBackResult<>();
         String playBackTimeOutTaskKey = UUID.randomUUID().toString();
         dynamicTask.startDelay(playBackTimeOutTaskKey, () -> {
             logger.warn(String.format("璁惧鍥炴斁瓒呮椂锛宒eviceId锛�%s 锛宑hannelId锛�%s", deviceId, channelId));
             playBackResult.setCode(ErrorCode.ERROR100.getCode());
             playBackResult.setMsg("鍥炴斁瓒呮椂");
-            playBackResult.setData(requestMessage);
 
             try {
                 cmder.streamByeCmd(device, channelId, ssrcInfo.getStream(), null);
@@ -679,19 +695,14 @@
                 mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream());
                 streamSession.remove(deviceId, channelId, ssrcInfo.getStream());
             }
-
             // 鍥炲涔嬪墠鎵�鏈夌殑鐐规挱璇锋眰
             playBackCallback.call(playBackResult);
-            result.setResult(WVPResult.fail(ErrorCode.ERROR100.getCode(), "鍥炴斁瓒呮椂"));
-            resultHolder.exist(DeferredResultHolder.CALLBACK_CMD_PLAYBACK + deviceId + channelId, uuid);
         }, userSetting.getPlayTimeout());
 
         SipSubscribe.Event errorEvent = event -> {
             dynamicTask.stop(playBackTimeOutTaskKey);
-            requestMessage.setData(WVPResult.fail(ErrorCode.ERROR100.getCode(), String.format("鍥炴斁澶辫触锛� 閿欒鐮侊細 %s, %s", event.statusCode, event.msg)));
             playBackResult.setCode(ErrorCode.ERROR100.getCode());
             playBackResult.setMsg(String.format("鍥炴斁澶辫触锛� 閿欒鐮侊細 %s, %s", event.statusCode, event.msg));
-            playBackResult.setData(requestMessage);
             playBackResult.setEvent(event);
             playBackCallback.call(playBackResult);
             streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
@@ -709,11 +720,9 @@
                 return;
             }
             redisCatchStorage.startPlayback(streamInfo, inviteStreamInfo.getCallId());
-            WVPResult<StreamInfo> success = WVPResult.success(streamInfo);
-            requestMessage.setData(success);
             playBackResult.setCode(ErrorCode.SUCCESS.getCode());
             playBackResult.setMsg(ErrorCode.SUCCESS.getMsg());
-            playBackResult.setData(requestMessage);
+            playBackResult.setData(streamInfo);
             playBackResult.setMediaServerItem(inviteStreamInfo.getMediaServerItem());
             playBackResult.setResponse(inviteStreamInfo.getResponse());
             playBackCallback.call(playBackResult);
@@ -760,7 +769,7 @@
                                             logger.info("[ZLM HOOK] ssrc淇鍚庢敹鍒拌闃呮秷鎭細 " + response.toJSONString());
                                             dynamicTask.stop(playBackTimeOutTaskKey);
                                             // hook鍝嶅簲
-                                            onPublishHandlerForPlayback(mediaServerItemInUse, response, device.getDeviceId(), channelId, uuid);
+                                            onPublishHandlerForPlayback(mediaServerItemInUse, response, device.getDeviceId(), channelId, playBackCallback);
                                             hookEvent.call(new InviteStreamInfo(mediaServerItem, null, eventResult.callId, "rtp", ssrcInfo.getStream()));
                                         });
                                     }
@@ -780,50 +789,45 @@
             eventResult.msg = "鍛戒护鍙戦�佸け璐�";
             errorEvent.response(eventResult);
         }
-        return result;
     }
 
 
 
     @Override
-    public DeferredResult<WVPResult<StreamInfo>> download(String deviceId, String channelId, String startTime, String endTime, int downloadSpeed, InviteStreamCallback infoCallBack, PlayBackCallback hookCallBack) {
+    public void download(String deviceId, String channelId, String startTime, String endTime, int downloadSpeed, InviteStreamCallback infoCallBack, PlayBackCallback playBackCallback) {
         Device device = storager.queryVideoDevice(deviceId);
         if (device == null) {
-            return null;
+            return;
         }
-        MediaServerItem newMediaServerItem = getNewMediaServerItem(device);
+        MediaServerItem newMediaServerItem = getNewMediaServerItemHasAssist(device);
+        if (newMediaServerItem == null) {
+            PlayBackResult<StreamInfo> downloadResult = new PlayBackResult<>();
+            downloadResult.setCode(ErrorCode.ERROR100.getCode());
+            downloadResult.setMsg("鏈壘鍒癮ssist鏈嶅姟");
+            playBackCallback.call(downloadResult);
+            return;
+        }
         SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, device.isSsrcCheck(), true);
 
-        return download(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, downloadSpeed, infoCallBack, hookCallBack);
+        download(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, downloadSpeed, infoCallBack, playBackCallback);
     }
 
+
     @Override
-    public DeferredResult<WVPResult<StreamInfo>> download(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, String deviceId, String channelId, String startTime, String endTime, int downloadSpeed, InviteStreamCallback infoCallBack, PlayBackCallback hookCallBack) {
+    public void download(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, String deviceId, String channelId, String startTime, String endTime, int downloadSpeed, InviteStreamCallback infoCallBack, PlayBackCallback hookCallBack) {
         if (mediaServerItem == null || ssrcInfo == null) {
-            return null;
+            return;
         }
-        String uuid = UUID.randomUUID().toString();
-        String key = DeferredResultHolder.CALLBACK_CMD_DOWNLOAD + deviceId + channelId;
-        DeferredResult<WVPResult<StreamInfo>> result = new DeferredResult<>(30000L);
+
         Device device = storager.queryVideoDevice(deviceId);
         if (device == null) {
             throw new ControllerException(ErrorCode.ERROR400.getCode(), "璁惧锛�" + deviceId + "涓嶅瓨鍦�");
         }
-
-        resultHolder.put(key, uuid, result);
-        RequestMessage requestMessage = new RequestMessage();
-        requestMessage.setId(uuid);
-        requestMessage.setKey(key);
-        WVPResult<StreamInfo> wvpResult = new WVPResult<>();
-        requestMessage.setData(wvpResult);
-        PlayBackResult<RequestMessage> downloadResult = new PlayBackResult<>();
-        downloadResult.setData(requestMessage);
+        PlayBackResult<StreamInfo> downloadResult = new PlayBackResult<>();
 
         String downLoadTimeOutTaskKey = UUID.randomUUID().toString();
         dynamicTask.startDelay(downLoadTimeOutTaskKey, () -> {
             logger.warn(String.format("褰曞儚涓嬭浇璇锋眰瓒呮椂锛宒eviceId锛�%s 锛宑hannelId锛�%s", deviceId, channelId));
-            wvpResult.setCode(ErrorCode.ERROR100.getCode());
-            wvpResult.setMsg("褰曞儚涓嬭浇璇锋眰瓒呮椂");
             downloadResult.setCode(ErrorCode.ERROR100.getCode());
             downloadResult.setMsg("褰曞儚涓嬭浇璇锋眰瓒呮椂");
             hookCallBack.call(downloadResult);
@@ -838,16 +842,12 @@
                 mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream());
                 streamSession.remove(deviceId, channelId, ssrcInfo.getStream());
             }
-            // 鍥炲涔嬪墠鎵�鏈夌殑鐐规挱璇锋眰
-            hookCallBack.call(downloadResult);
         }, userSetting.getPlayTimeout());
 
         SipSubscribe.Event errorEvent = event -> {
             dynamicTask.stop(downLoadTimeOutTaskKey);
             downloadResult.setCode(ErrorCode.ERROR100.getCode());
             downloadResult.setMsg(String.format("褰曞儚涓嬭浇澶辫触锛� 閿欒鐮侊細 %s, %s", event.statusCode, event.msg));
-            wvpResult.setCode(ErrorCode.ERROR100.getCode());
-            wvpResult.setMsg(String.format("褰曞儚涓嬭浇澶辫触锛� 閿欒鐮侊細 %s, %s", event.statusCode, event.msg));
             downloadResult.setEvent(event);
             hookCallBack.call(downloadResult);
             streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
@@ -862,11 +862,9 @@
                         streamInfo.setStartTime(startTime);
                         streamInfo.setEndTime(endTime);
                         redisCatchStorage.startDownload(streamInfo, inviteStreamInfo.getCallId());
-                        wvpResult.setCode(ErrorCode.SUCCESS.getCode());
-                        wvpResult.setMsg(ErrorCode.SUCCESS.getMsg());
-                        wvpResult.setData(streamInfo);
                         downloadResult.setCode(ErrorCode.SUCCESS.getCode());
                         downloadResult.setMsg(ErrorCode.SUCCESS.getMsg());
+                        downloadResult.setData(streamInfo);
                         downloadResult.setMediaServerItem(inviteStreamInfo.getMediaServerItem());
                         downloadResult.setResponse(inviteStreamInfo.getResponse());
                         hookCallBack.call(downloadResult);
@@ -878,7 +876,6 @@
             eventResult.msg = "鍛戒护鍙戦�佸け璐�";
             errorEvent.response(eventResult);
         }
-        return result;
     }
 
     @Override
@@ -898,7 +895,10 @@
             }
             if (mediaServerItem.getRecordAssistPort() > 0) {
                 JSONObject jsonObject = assistRESTfulUtils.fileDuration(mediaServerItem, streamInfo.getApp(), streamInfo.getStream(), null);
-                if (jsonObject != null && jsonObject.getInteger("code") == 0) {
+                if (jsonObject == null) {
+                    throw new ControllerException(ErrorCode.ERROR100.getCode(), "杩炴帴Assist鏈嶅姟澶辫触");
+                }
+                if (jsonObject.getInteger("code") == 0) {
                     long duration = jsonObject.getLong("data");
 
                     if (duration == 0) {
@@ -996,7 +996,7 @@
             logger.warn("寮�鍚闊冲箍鎾殑鏃跺�欐湭鎵惧埌閫氶亾锛� {}", channelId);
             return null;
         }
-        MediaServerItem mediaServerItem = mediaServerService.getMediaServerForMinimumLoad();
+        MediaServerItem mediaServerItem = mediaServerService.getMediaServerForMinimumLoad(null);
         String app = "broadcast";
         // TODO 浠巗ip user agent涓垽鏂槸浠�涔堝搧鐗岃澶囷紝澶у崕榛樿浣跨敤talk妯″紡锛屽叾浠栦娇鐢╞roadcast妯″紡
 //        String app = "talk";
@@ -1005,7 +1005,7 @@
         AudioBroadcastResult audioBroadcastResult = new AudioBroadcastResult();
         audioBroadcastResult.setApp(app);
         audioBroadcastResult.setStream(stream);
-        audioBroadcastResult.setStreamInfo(mediaService.getStreamInfoByAppAndStream(mediaServerItem, app, stream, null, null, null,false));
+        audioBroadcastResult.setStreamInfo(new StreamContent(mediaService.getStreamInfoByAppAndStream(mediaServerItem, app, stream, null, null, null,false)));
         audioBroadcastResult.setCodec("G.711");
         return audioBroadcastResult;
     }
@@ -1034,8 +1034,7 @@
                     event.call("璇煶骞挎挱宸茬粡寮�鍚�");
                     return;
                 } else {
-                    audioBroadcastManager.del(deviceChannel.getDeviceId(), channelId);
-                    redisCatchStorage.deleteSendRTPServer(device.getDeviceId(), channelId, sendRtpItem.getCallId(), sendRtpItem.getStreamId());
+                    stopAudioBroadcast(device.getDeviceId(), channelId);
                 }
             }
         }
@@ -1044,7 +1043,7 @@
         cmder.audioBroadcastCmd(device, channelId, eventResultForOk -> {
             // 鍙戦�佹垚鍔�
             AudioBroadcastCatch audioBroadcastCatch = new AudioBroadcastCatch(device.getDeviceId(), channelId, AudioBroadcastCatchStatus.Ready);
-            audioBroadcastManager.add(audioBroadcastCatch);
+            audioBroadcastManager.update(audioBroadcastCatch);
         }, eventResultForError -> {
             // 鍙戦�佸け璐�
             logger.error("璇煶骞挎挱鍙戦�佸け璐ワ細 {}:{}", channelId, eventResultForError.msg);
@@ -1057,28 +1056,37 @@
 
     @Override
     public void stopAudioBroadcast(String deviceId, String channelId) {
-        AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(deviceId, channelId);
-        if (audioBroadcastCatch != null) {
+        List<AudioBroadcastCatch> audioBroadcastCatchList = new ArrayList<>();
+        if (channelId == null) {
+            audioBroadcastCatchList.addAll(audioBroadcastManager.get(deviceId));
+        }else {
+            audioBroadcastCatchList.add(audioBroadcastManager.get(deviceId, channelId));
+        }
+        if (audioBroadcastCatchList.size() > 0) {
+            for (AudioBroadcastCatch audioBroadcastCatch : audioBroadcastCatchList) {
+                Device device = deviceService.getDevice(deviceId);
+                if (device == null || audioBroadcastCatch == null ) {
+                    return;
+                }
+                SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(deviceId, audioBroadcastCatch.getChannelId(), null, null);
+                if (sendRtpItem != null) {
+                    redisCatchStorage.deleteSendRTPServer(deviceId, sendRtpItem.getChannelId(), null, null);
+                    MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
+                    Map<String, Object> param = new HashMap<>();
+                    param.put("vhost", "__defaultVhost__");
+                    param.put("app", sendRtpItem.getApp());
+                    param.put("stream", sendRtpItem.getStreamId());
+                    zlmresTfulUtils.stopSendRtp(mediaInfo, param);
+                    try {
+                        cmder.streamByeCmd(device, sendRtpItem.getChannelId(), audioBroadcastCatch.getSipTransactionInfo(), null);
+                    } catch (InvalidArgumentException | ParseException | SipException |
+                             SsrcTransactionNotFoundException e) {
+                        logger.error("[娑堟伅鍙戦�佸け璐 鍙戦�佽闊冲枈璇滲YE澶辫触");
+                    }
+                }
 
-            Device device = deviceService.getDevice(deviceId);
-            if (device == null) {
-                return;
+                audioBroadcastManager.del(deviceId, channelId);
             }
-//            if (audioBroadcastCatch.getStatus() == AudioBroadcastCatchStatus.Ok) {
-//                cmder.streamByeCmd(device, audioBroadcastCatch.getChannelId(), null, audioBroadcastCatch.getSipTransactionInfo().getCallId());
-//            }
-            SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(deviceId, audioBroadcastCatch.getChannelId(), null, null);
-            if (sendRtpItem != null) {
-                redisCatchStorage.deleteSendRTPServer(deviceId, sendRtpItem.getChannelId(), null, null);
-                MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
-                Map<String, Object> param = new HashMap<>();
-                param.put("vhost", "__defaultVhost__");
-                param.put("app", sendRtpItem.getApp());
-                param.put("stream", sendRtpItem.getStreamId());
-                zlmresTfulUtils.stopSendRtp(mediaInfo, param);
-            }
-
-            audioBroadcastManager.del(deviceId, channelId);
         }
     }
 
@@ -1181,4 +1189,100 @@
         Device device = storager.queryVideoDevice(streamInfo.getDeviceID());
         cmder.playResumeCmd(device, streamInfo);
     }
+
+    @Override
+    public void startPushStream(SendRtpItem sendRtpItem, SIPResponse sipResponse, ParentPlatform platform, CallIdHeader callIdHeader) {
+
+        // 寮�濮嬪彂娴�
+        // 鍙栨秷璁剧疆鐨勮秴鏃朵换鍔�
+//			String channelId = request.getCallIdHeader().getCallId();
+
+        String is_Udp = sendRtpItem.isTcp() ? "0" : "1";
+        MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
+        logger.info("鏀跺埌ACK锛宺tp/{}寮�濮嬪悜涓婄骇鎺ㄦ祦, 鐩爣={}:{}锛孲SRC={}, RTCP={}", sendRtpItem.getStreamId(),
+                sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc(), sendRtpItem.isRtcp());
+        Map<String, Object> param = new HashMap<>(12);
+        param.put("vhost","__defaultVhost__");
+        param.put("app",sendRtpItem.getApp());
+        param.put("stream",sendRtpItem.getStreamId());
+        param.put("ssrc", sendRtpItem.getSsrc());
+        param.put("src_port", sendRtpItem.getLocalPort());
+        param.put("pt", sendRtpItem.getPt());
+        param.put("use_ps", sendRtpItem.isUsePs() ? "1" : "0");
+        param.put("only_audio", sendRtpItem.isOnlyAudio() ? "1" : "0");
+        param.put("is_udp", is_Udp);
+        if (!sendRtpItem.isTcp()) {
+            // udp妯″紡涓嬪紑鍚痳tcp淇濇椿
+            param.put("udp_rtcp_timeout", sendRtpItem.isRtcp()? "1":"0");
+        }
+
+        if (mediaInfo == null) {
+            RequestPushStreamMsg requestPushStreamMsg = RequestPushStreamMsg.getInstance(
+                    sendRtpItem.getMediaServerId(), sendRtpItem.getApp(), sendRtpItem.getStreamId(),
+                    sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc(), sendRtpItem.isTcp(),
+                    sendRtpItem.getLocalPort(), sendRtpItem.getPt(), sendRtpItem.isUsePs(), sendRtpItem.isOnlyAudio());
+            redisGbPlayMsgListener.sendMsgForStartSendRtpStream(sendRtpItem.getServerId(), requestPushStreamMsg, json -> {
+                startSendRtpStreamHand(sendRtpItem, platform, json, param, callIdHeader);
+            });
+        } else {
+            // 濡傛灉鏄潪涓ユ牸妯″紡锛岄渶瑕佸叧闂鍙e崰鐢�
+            JSONObject startSendRtpStreamResult = null;
+            if (sendRtpItem.getLocalPort() != 0) {
+                HookSubscribeForRtpServerTimeout hookSubscribeForRtpServerTimeout = HookSubscribeFactory.on_rtp_server_timeout(sendRtpItem.getSsrc(), null, mediaInfo.getId());
+                hookSubscribe.removeSubscribe(hookSubscribeForRtpServerTimeout);
+                if (zlmrtpServerFactory.releasePort(mediaInfo, sendRtpItem.getSsrc())) {
+                    if (sendRtpItem.isTcpActive()) {
+                        startSendRtpStreamResult = zlmrtpServerFactory.startSendRtpPassive(mediaInfo, param);
+                    }else {
+                        param.put("dst_url", sendRtpItem.getIp());
+                        param.put("dst_port", sendRtpItem.getPort());
+                        startSendRtpStreamResult = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
+                    }
+                }
+            }else {
+                if (sendRtpItem.isTcpActive()) {
+                    startSendRtpStreamResult = zlmrtpServerFactory.startSendRtpPassive(mediaInfo, param);
+                }else {
+                    param.put("dst_url", sendRtpItem.getIp());
+                    param.put("dst_port", sendRtpItem.getPort());
+                    startSendRtpStreamResult = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
+                }
+            }
+            if (startSendRtpStreamResult != null) {
+                startSendRtpStreamHand(sendRtpItem, platform, startSendRtpStreamResult, param, callIdHeader);
+            }
+        }
+    }
+
+    @Override
+    public void startSendRtpStreamHand(SendRtpItem sendRtpItem, ParentPlatform parentPlatform,
+                                       JSONObject jsonObject, Map<String, Object> param, CallIdHeader callIdHeader) {
+        if (jsonObject == null) {
+            logger.error("RTP鎺ㄦ祦澶辫触: 璇锋鏌LM鏈嶅姟");
+        } else if (jsonObject.getInteger("code") == 0) {
+            logger.info("璋冪敤ZLM鎺ㄦ祦鎺ュ彛, 缁撴灉锛� {}",  jsonObject);
+            logger.info("RTP鎺ㄦ祦鎴愬姛[ {}/{} ]锛寋}->{}:{}, " ,param.get("app"), param.get("stream"), jsonObject.getString("local_port"), param.get("dst_url"), param.get("dst_port"));
+        } else {
+            logger.error("RTP鎺ㄦ祦澶辫触: {}, 鍙傛暟锛歿}",jsonObject.getString("msg"), JSON.toJSONString(param));
+            if (sendRtpItem.isOnlyAudio()) {
+                Device device = deviceService.getDevice(sendRtpItem.getDeviceId());
+                AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
+                if (audioBroadcastCatch != null) {
+                    try {
+                        cmder.streamByeCmd(device, sendRtpItem.getChannelId(), audioBroadcastCatch.getSipTransactionInfo(), null);
+                    } catch (SipException | ParseException | InvalidArgumentException |
+                             SsrcTransactionNotFoundException e) {
+                        logger.error("[鍛戒护鍙戦�佸け璐 鍋滄璇煶瀵硅: {}", e.getMessage());
+                    }
+                }
+            }else {
+                // 鍚戜笂绾у钩鍙�
+                try {
+                    commanderForPlatform.streamByeCmd(parentPlatform, callIdHeader.getCallId());
+                } catch (SipException | InvalidArgumentException | ParseException e) {
+                    logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 鍙戦�丅YE: {}", e.getMessage());
+                }
+            }
+        }
+    }
 }

--
Gitblit v1.8.0