From 663130df4556c35b8b390a74df571af8185d974d Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期日, 19 二月 2023 12:46:53 +0800
Subject: [PATCH] 完善支持语音对讲talk

---
 src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java |  459 ++++++++++++++++++++++++++++++++------------------------
 1 files changed, 263 insertions(+), 196 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 56efa72..1708f6e 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
@@ -41,7 +41,7 @@
 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 com.genersoft.iot.vmp.vmanager.gb28181.play.bean.AudioEvent;
 import gov.nist.javax.sip.message.SIPResponse;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -134,8 +134,8 @@
 
     @Override
     public void play(MediaServerItem mediaServerItem, String deviceId, String channelId,
-                                 ZlmHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent,
-                                 Runnable timeoutCallback) {
+                     ZlmHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent,
+                     Runnable timeoutCallback) {
         if (mediaServerItem == null) {
             throw new ControllerException(ErrorCode.ERROR100.getCode(), "鏈壘鍒板彲鐢ㄧ殑zlm");
         }
@@ -243,193 +243,147 @@
         }
     }
 
-    @Override
-    public void talk(MediaServerItem mediaServerItem, Device device, String channelId,
-                     ZlmHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent,
-                     Runnable timeoutCallback) {
-        String streamId = null;
-        if (mediaServerItem.isRtpEnable()) {
-            streamId = String.format("%s_%s", device.getDeviceId(), channelId);
+    private void talk(MediaServerItem mediaServerItem, Device device, String channelId, String stream,
+                      ZlmHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent,
+                      Runnable timeoutCallback, AudioEvent audioEvent) {
+
+        String playSsrc = mediaServerItem.getSsrcConfig().getPlaySsrc();
+        if (playSsrc == null) {
+            audioEvent.call("ssrc宸茬粡鐢ㄥ敖");
+            return;
         }
-        SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, device.isSsrcCheck(), false);
-        logger.info("[瀵硅寮�濮媇 deviceId: {}, channelId: {},鏀舵祦绔彛锛� {}, 鏀舵祦妯″紡锛歿}, SSRC: {}, SSRC鏍¢獙锛歿}", device.getDeviceId(), channelId, ssrcInfo.getPort(), device.getStreamMode(), ssrcInfo.getSsrc(), device.isSsrcCheck());
+        SendRtpItem sendRtpItem = new SendRtpItem();
+        sendRtpItem.setApp("talk");
+        sendRtpItem.setStream(stream);
+        sendRtpItem.setSsrc(playSsrc);
+        sendRtpItem.setDeviceId(device.getDeviceId());
+        sendRtpItem.setPlatformId(device.getDeviceId());
+        sendRtpItem.setChannelId(channelId);
+        sendRtpItem.setRtcp(false);
+        sendRtpItem.setMediaServerId(mediaServerItem.getId());
+        sendRtpItem.setOnlyAudio(true);
+        sendRtpItem.setPlayType(InviteStreamType.TALK);
+        sendRtpItem.setPt(8);
+        sendRtpItem.setStatus(1);
+        sendRtpItem.setTcpActive(false);
+        sendRtpItem.setTcp(true);
+        sendRtpItem.setUsePs(false);
+        sendRtpItem.setReceiveStream(stream);
+
+
+        int port = zlmrtpServerFactory.keepPort(mediaServerItem, playSsrc);
+        //绔彛鑾峰彇澶辫触鐨剆srcInfo 娌℃湁蹇呰鍙戦�佺偣鎾寚浠�
+        if (port <= 0) {
+            logger.info("[璇煶瀵硅] 绔彛鍒嗛厤寮傚父锛宒eviceId={},channelId={}", device.getDeviceId(), channelId);
+            audioEvent.call("绔彛鍒嗛厤寮傚父");
+            return;
+        }
+        sendRtpItem.setLocalPort(port);
+        sendRtpItem.setPort(port);
+        logger.info("[璇煶瀵硅]寮�濮� deviceId: {}, channelId: {},鏀舵祦绔彛锛� {}, 鏀舵祦妯″紡锛歿}, SSRC: {}, SSRC鏍¢獙锛歿}", device.getDeviceId(), channelId, sendRtpItem.getLocalPort(), device.getStreamMode(), sendRtpItem.getSsrc(), false);
         // 瓒呮椂澶勭悊
         String timeOutTaskKey = UUID.randomUUID().toString();
-        SSRCInfo finalSsrcInfo = ssrcInfo;
-        System.out.println("璁剧疆瓒呮椂浠诲姟锛� " + timeOutTaskKey);
         dynamicTask.startDelay(timeOutTaskKey, () -> {
 
-            logger.info("[瀵硅瓒呮椂] 鏀舵祦瓒呮椂 deviceId: {}, channelId: {}锛岀鍙o細{}, SSRC: {}", device.getDeviceId(), channelId, finalSsrcInfo.getPort(), finalSsrcInfo.getSsrc());
+            logger.info("[璇煶瀵硅] 鏀舵祦瓒呮椂 deviceId: {}, channelId: {}锛岀鍙o細{}, SSRC: {}", device.getDeviceId(), channelId, sendRtpItem.getPort(), sendRtpItem.getSsrc());
             timeoutCallback.run();
             // 鐐规挱瓒呮椂鍥炲BYE 鍚屾椂閲婃斁ssrc浠ュ強姝ゆ鐐规挱鐨勮祫婧�
             try {
-                cmder.streamByeCmd(device, channelId, finalSsrcInfo.getStream(), null);
-            } catch (InvalidArgumentException | ParseException | SipException e) {
-                logger.error("[瀵硅瓒呮椂]锛� 鍙戦�丅YE澶辫触 {}", e.getMessage());
-            } catch (SsrcTransactionNotFoundException e) {
+                cmder.streamByeCmd(device, channelId, sendRtpItem.getStream(), null);
+            } catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) {
+                logger.error("[璇煶瀵硅]瓒呮椂锛� 鍙戦�丅YE澶辫触 {}", e.getMessage());
+            } finally {
                 timeoutCallback.run();
-                mediaServerService.releaseSsrc(mediaServerItem.getId(), finalSsrcInfo.getSsrc());
-                mediaServerService.closeRTPServer(mediaServerItem, finalSsrcInfo.getStream());
-                streamSession.remove(device.getDeviceId(), channelId, finalSsrcInfo.getStream());
+                mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpItem.getSsrc());
+                streamSession.remove(device.getDeviceId(), channelId, sendRtpItem.getStream());
             }
         }, userSetting.getPlayTimeout());
-        final String ssrc = ssrcInfo.getSsrc();
-        final String stream = ssrcInfo.getStream();
-        //绔彛鑾峰彇澶辫触鐨剆srcInfo 娌℃湁蹇呰鍙戦�佺偣鎾寚浠�
-        if (ssrcInfo.getPort() <= 0) {
-            logger.info("[瀵硅] 绔彛鍒嗛厤寮傚父锛宒eviceId={},channelId={},ssrcInfo={}", device.getDeviceId(), channelId, ssrcInfo);
-            return;
-        }
 
         String callId = SipUtils.getNewCallId();
-        boolean pushing = false;
+
+        zlmrtpServerFactory.releasePort(mediaServerItem, playSsrc);
+        Map<String, Object> param = new HashMap<>(12);
+        param.put("vhost","__defaultVhost__");
+        param.put("app", sendRtpItem.getApp());
+        param.put("stream", sendRtpItem.getStream());
+        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", sendRtpItem.isTcp() ? "0" : "1");
+        param.put("recv_stream_id", sendRtpItem.getReceiveStream());
+        param.put("close_delay_ms", userSetting.getPlayTimeout() * 1000);
+
+        zlmrtpServerFactory.startSendRtpPassive(mediaServerItem, param, jsonObject -> {
+            if (jsonObject == null || jsonObject.getInteger("code") != 0 ) {
+                mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpItem.getSsrc());
+                logger.info("[璇煶瀵硅]澶辫触 deviceId: {}, channelId: {}", device.getDeviceId(), channelId);
+                audioEvent.call("澶辫触, " + jsonObject.getString("msg"));
+                // 鏌ョ湅鏄惁宸茬粡寤虹珛浜嗛�氶亾锛屽瓨鍦ㄥ垯鍙戦�乥ye
+                stopTalk(device, channelId);
+            }
+        });
+
+
         // 鏌ョ湅璁惧鏄惁宸茬粡鍦ㄦ帹娴�
-//        MediaItem mediaItem = zlmrtpServerFactory.getMediaInfo(mediaServerItem, "rtp",ssrcInfo.getStream());
-//        if (mediaItem != null) {
-//            SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem,
-//                    mediaItem.getOriginSock().getPeer_ip(), mediaItem.getOriginSock().getPeer_port(), ssrcInfo.getSsrc(), device.getDeviceId(),
-//                    device.getDeviceId(), channelId,
-//                    false);
-//
-//            sendRtpItem.setTcpActive(false);
-//            sendRtpItem.setCallId(callId);
-//            sendRtpItem.setPlayType(InviteStreamType.TALK);
-//            sendRtpItem.setStatus(1);
-//            sendRtpItem.setIp(mediaItem.getOriginSock().getPeer_ip());
-//            sendRtpItem.setPort(mediaItem.getOriginSock().getPeer_port());
-//            sendRtpItem.setTcpActive(false);
-//            sendRtpItem.setStreamId(ssrcInfo.getStream());
-//            sendRtpItem.setApp("1000");
-//            sendRtpItem.setStreamId("1000");
-//            sendRtpItem.setSsrc(ssrc);
-//            sendRtpItem.setOnlyAudio(true);
-//            redisCatchStorage.updateSendRTPSever(sendRtpItem);
-//
-//            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("dst_url", sendRtpItem.getIp());
-//            param.put("dst_port", sendRtpItem.getPort());
-//            param.put("src_port", sendRtpItem.getLocalPort());
-//            param.put("pt", sendRtpItem.getPt());
-//            param.put("use_ps", sendRtpItem.isUsePs() ? "1" : "0");
-//            param.put("is_udp", sendRtpItem.isTcp() ? "0" : "1");
-//            param.put("only_audio", sendRtpItem.isOnlyAudio() ? "1" : "0");
-//            JSONObject jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaServerItem, param);
-//            System.out.println(2222);
-//            System.out.println(jsonObject);
-//        }else {
-            try {
-                cmder.talkStreamCmd(mediaServerItem, ssrcInfo, device, channelId, callId, (MediaServerItem mediaServerItemInuse, JSONObject response) -> {
-                    logger.info("[瀵硅] 娴佸凡鐢熸垚锛� 寮�濮嬫帹娴侊細 " + response.toJSONString());
-                    dynamicTask.stop(timeOutTaskKey);
-                    // TODO 鏆備笉鍋氬鐞�
-                }, (MediaServerItem mediaServerItemInuse, JSONObject json) -> {
-                    logger.info("[瀵硅] 璁惧寮�濮嬫帹娴侊細 " + json.toJSONString());
-                    dynamicTask.stop(timeOutTaskKey);
-                    // 鑾峰彇杩滅▼IP绔彛 浣滀负鍥炲璇煶娴佺殑鍦板潃
-                    String ip = json.getString("ip");
-                    Integer port = json.getInteger("port");
-                    logger.info("[璁惧寮�濮嬫帹娴乚{}/{}, 鏉ヨ嚜ip锛歿}, 绔彛锛歿}", device.getDeviceId(), channelId, ip, port);
-                    // 鏌ョ湅骞冲彴鎺ㄦ祦鏄惁灏辩华
-//                    Boolean ready = zlmrtpServerFactory.isStreamReady(mediaServerItemInuse, "talk", stream);
-//                    if (!ready) {
-//                        try {
-//                            cmder.streamByeCmd(device, channelId, finalSsrcInfo.getStream(), null);
-//                        } catch (InvalidArgumentException | ParseException | SipException e) {
-//                            logger.error("[瀵硅瓒呮椂]锛� 鍙戦�丅YE澶辫触 {}", e.getMessage());
-//                        } catch (SsrcTransactionNotFoundException e) {
-//                            timeoutCallback.run();
-//                            mediaServerService.releaseSsrc(mediaServerItem.getId(), finalSsrcInfo.getSsrc());
-//                            mediaServerService.closeRTPServer(mediaServerItem, finalSsrcInfo.getStream());
-//                            streamSession.remove(device.getDeviceId(), channelId, finalSsrcInfo.getStream());
-//                        }
-//                    }else {
-//                        try {
-//                            Thread.sleep(1000);
-//                        } catch (InterruptedException e) {
-//                            throw new RuntimeException(e);
-//                        }
-                        SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, ip, port, ssrcInfo.getSsrc(), device.getDeviceId(),
-                                device.getDeviceId(), channelId,
-                                false, false);
+        try {
+            cmder.talkStreamCmd(mediaServerItem, sendRtpItem, device, channelId, callId, (MediaServerItem mediaServerItemInuse, JSONObject response) -> {
+                logger.info("[璇煶瀵硅] 娴佸凡鐢熸垚锛� 寮�濮嬫帹娴侊細 " + response.toJSONString());
+                dynamicTask.stop(timeOutTaskKey);
+                // TODO 鏆備笉鍋氬鐞�
+            }, (MediaServerItem mediaServerItemInuse, JSONObject json) -> {
+                logger.info("[璇煶瀵硅] 璁惧寮�濮嬫帹娴侊細 " + json.toJSONString());
+                dynamicTask.stop(timeOutTaskKey);
 
+            }, (event) -> {
+                dynamicTask.stop(timeOutTaskKey);
 
-//                        if (sendRtpItem.getLocalPort() == 0) {
-//                            logger.warn("鏈嶅姟鍣ㄧ鍙h祫婧愪笉瓒�");
-//                            try {
-//                                cmder.streamByeCmd(device, channelId, finalSsrcInfo.getStream(), null);
-//                            } catch (InvalidArgumentException | ParseException | SipException e) {
-//                                logger.error("[瀵硅瓒呮椂]锛� 鍙戦�丅YE澶辫触 {}", e.getMessage());
-//                            } catch (SsrcTransactionNotFoundException e) {
-//                                timeoutCallback.run();
-//                                mediaServerService.releaseSsrc(mediaServerItem.getId(), finalSsrcInfo.getSsrc());
-//                                mediaServerService.closeRTPServer(mediaServerItem, finalSsrcInfo.getStream());
-//                                streamSession.remove(device.getDeviceId(), channelId, finalSsrcInfo.getStream());
-//                            }
-//                            return;
-//                        }
-                        sendRtpItem.setTcpActive(false);
-                        sendRtpItem.setCallId(callId);
-                        sendRtpItem.setPlayType(InviteStreamType.TALK);
-                        sendRtpItem.setStatus(1);
-                        sendRtpItem.setIp(ip);
-                        sendRtpItem.setPort(port);
-                        sendRtpItem.setTcpActive(false);
-                        sendRtpItem.setApp("1000");
-                        sendRtpItem.setStreamId("1000");
-                        sendRtpItem.setSsrc(ssrc);
-                        sendRtpItem.setOnlyAudio(true);
-                        sendRtpItem.setRtcp(false);
+                if (event.event instanceof ResponseEvent) {
+                    ResponseEvent responseEvent = (ResponseEvent) event.event;
+                    if (responseEvent.getResponse() instanceof SIPResponse) {
+                        SIPResponse response = (SIPResponse) responseEvent.getResponse();
+                        sendRtpItem.setFromTag(response.getFromTag());
+                        sendRtpItem.setToTag(response.getToTag());
+                        sendRtpItem.setCallId(response.getCallIdHeader().getCallId());
                         redisCatchStorage.updateSendRTPSever(sendRtpItem);
 
-                        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("dst_url", sendRtpItem.getIp());
-                        param.put("dst_port", sendRtpItem.getPort());
-                        param.put("src_port", sendRtpItem.getLocalPort());
-                        param.put("pt", sendRtpItem.getPt());
-                        param.put("use_ps", sendRtpItem.isUsePs() ? "1" : "0");
-                        param.put("is_udp", sendRtpItem.isTcp() ? "0" : "1");
-                        param.put("only_audio", sendRtpItem.isOnlyAudio() ? "1" : "0");
-                        JSONObject jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaServerItemInuse, param);
-                        System.out.println(11111);
-                        System.out.println(sendRtpItem.getIp() + ":" + sendRtpItem.getPort());
-//                        System.out.println(jsonObject);
-//                    }
+                        streamSession.put(device.getDeviceId(), channelId, response.getCallIdHeader().getCallId(),
+                                sendRtpItem.getStream(), sendRtpItem.getSsrc(), sendRtpItem.getMediaServerId(),
+                                response, VideoStreamSessionManager.SessionType.talk);
+                    } else {
+                        logger.error("[璇煶瀵硅]鏀跺埌鐨勬秷鎭敊璇紝response涓嶆槸SIPResponse");
+                    }
+                } else {
+                    logger.error("[璇煶瀵硅]鏀跺埌鐨勬秷鎭敊璇紝event涓嶆槸ResponseEvent");
+                }
 
-                }, (event) -> {
-
-                }, (event) -> {
-                    dynamicTask.stop(timeOutTaskKey);
-                    mediaServerService.closeRTPServer(mediaServerItem, finalSsrcInfo.getStream());
-                    // 閲婃斁ssrc
-                    mediaServerService.releaseSsrc(mediaServerItem.getId(), finalSsrcInfo.getSsrc());
-
-                    streamSession.remove(device.getDeviceId(), channelId, finalSsrcInfo.getStream());
-                    errorEvent.response(event);
-                });
-            } catch (InvalidArgumentException | SipException | ParseException e) {
-
-                logger.error("[鍛戒护鍙戦�佸け璐 瀵硅娑堟伅: {}", e.getMessage());
+            }, (event) -> {
                 dynamicTask.stop(timeOutTaskKey);
-                mediaServerService.closeRTPServer(mediaServerItem, finalSsrcInfo.getStream());
+                mediaServerService.closeRTPServer(mediaServerItem, sendRtpItem.getStream());
                 // 閲婃斁ssrc
-                mediaServerService.releaseSsrc(mediaServerItem.getId(), finalSsrcInfo.getSsrc());
+                mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpItem.getSsrc());
+                streamSession.remove(device.getDeviceId(), channelId, sendRtpItem.getStream());
+                errorEvent.response(event);
+            });
+        } catch (InvalidArgumentException | SipException | ParseException e) {
 
-                streamSession.remove(device.getDeviceId(), channelId, finalSsrcInfo.getStream());
-                SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(new CmdSendFailEvent(null));
-                eventResult.msg = "鍛戒护鍙戦�佸け璐�";
-                errorEvent.response(eventResult);
-            }
+            logger.error("[鍛戒护鍙戦�佸け璐 瀵硅娑堟伅: {}", e.getMessage());
+            dynamicTask.stop(timeOutTaskKey);
+            mediaServerService.closeRTPServer(mediaServerItem, sendRtpItem.getStream());
+            // 閲婃斁ssrc
+            mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpItem.getSsrc());
+
+            streamSession.remove(device.getDeviceId(), channelId, sendRtpItem.getStream());
+            SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(new CmdSendFailEvent(null));
+            eventResult.msg = "鍛戒护鍙戦�佸け璐�";
+            errorEvent.response(eventResult);
+        }
 //        }
 
     }
+
+
 
     @Override
     public void play(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId,
@@ -446,7 +400,8 @@
                 // 鐐规挱瓒呮椂鍥炲BYE 鍚屾椂閲婃斁ssrc浠ュ強姝ゆ鐐规挱鐨勮祫婧�
                 try {
                     cmder.streamByeCmd(device, channelId, ssrcInfo.getStream(), null);
-                } catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) {
+                } catch (InvalidArgumentException | ParseException | SipException |
+                         SsrcTransactionNotFoundException e) {
                     logger.error("[鐐规挱瓒呮椂]锛� 鍙戦�丅YE澶辫触 {}", e.getMessage());
                 } finally {
                     timeoutCallback.run(1, "鏀舵祦瓒呮椂");
@@ -483,7 +438,7 @@
                 onPublishHandlerForPlay(mediaServerItemInuse, response, device.getDeviceId(), channelId);
                 hookEvent.response(mediaServerItemInuse, response);
                 logger.info("[鐐规挱鎴愬姛] deviceId: {}, channelId: {}", device.getDeviceId(), channelId);
-                String streamUrl = String.format("http://127.0.0.1:%s/%s/%s.live.flv", mediaServerItemInuse.getHttpPort(), "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";
                 // 璇锋眰鎴浘
@@ -652,8 +607,8 @@
 
     @Override
     public void playBack(String deviceId, String channelId, String startTime,
-                                                          String endTime, InviteStreamCallback inviteStreamCallback,
-                                                          PlayBackCallback callback) {
+                         String endTime, InviteStreamCallback inviteStreamCallback,
+                         PlayBackCallback callback) {
         Device device = storager.queryVideoDevice(deviceId);
         if (device == null) {
             return;
@@ -666,9 +621,9 @@
 
     @Override
     public void playBack(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo,
-                                                          String deviceId, String channelId, String startTime,
-                                                          String endTime, InviteStreamCallback infoCallBack,
-                                                          PlayBackCallback playBackCallback) {
+                         String deviceId, String channelId, String startTime,
+                         String endTime, InviteStreamCallback infoCallBack,
+                         PlayBackCallback playBackCallback) {
         if (mediaServerItem == null || ssrcInfo == null) {
             return;
         }
@@ -790,7 +745,6 @@
             errorEvent.response(eventResult);
         }
     }
-
 
 
     @Override
@@ -977,7 +931,7 @@
                         cmder.streamByeCmd(device, ssrcTransaction.getChannelId(),
                                 ssrcTransaction.getStream(), null);
                     } catch (InvalidArgumentException | ParseException | SipException |
-                            SsrcTransactionNotFoundException e) {
+                             SsrcTransactionNotFoundException e) {
                         logger.error("[zlm绂荤嚎]涓烘鍦ㄤ娇鐢ㄦzlm鐨勮澶囷紝 鍙戦�丅YE澶辫触 {}", e.getMessage());
                     }
                 }
@@ -987,6 +941,7 @@
 
     @Override
     public AudioBroadcastResult audioBroadcast(Device device, String channelId) {
+        // TODO 蹇呴』澶氱鍙fā寮忔墠鏀寔璇煶鍠婅瘽楣よ闊冲璁�
         if (device == null || channelId == null) {
             return null;
         }
@@ -1005,13 +960,13 @@
         AudioBroadcastResult audioBroadcastResult = new AudioBroadcastResult();
         audioBroadcastResult.setApp(app);
         audioBroadcastResult.setStream(stream);
-        audioBroadcastResult.setStreamInfo(new StreamContent(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;
     }
 
     @Override
-    public void audioBroadcastCmd(Device device, String channelId, int timeout, AudioBroadcastEvent event) throws InvalidArgumentException, ParseException, SipException {
+    public void audioBroadcastCmd(Device device, String channelId, MediaServerItem mediaServerItem, int timeout, AudioEvent event) throws InvalidArgumentException, ParseException, SipException {
         if (device == null || channelId == null) {
             return;
         }
@@ -1027,8 +982,8 @@
             SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(device.getDeviceId(), channelId, null, null);
             if (sendRtpItem != null && sendRtpItem.isOnlyAudio()) {
                 // 鏌ヨ娴佹槸鍚﹀瓨鍦紝涓嶅瓨鍦ㄥ垯璁や负鏄紓甯哥姸鎬�
-                MediaServerItem mediaServerItem = mediaServerService.getOne(sendRtpItem.getMediaServerId());
-                Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, sendRtpItem.getApp(), sendRtpItem.getStreamId());
+                MediaServerItem mediaServer = mediaServerService.getOne(sendRtpItem.getMediaServerId());
+                Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServer, sendRtpItem.getApp(), sendRtpItem.getStream());
                 if (streamReady) {
                     logger.warn("璇煶骞挎挱宸茬粡寮�鍚細 {}", channelId);
                     event.call("璇煶骞挎挱宸茬粡寮�鍚�");
@@ -1038,11 +993,23 @@
                 }
             }
         }
+        SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(device.getDeviceId(), channelId, null, null);
+        if (sendRtpItem != null) {
+            MediaServerItem mediaServer = mediaServerService.getOne(sendRtpItem.getMediaServerId());
+            Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServer, "rtp", sendRtpItem.getReceiveStream());
+            if (streamReady) {
+                logger.warn("[璇煶瀵硅] 杩涜涓細 {}", channelId);
+                event.call("璇煶瀵硅杩涜涓�");
+                return;
+            } else {
+                stopTalk(device, channelId);
+            }
+        }
 
         // 鍙戦�侀�氱煡
         cmder.audioBroadcastCmd(device, channelId, eventResultForOk -> {
             // 鍙戦�佹垚鍔�
-            AudioBroadcastCatch audioBroadcastCatch = new AudioBroadcastCatch(device.getDeviceId(), channelId, AudioBroadcastCatchStatus.Ready);
+            AudioBroadcastCatch audioBroadcastCatch = new AudioBroadcastCatch(device.getDeviceId(), channelId, AudioBroadcastCatchStatus.Ready, mediaServerItem);
             audioBroadcastManager.update(audioBroadcastCatch);
         }, eventResultForError -> {
             // 鍙戦�佸け璐�
@@ -1053,19 +1020,18 @@
     }
 
 
-
     @Override
     public void stopAudioBroadcast(String deviceId, String channelId) {
         List<AudioBroadcastCatch> audioBroadcastCatchList = new ArrayList<>();
         if (channelId == null) {
             audioBroadcastCatchList.addAll(audioBroadcastManager.get(deviceId));
-        }else {
+        } 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 ) {
+                if (device == null || audioBroadcastCatch == null) {
                     return;
                 }
                 SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(deviceId, audioBroadcastCatch.getChannelId(), null, null);
@@ -1075,7 +1041,7 @@
                     Map<String, Object> param = new HashMap<>();
                     param.put("vhost", "__defaultVhost__");
                     param.put("app", sendRtpItem.getApp());
-                    param.put("stream", sendRtpItem.getStreamId());
+                    param.put("stream", sendRtpItem.getStream());
                     zlmresTfulUtils.stopSendRtp(mediaInfo, param);
                     try {
                         cmder.streamByeCmd(device, sendRtpItem.getChannelId(), audioBroadcastCatch.getSipTransactionInfo(), null);
@@ -1199,12 +1165,12 @@
 
         String is_Udp = sendRtpItem.isTcp() ? "0" : "1";
         MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
-        logger.info("鏀跺埌ACK锛宺tp/{}寮�濮嬪悜涓婄骇鎺ㄦ祦, 鐩爣={}:{}锛孲SRC={}, RTCP={}", sendRtpItem.getStreamId(),
+        logger.info("鏀跺埌ACK锛宺tp/{}寮�濮嬪悜涓婄骇鎺ㄦ祦, 鐩爣={}:{}锛孲SRC={}, RTCP={}", sendRtpItem.getStream(),
                 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("vhost", "__defaultVhost__");
+        param.put("app", sendRtpItem.getApp());
+        param.put("stream", sendRtpItem.getStream());
         param.put("ssrc", sendRtpItem.getSsrc());
         param.put("src_port", sendRtpItem.getLocalPort());
         param.put("pt", sendRtpItem.getPt());
@@ -1213,12 +1179,12 @@
         param.put("is_udp", is_Udp);
         if (!sendRtpItem.isTcp()) {
             // udp妯″紡涓嬪紑鍚痳tcp淇濇椿
-            param.put("udp_rtcp_timeout", sendRtpItem.isRtcp()? "1":"0");
+            param.put("udp_rtcp_timeout", sendRtpItem.isRtcp() ? "1" : "0");
         }
 
         if (mediaInfo == null) {
             RequestPushStreamMsg requestPushStreamMsg = RequestPushStreamMsg.getInstance(
-                    sendRtpItem.getMediaServerId(), sendRtpItem.getApp(), sendRtpItem.getStreamId(),
+                    sendRtpItem.getMediaServerId(), sendRtpItem.getApp(), sendRtpItem.getStream(),
                     sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc(), sendRtpItem.isTcp(),
                     sendRtpItem.getLocalPort(), sendRtpItem.getPt(), sendRtpItem.isUsePs(), sendRtpItem.isOnlyAudio());
             redisGbPlayMsgListener.sendMsgForStartSendRtpStream(sendRtpItem.getServerId(), requestPushStreamMsg, json -> {
@@ -1233,16 +1199,16 @@
                 if (zlmrtpServerFactory.releasePort(mediaInfo, sendRtpItem.getSsrc())) {
                     if (sendRtpItem.isTcpActive()) {
                         startSendRtpStreamResult = zlmrtpServerFactory.startSendRtpPassive(mediaInfo, param);
-                    }else {
+                    } else {
                         param.put("dst_url", sendRtpItem.getIp());
                         param.put("dst_port", sendRtpItem.getPort());
                         startSendRtpStreamResult = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
                     }
                 }
-            }else {
+            } else {
                 if (sendRtpItem.isTcpActive()) {
                     startSendRtpStreamResult = zlmrtpServerFactory.startSendRtpPassive(mediaInfo, param);
-                }else {
+                } else {
                     param.put("dst_url", sendRtpItem.getIp());
                     param.put("dst_port", sendRtpItem.getPort());
                     startSendRtpStreamResult = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
@@ -1260,10 +1226,10 @@
         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"));
+            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));
+            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());
@@ -1275,7 +1241,7 @@
                         logger.error("[鍛戒护鍙戦�佸け璐 鍋滄璇煶瀵硅: {}", e.getMessage());
                     }
                 }
-            }else {
+            } else {
                 // 鍚戜笂绾у钩鍙�
                 try {
                     commanderForPlatform.streamByeCmd(parentPlatform, callIdHeader.getCallId());
@@ -1285,4 +1251,105 @@
             }
         }
     }
+
+    @Override
+    public void talkCmd(Device device, String channelId, MediaServerItem mediaServerItem, String stream, AudioEvent event) {
+        if (device == null || channelId == null) {
+            return;
+        }
+        // TODO 蹇呴』澶氱鍙fā寮忔墠鏀寔璇煶鍠婅瘽楣よ闊冲璁�
+        logger.info("[璇煶瀵硅] device锛� {}, channel: {}", device.getDeviceId(), channelId);
+        DeviceChannel deviceChannel = storager.queryChannel(device.getDeviceId(), channelId);
+        if (deviceChannel == null) {
+            logger.warn("寮�鍚闊冲璁茬殑鏃跺�欐湭鎵惧埌閫氶亾锛� {}", channelId);
+            event.call("寮�鍚闊冲璁茬殑鏃跺�欐湭鎵惧埌閫氶亾");
+            return;
+        }
+        // 鏌ヨ閫氶亾浣跨敤鐘舵��
+        if (audioBroadcastManager.exit(device.getDeviceId(), channelId)) {
+            SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(device.getDeviceId(), channelId, null, null);
+            if (sendRtpItem != null && sendRtpItem.isOnlyAudio()) {
+                // 鏌ヨ娴佹槸鍚﹀瓨鍦紝涓嶅瓨鍦ㄥ垯璁や负鏄紓甯哥姸鎬�
+                MediaServerItem mediaServer = mediaServerService.getOne(sendRtpItem.getMediaServerId());
+                Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServer, sendRtpItem.getApp(), sendRtpItem.getStream());
+                if (streamReady) {
+                    logger.warn("[璇煶瀵硅] 姝e湪璇煶骞挎挱锛屾棤娉曞紑鍚闊抽�氳瘽锛� {}", channelId);
+                    event.call("姝e湪璇煶骞挎挱");
+                    return;
+                } else {
+                    stopAudioBroadcast(device.getDeviceId(), channelId);
+                }
+            }
+        }
+
+        SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(device.getDeviceId(), channelId, stream, null);
+        if (sendRtpItem != null) {
+            MediaServerItem mediaServer = mediaServerService.getOne(sendRtpItem.getMediaServerId());
+            Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServer, "rtp", sendRtpItem.getReceiveStream());
+            if (streamReady) {
+                logger.warn("[璇煶瀵硅] 杩涜涓細 {}", channelId);
+                event.call("璇煶瀵硅杩涜涓�");
+                return;
+            } else {
+                stopTalk(device, channelId);
+            }
+        }
+
+        talk(mediaServerItem, device, channelId, stream, (MediaServerItem mediaServerItem1, JSONObject response) -> {
+            logger.info("[璇煶瀵硅] 鏀跺埌璁惧鍙戞潵鐨勬祦");
+        }, eventResult -> {
+            logger.warn("[璇煶瀵硅] 澶辫触锛寋}/{}, 閿欒鐮� {} {}", device.getDeviceId(), channelId, eventResult.statusCode, eventResult.msg);
+            event.call("澶辫触锛岄敊璇爜 " + eventResult.statusCode + ", " + eventResult.msg);
+        }, () -> {
+            logger.warn("[璇煶瀵硅] 澶辫触锛寋}/{} 瓒呮椂", device.getDeviceId(), channelId);
+            event.call("澶辫触锛岃秴鏃� ");
+            stopTalk(device, channelId);
+        }, errorMsg -> {
+            logger.warn("[璇煶瀵硅] 澶辫触锛寋}/{} {}", device.getDeviceId(), channelId, errorMsg);
+            event.call(errorMsg);
+            stopTalk(device, channelId);
+        });
+    }
+
+    private void stopTalk(Device device, String channelId) {
+        stopTalk(device, channelId, null);
+    }
+
+    @Override
+    public void stopTalk(Device device, String channelId, Boolean streamIsReady) {
+        logger.info("[璇煶瀵硅] 鍋滄锛� {}/{}", device.getDeviceId(), channelId);
+        SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(device.getDeviceId(), channelId, null, null);
+        if (sendRtpItem == null) {
+            logger.info("[璇煶瀵硅] 鍋滄澶辫触锛� 鏈壘鍒板彂閫佷俊鎭紝鍙兘宸茬粡鍋滄");
+            return;
+        }
+        // 鍋滄鍚戣澶囨帹娴�
+        String mediaServerId = sendRtpItem.getMediaServerId();
+        if (mediaServerId == null) {
+            return;
+        }
+
+        MediaServerItem mediaServer = mediaServerService.getOne(mediaServerId);
+
+        if (streamIsReady == null || streamIsReady) {
+            Map<String, Object> param = new HashMap<>();
+            param.put("vhost", "__defaultVhost__");
+            param.put("app", sendRtpItem.getApp());
+            param.put("stream", sendRtpItem.getStream());
+            param.put("ssrc", sendRtpItem.getSsrc());
+            zlmrtpServerFactory.stopSendRtpStream(mediaServer, param);
+        }
+
+        mediaServer.getSsrcConfig().releaseSsrc(sendRtpItem.getSsrc());
+
+        SsrcTransaction ssrcTransaction = streamSession.getSsrcTransaction(device.getDeviceId(), channelId, null, sendRtpItem.getStream());
+        if (ssrcTransaction != null) {
+            try {
+                cmder.streamByeCmd(device, channelId, sendRtpItem.getStream(), null);
+            } catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException  e) {
+                logger.info("[璇煶瀵硅] 鍋滄娑堟伅鍙戦�佸け璐ワ紝鍙兘宸茬粡鍋滄");
+            }
+        }
+        redisCatchStorage.deleteSendRTPServer(device.getDeviceId(), channelId,null, null);
+    }
 }

--
Gitblit v1.8.0