From c930d2d79dbf8f29a09aac88f32596d54b200d9f Mon Sep 17 00:00:00 2001
From: shikong <919411476@qq.com>
Date: Sun, 8 Oct 2023 10:46:57 +0800
Subject: [PATCH] =?UTF-8?q?=E5=B0=9D=E8=AF=95=E4=BD=BF=E7=94=A8=20zlmStrea?=
 =?UTF-8?q?mProxy?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../mocking/config/sip/ZlmHookConfig.java     |  1 +
 .../service/device/DeviceProxyService.java    | 47 ++++++++++++++-----
 2 files changed, 36 insertions(+), 12 deletions(-)

diff --git a/gb28181-mocking-service/src/main/java/cn/skcks/docking/gb28181/mocking/config/sip/ZlmHookConfig.java b/gb28181-mocking-service/src/main/java/cn/skcks/docking/gb28181/mocking/config/sip/ZlmHookConfig.java
index 003b7e3..75547a0 100644
--- a/gb28181-mocking-service/src/main/java/cn/skcks/docking/gb28181/mocking/config/sip/ZlmHookConfig.java
+++ b/gb28181-mocking-service/src/main/java/cn/skcks/docking/gb28181/mocking/config/sip/ZlmHookConfig.java
@@ -13,4 +13,5 @@ public class ZlmHookConfig {
     private String hook;
     private Duration delay = Duration.ofMillis(100);
     private Duration fixed = Duration.ofSeconds(30);
+    private Boolean proxy = true;
 }
diff --git a/gb28181-mocking-service/src/main/java/cn/skcks/docking/gb28181/mocking/service/device/DeviceProxyService.java b/gb28181-mocking-service/src/main/java/cn/skcks/docking/gb28181/mocking/service/device/DeviceProxyService.java
index 44658b4..df234f0 100644
--- a/gb28181-mocking-service/src/main/java/cn/skcks/docking/gb28181/mocking/service/device/DeviceProxyService.java
+++ b/gb28181-mocking-service/src/main/java/cn/skcks/docking/gb28181/mocking/service/device/DeviceProxyService.java
@@ -11,6 +11,7 @@ import cn.skcks.docking.gb28181.core.sip.gb28181.sdp.GB28181Description;
 import cn.skcks.docking.gb28181.core.sip.message.subscribe.GenericSubscribe;
 import cn.skcks.docking.gb28181.core.sip.utils.SipUtil;
 import cn.skcks.docking.gb28181.media.config.ZlmMediaConfig;
+import cn.skcks.docking.gb28181.media.dto.proxy.AddStreamProxy;
 import cn.skcks.docking.gb28181.media.dto.rtp.StartSendRtp;
 import cn.skcks.docking.gb28181.media.dto.rtp.StartSendRtpResp;
 import cn.skcks.docking.gb28181.media.proxy.ZlmMediaService;
@@ -107,12 +108,23 @@ public class DeviceProxyService {
                 zlmStreamChangeHookService.getUnregistHandler().put(callId,()->{
                     sendBye(request,device,key);
                 });
-                FfmpegExecuteResultHandler executeResultHandler = mediaStatus(request, device, key);
-                String zlmRtpUrl = "rtmp://" + zlmMediaConfig.getIp() + ":" + zlmRtmpConfig.getPort() + "/live/" + callId;
-                Executor executor = pushRtpTask(fromUrl, zlmRtpUrl, time + 60, executeResultHandler);
-                scheduledExecutorService.schedule(subscriber::onComplete, time + 60, TimeUnit.SECONDS);
-                callbackTask.put(device.getDeviceCode(), executor);
-                executeResultHandler.waitFor();
+                if(zlmHookConfig.getProxy()){
+                    AddStreamProxy addStreamProxy = AddStreamProxy.builder()
+                            .app("live")
+                            .stream(callId)
+                            .url(fromUrl)
+                            .timeoutSec(30.0)
+                            .build();
+                    zlmMediaService.addStreamProxy(addStreamProxy);
+                    scheduledExecutorService.schedule(subscriber::onComplete, time + 60, TimeUnit.SECONDS);
+                } else {
+                    FfmpegExecuteResultHandler executeResultHandler = mediaStatus(request, device, key);
+                    String zlmRtpUrl = "rtmp://" + zlmMediaConfig.getIp() + ":" + zlmRtmpConfig.getPort() + "/live/" + callId;
+                    Executor executor = pushRtpTask(fromUrl, zlmRtpUrl, time + 60, executeResultHandler);
+                    scheduledExecutorService.schedule(subscriber::onComplete, time + 60, TimeUnit.SECONDS);
+                    callbackTask.put(device.getDeviceCode(), executor);
+                    executeResultHandler.waitFor();
+                }
             } catch (Exception e) {
                 throw new RuntimeException(e);
             }
@@ -147,12 +159,23 @@ public class DeviceProxyService {
                 zlmStreamChangeHookService.getUnregistHandler().put(callId,()->{
                     sendBye(request,device,key);
                 });
-                FfmpegExecuteResultHandler executeResultHandler = mediaStatus(request, device, key);
-                String zlmRtpUrl = "rtmp://" + zlmMediaConfig.getIp() + ":" + zlmRtmpConfig.getPort() + "/live/" + callId;
-                Executor executor = pushDownload2RtpTask(fromUrl, zlmRtpUrl, time + 60, executeResultHandler);
-                scheduledExecutorService.schedule(subscriber::onComplete, time + 60, TimeUnit.SECONDS);
-                downloadTask.put(device.getDeviceCode(), executor);
-                executeResultHandler.waitFor();
+                if(zlmHookConfig.getProxy()){
+                    AddStreamProxy addStreamProxy = AddStreamProxy.builder()
+                            .app("live")
+                            .stream(callId)
+                            .url(fromUrl)
+                            .timeoutSec(30.0)
+                            .build();
+                    zlmMediaService.addStreamProxy(addStreamProxy);
+                    scheduledExecutorService.schedule(subscriber::onComplete, time + 60, TimeUnit.SECONDS);
+                } else {
+                    FfmpegExecuteResultHandler executeResultHandler = mediaStatus(request, device, key);
+                    String zlmRtpUrl = "rtmp://" + zlmMediaConfig.getIp() + ":" + zlmRtmpConfig.getPort() + "/live/" + callId;
+                    Executor executor = pushDownload2RtpTask(fromUrl, zlmRtpUrl, time + 60, executeResultHandler);
+                    scheduledExecutorService.schedule(subscriber::onComplete, time + 60, TimeUnit.SECONDS);
+                    downloadTask.put(device.getDeviceCode(), executor);
+                    executeResultHandler.waitFor();
+                }
             } catch (Exception e) {
                 throw new RuntimeException(e);
             }