下载 调整

gb28181-wvp-proxy-service/src/main/java/cn/skcks/docking/gb28181/wvp/config/ZlmRtspConfig.java

@@ -0,0 +1,12 @@
+package cn.skcks.docking.gb28181.wvp.config;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+
+@Data
+@Configuration
+@ConfigurationProperties(prefix = "media.rtsp")
+public class ZlmRtspConfig {
+    private String url;
+}

gb28181-wvp-proxy-service/src/main/java/cn/skcks/docking/gb28181/wvp/service/ffmpeg/FfmpegSupportService.java

@@ -21,7 +21,7 @@ public class FfmpegSupportService {
     public Executor downloadToStream(String input, long time, TimeUnit unit, ExecuteStreamHandler streamHandler, ExecuteResultHandler executeResultHandler) {
         FfmpegConfig.Rtp rtp = ffmpegConfig.getRtp();
         FfmpegConfig.Debug debug = ffmpegConfig.getDebug();
-        String inputParam = debug.getInput() ? rtp.getInput() : StringUtils.joinWith(" ", rtp.getInput(), input);
+        String inputParam = debug.getDownload() ? rtp.getDownload() : StringUtils.joinWith(" ", rtp.getDownload(), input);
         log.info("视频输入参数 {}", inputParam);
 
         String outputParam = debug.getOutput() ? rtp.getOutput() : StringUtils.joinWith(" ", rtp.getOutput(), "-");

gb28181-wvp-proxy-service/src/main/java/cn/skcks/docking/gb28181/wvp/service/gb28181/Gb28181DownloadService.java

@@ -5,6 +5,7 @@ import cn.hutool.core.date.DateUnit;
 import cn.hutool.core.date.DateUtil;
 import cn.hutool.core.date.LocalDateTimeUtil;
 import cn.hutool.core.io.IoUtil;
+import cn.hutool.core.util.IdUtil;
 import cn.skcks.docking.gb28181.common.json.JsonException;
 import cn.skcks.docking.gb28181.common.json.JsonResponse;
 import cn.skcks.docking.gb28181.core.sip.gb28181.constant.GB28181Constant;
@@ -15,6 +16,7 @@ import cn.skcks.docking.gb28181.core.sip.message.processor.MessageProcessor;
 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.rtp.CloseRtpServer;
 import cn.skcks.docking.gb28181.media.dto.rtp.GetRtpInfoResp;
 import cn.skcks.docking.gb28181.media.dto.rtp.OpenRtpServer;
 import cn.skcks.docking.gb28181.media.dto.rtp.OpenRtpServerResp;
@@ -23,17 +25,20 @@ import cn.skcks.docking.gb28181.media.proxy.ZlmMediaService;
 import cn.skcks.docking.gb28181.service.ssrc.SsrcService;
 import cn.skcks.docking.gb28181.wvp.config.ProxySipConfig;
 import cn.skcks.docking.gb28181.wvp.config.WvpProxyConfig;
+import cn.skcks.docking.gb28181.wvp.config.ZlmRtspConfig;
 import cn.skcks.docking.gb28181.wvp.orm.mybatis.dynamic.model.WvpProxyDevice;
 import cn.skcks.docking.gb28181.wvp.orm.mybatis.dynamic.model.WvpProxyDocking;
 import cn.skcks.docking.gb28181.wvp.service.device.DeviceService;
 import cn.skcks.docking.gb28181.wvp.service.docking.DockingService;
 import cn.skcks.docking.gb28181.wvp.service.video.VideoService;
 import cn.skcks.docking.gb28181.wvp.sip.request.SipRequestBuilder;
+import cn.skcks.docking.gb28181.wvp.sip.response.SipResponseBuilder;
 import cn.skcks.docking.gb28181.wvp.sip.sender.SipSender;
 import cn.skcks.docking.gb28181.wvp.sip.subscribe.SipSubscribe;
 import gov.nist.javax.sdp.MediaDescriptionImpl;
 import gov.nist.javax.sdp.fields.TimeField;
 import gov.nist.javax.sdp.fields.URIField;
+import gov.nist.javax.sip.message.SIPRequest;
 import gov.nist.javax.sip.message.SIPResponse;
 import jakarta.servlet.AsyncContext;
 import jakarta.servlet.http.HttpServletRequest;
@@ -56,7 +61,6 @@ import java.nio.charset.StandardCharsets;
 import java.text.MessageFormat;
 import java.time.ZoneId;
 import java.util.Date;
-import java.util.List;
 import java.util.Optional;
 import java.util.Vector;
 import java.util.concurrent.*;
@@ -74,8 +78,8 @@ public class Gb28181DownloadService {
     private final SipSender sender;
     private final SipSubscribe subscribe;
     private final VideoService videoService;
-
     private final WvpProxyConfig wvpProxyConfig;
+    private final ZlmRtspConfig zlmRtspConfig;
     private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
 
     public void header(HttpServletResponse response) {
@@ -128,16 +132,8 @@ public class Gb28181DownloadService {
         asyncContext.start(()->{
             HttpServletResponse asyncResponse = (HttpServletResponse)asyncContext.getResponse();
             try{
-                if(fileHeader){
-                    header(response, StringUtils.joinWith("_",
-                            deviceCode,
-                            DateUtil.format(startTime, DatePattern.PURE_DATETIME_FORMAT),
-                            DateUtil.format(endTime, DatePattern.PURE_DATETIME_FORMAT)));
-                } else {
-                    header(response);
-                }
-
                 download(deviceCode, startTime,endTime).whenComplete((url, e)->{
+                    writeFileHeader(response,deviceCode,startTime,endTime,fileHeader);
                     if(e != null){
                         writeErrorToResponse(asyncResponse, JsonResponse.error(e.getMessage()));
                     } else if(StringUtils.isBlank(url)){
@@ -156,15 +152,26 @@ public class Gb28181DownloadService {
         });
     }
 
+    private void writeFileHeader(HttpServletResponse response, String deviceCode, Date startTime, Date endTime, Boolean fileHeader){
+        if(fileHeader){
+            header(response, StringUtils.joinWith("_",
+                    deviceCode,
+                    DateUtil.format(startTime, DatePattern.PURE_DATETIME_FORMAT),
+                    DateUtil.format(endTime, DatePattern.PURE_DATETIME_FORMAT)));
+        } else {
+            header(response);
+        }
+    }
+
     @SneakyThrows
     public CompletableFuture<String> download(String deviceCode, Date startTime, Date endTime) {
-        List<WvpProxyDevice> deviceByDeviceCode = deviceService.getDeviceByDeviceCode(deviceCode);
+        Optional<WvpProxyDevice> deviceByDeviceCode = deviceService.getDeviceByDeviceCode(deviceCode);
         if (deviceByDeviceCode.isEmpty()) {
             String reason = MessageFormat.format("未能找到 设备编码 为 {0} 的设备", deviceCode);
             log.error("{}",reason);
             throw new JsonException(reason);
         } else {
-            WvpProxyDevice device = deviceByDeviceCode.get(0);
+            WvpProxyDevice device = deviceByDeviceCode.get();
             return download(device.getGbDeviceId(), device.getGbDeviceChannelId(), startTime, endTime);
         }
     }
@@ -192,7 +199,7 @@ public class Gb28181DownloadService {
         ZoneId zoneId = ZoneId.of(GB28181Constant.TIME_ZONE);
         long start =  LocalDateTimeUtil.of(startTime.toInstant(), zoneId).atZone(zoneId).toEpochSecond();
         long end =  LocalDateTimeUtil.of(endTime.toInstant(), zoneId).atZone(zoneId).toEpochSecond();
-        String streamId = MediaSdpHelper.getStreamId(gbDeviceId, channel, String.valueOf(start), String.valueOf(end));
+        String streamId = MediaSdpHelper.getStreamId(gbDeviceId, channel, String.valueOf(start), String.valueOf(end), IdUtil.getSnowflakeNextIdStr());
         int streamMode = proxySipConfig.getTransport().equalsIgnoreCase(ListeningPoint.UDP) ? 0 : 1;
         String ip = zlmMediaConfig.getIp();
         int port = openRtpServer(streamId, streamMode);
@@ -223,15 +230,15 @@ public class Gb28181DownloadService {
             CallIdHeader callId = provider.getNewCallId();
             String subscribeKey = GenericSubscribe.Helper.getKey(Request.INVITE, callId.getCallId());
             subscribe.getInviteSubscribe().addPublisher(subscribeKey);
-            Flow.Subscriber<SIPResponse> subscriber = inviteSubscriber(docking,device,subscribeKey, ssrc, streamId, result);
+            Flow.Subscriber<SIPResponse> subscriber = inviteSubscriber(docking,device,subscribeKey, ssrc, streamId, result, time + 60, TimeUnit.SECONDS);
             subscribe.getInviteSubscribe().addSubscribe(subscribeKey, subscriber);
-            scheduledExecutorService.schedule(subscriber::onComplete, time + 60, TimeUnit.SECONDS);
             return SipRequestBuilder.createInviteRequest(ip, port, docking, device.getGbDeviceChannelId(), description.toString(), SipUtil.generateViaTag(), SipUtil.generateFromTag(), null, ssrc, callId);
         };
     }
 
-    public Flow.Subscriber<SIPResponse> inviteSubscriber(WvpProxyDocking docking, WvpProxyDevice device, String subscribeKey,String ssrc,String streamId,CompletableFuture<String> result){
+    public Flow.Subscriber<SIPResponse> inviteSubscriber(WvpProxyDocking docking, WvpProxyDevice device, String subscribeKey,String ssrc,String streamId,CompletableFuture<String> result, long time, TimeUnit unit){
-        return new Flow.Subscriber<>() {
+        ScheduledFuture<?>[] schedule = new ScheduledFuture<?>[1];
+        Flow.Subscriber<SIPResponse> subscriber = new Flow.Subscriber<>() {
             private Flow.Subscription subscription;
 
             @Override
@@ -255,11 +262,15 @@ public class Gb28181DownloadService {
                         String fromTag = item.getFromTag();
                         String toTag = item.getToTag();
                         String callId = item.getCallId().getCallId();
+                        String key = GenericSubscribe.Helper.getKey(Request.BYE, callId);
+                        subscribe.getByeSubscribe().addPublisher(key);
+                        subscribe.getByeSubscribe().addSubscribe(key, byeSubscriber(key,streamId, time, unit));
                         return SipRequestBuilder.createAckRequest(Response.OK, ip, port, docking, device.getGbDeviceChannelId(), fromTag, toTag, callId);
                     }));
                     result.complete(videoUrl(streamId));
                 } else {
                     log.info("订阅 {} {} 连接流媒体服务时出现异常, 终止订阅", MessageProcessor.Method.INVITE, subscribeKey);
+                    zlmMediaService.closeRtpServer(new CloseRtpServer(streamId));
                     result.complete("");
                     ssrcService.releaseSsrc(zlmMediaConfig.getId(), ssrc);
                     onComplete();
@@ -274,7 +285,46 @@ public class Gb28181DownloadService {
             @Override
             public void onComplete() {
                 subscribe.getInviteSubscribe().delPublisher(subscribeKey);
+                schedule[0].cancel(true);
             }
         };
+        schedule[0] = scheduledExecutorService.schedule(subscriber::onComplete, time, unit);
+        return subscriber;
+    }
+
+    public Flow.Subscriber<SIPRequest> byeSubscriber(String key,String streamId, long time, TimeUnit unit){
+        ScheduledFuture<?>[] schedule = new ScheduledFuture<?>[1];
+        Flow.Subscriber<SIPRequest> subscriber = new Flow.Subscriber<>() {
+            @Override
+            public void onSubscribe(Flow.Subscription subscription) {
+                log.info("创建订阅 {}", key);
+                subscription.request(1);
+            }
+
+            @SneakyThrows
+            @Override
+            public void onNext(SIPRequest request) {
+                String transport = request.getTopmostViaHeader().getTransport();
+                String ip = request.getLocalAddress().getHostAddress();
+                sender.getProvider(transport,ip)
+                        .sendResponse(SipResponseBuilder.response(request, Response.OK, "OK"));
+                onComplete();
+            }
+
+            @Override
+            public void onError(Throwable throwable) {
+                throwable.printStackTrace();
+                onComplete();
+            }
+
+            @Override
+            public void onComplete() {
+                subscribe.getByeSubscribe().delPublisher(key);
+                schedule[0].cancel(true);
+                zlmMediaService.closeRtpServer(new CloseRtpServer(streamId));
+            }
+        };
+        schedule[0] = scheduledExecutorService.schedule(subscriber::onComplete, time, unit);
+        return subscriber;
     }
 }

gb28181-wvp-proxy-service/src/main/java/cn/skcks/docking/gb28181/wvp/service/video/VideoService.java

@@ -22,6 +22,7 @@ import org.springframework.stereotype.Service;
 import java.io.*;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
@@ -163,14 +164,13 @@ public class VideoService {
         Executor executor = ffmpegSupportService.downloadToStream(url, time, TimeUnit.SECONDS,streamHandler,defaultExecuteResultHandler);
         log.info("开始录制 {}", url);
         ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
-        scheduledExecutorService.schedule(() -> {
+        ScheduledFuture<?> schedule = scheduledExecutorService.schedule(() -> {
             log.info("到达结束时间, 结束录制 {}", url);
             executor.getWatchdog().destroyProcess();
             log.info("结束录制 {}", url);
         }, time, TimeUnit.SECONDS);
         defaultExecuteResultHandler.waitFor();
-        if(errorStream.size() > 0){
+        schedule.cancel(true);
-            log.error("{}", errorStream);
+        log.info("结束录制 {}", url);
-        }
     }
 }

gb28181-wvp-proxy-service/src/main/java/cn/skcks/docking/gb28181/wvp/sip/message/bye/request/request/ByeRequestProcessor.java

@@ -40,6 +40,7 @@ public class ByeRequestProcessor implements MessageProcessor {
         SIPRequest request = (SIPRequest) requestEvent.getRequest();
         String callId = request.getCallId().getCallId();
         String key = GenericSubscribe.Helper.getKey(Request.BYE, callId);
+        log.info("key {}", key);
         String ip = request.getLocalAddress().getHostAddress();
         String transport = request.getTopmostViaHeader().getTransport();
         Optional.ofNullable(subscribe.getByeSubscribe().getPublisher(key))

gb28181-wvp-proxy-starter/src/main/resources/application-local.yml

@@ -40,6 +40,8 @@ media:
   id: amrWMKmbKqoBjRQ9
 #  secret: 035c73f7-bb6b-4889-a715-d9eb2d1925cc
   secret: 4155cca6-2f9f-11ee-85e6-8de4ce2e7333
+  rtsp:
+    url: 'rtmp://10.10.10.200:1935'
 
 proxy:
   wvp:
@@ -65,6 +67,7 @@ ffmpeg-support:
   rtp:
 #    input: -i http://10.10.10.200:5080/live/test.live.flv
     input: -re -i
+    download: -i
     output: -vcodec copy -acodec copy -preset ultrafast -movflags empty_moov+frag_keyframe+default_base_moof -f mp4 # -rtsp_transport tcp
   debug:
     download: false

gb28181-wvp-proxy-starter/src/main/resources/application.yml

@@ -39,6 +39,8 @@ spring:
 media:
   ip: 192.168.1.241
   url: 'http://192.168.1.241:5080'
+  rtsp:
+    url: 'rtmp://192.168.1.241:1935'
   #  url: 'http://10.10.10.200:12580/anything/'
   id: amrWMKmbKqoBjRQ9
   secret: 035c73f7-bb6b-4889-a715-d9eb2d1925cc
@@ -53,7 +55,7 @@ proxy:
     # 是否使用 wvp 的 api(wvp 的 并发有问题,仅保留用于兼容), 否则使用sip 信令直接操作设备
     enable: false
     # 是否使用 ffmpeg 编/解码, 否则使用内置 javacv
-    use-ffmpeg: false
+    use-ffmpeg: true
   gb28181:
     sip:
       id: 44050100002000000005
@@ -72,8 +74,9 @@ ffmpeg-support:
   ffmpeg: C:\ffmpeg\bin\ffmpeg.exe
   ffprobe: /usr/bin/ffmpeg/ffprobe
   rtp:
-    input: -i
+    input: -re -i
-    output: -vcodec h264 -acodec aac -preset ultrafast -movflags empty_moov+frag_keyframe+default_base_moof -f mp4 # -rtsp_transport tcp
+    download: -i
+    output: -vcodec libx264 -acodec aac -preset ultrafast -movflags empty_moov+frag_keyframe+default_base_moof -f mp4 # -rtsp_transport tcp
   debug:
     download: false
     input: false