可切换是否使用ffmpeg处理视频

gb28181-wvp-proxy-api/src/main/java/cn/skcks/docking/gb28181/wvp/api/device/dto/DevicePageDTO.java

@@ -10,10 +10,10 @@ public class DevicePageDTO {
     @Schema(description = "页数", example = "1")
     @NotNull(message = "page 不能为空")
     @Min(value = 1, message = "page 必须为正整数")
-    int page;
+    Integer page;
 
     @Schema(description = "每页条数", example = "10")
     @NotNull(message = "size 不能为空")
     @Min(value = 1, message = "size 必须为正整数")
-    int size;
+    Integer size;
 }

gb28181-wvp-proxy-api/src/main/java/cn/skcks/docking/gb28181/wvp/api/video/VideoController.java

@@ -15,6 +15,7 @@ import org.springdoc.core.annotations.ParameterObject;
 import org.springdoc.core.models.GroupedOpenApi;
 import org.springframework.context.annotation.Bean;
 import org.springframework.http.MediaType;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Controller;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RequestMapping;

gb28181-wvp-proxy-service/pom.xml

@@ -196,6 +196,13 @@
                 </exclusion>
             </exclusions>
         </dependency>
+
+        <!-- https://mvnrepository.com/artifact/org.apache.commons/commons-exec -->
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-exec</artifactId>
+            <version>1.3</version>
+        </dependency>
     </dependencies>
 
     <build>

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

@@ -0,0 +1,40 @@
+package cn.skcks.docking.gb28181.wvp.config;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+
+@ConfigurationProperties(prefix = "ffmpeg-support")
+@Configuration
+@Data
+public class FfmpegConfig {
+    private String ffmpeg;
+    private String ffprobe;
+
+    private Rtp rtp;
+
+    @Data
+    public static class Rtp {
+        private String download = "-i";
+        private String input = "-re -i";
+        private String output = "-vcodec h264 -acodec aac -f mp4";
+        private String logLevel = "fatal";
+    }
+
+
+    private Debug debug;
+
+    @Data
+    public static class Debug {
+        private Boolean download = false;
+        private Boolean input = false;
+        private Boolean output = false;
+    }
+
+    private Task task;
+
+    @Data
+    public static class Task {
+        private Integer max = 4;
+    }
+}

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

@@ -8,6 +8,7 @@ import org.springframework.context.annotation.Configuration;
 @ConfigurationProperties(prefix = "proxy.wvp")
 @Data
 public class WvpProxyConfig {
+    private Boolean enable = true;
     private String url;
     private String user;
     private String passwd;
@@ -15,4 +16,9 @@ public class WvpProxyConfig {
      * 是否尝试通过 wvp-assist 服务下载
      */
     private Boolean useWvpAssist = true;
+
+    /**
+     * 是否使用 ffmpeg 编/解码, 否则使用内置 javacv
+     */
+    private Boolean useFfmpeg = false;
 }

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

@@ -0,0 +1,51 @@
+package cn.skcks.docking.gb28181.wvp.service.ffmpeg;
+
+
+import cn.skcks.docking.gb28181.wvp.config.FfmpegConfig;
+import lombok.RequiredArgsConstructor;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.exec.*;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.stereotype.Service;
+
+import java.util.concurrent.TimeUnit;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class FfmpegSupportService {
+    private final FfmpegConfig ffmpegConfig;
+
+    @SneakyThrows
+    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);
+        log.info("视频输入参数 {}", inputParam);
+
+        String outputParam = debug.getOutput() ? rtp.getOutput() : StringUtils.joinWith(" ", rtp.getOutput(), "-");
+        log.info("视频输出参数 {}", outputParam);
+
+        return ffmpegExecutor(inputParam, outputParam, time, unit, streamHandler, executeResultHandler);
+    }
+
+    @SneakyThrows
+    public Executor ffmpegExecutor(String inputParam,String outputParam, long time, TimeUnit unit,ExecuteStreamHandler streamHandler,ExecuteResultHandler executeResultHandler){
+        FfmpegConfig.Rtp rtp = ffmpegConfig.getRtp();
+        String logLevelParam = StringUtils.joinWith(" ","-loglevel", rtp.getLogLevel());
+        String command = StringUtils.joinWith(" ", ffmpegConfig.getFfmpeg(), inputParam, outputParam, logLevelParam);
+        CommandLine commandLine = CommandLine.parse(command);
+        Executor executor = new DefaultExecutor();
+        ExecuteWatchdog watchdog = new ExecuteWatchdog(unit.toMillis(time));
+        executor.setStreamHandler(streamHandler);
+        executor.setExitValues(null);
+        executor.setWatchdog(watchdog);
+        if(executeResultHandler == null){
+            executor.execute(commandLine);
+        } else {
+            executor.execute(commandLine, executeResultHandler);
+        }
+        return executor;
+    }
+}

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

@@ -1,22 +1,26 @@
 package cn.skcks.docking.gb28181.wvp.service.video;
 
+import cn.skcks.docking.gb28181.wvp.config.WvpProxyConfig;
+import cn.skcks.docking.gb28181.wvp.service.ffmpeg.FfmpegSupportService;
 import jakarta.servlet.AsyncContext;
+import jakarta.servlet.ServletOutputStream;
 import jakarta.servlet.ServletResponse;
 import jakarta.servlet.http.HttpServletRequest;
 import jakarta.servlet.http.HttpServletResponse;
+import lombok.RequiredArgsConstructor;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.exec.*;
 import org.bytedeco.ffmpeg.avcodec.AVPacket;
 import org.bytedeco.ffmpeg.global.avcodec;
 import org.bytedeco.ffmpeg.global.avutil;
 import org.bytedeco.javacv.FFmpegFrameGrabber;
 import org.bytedeco.javacv.FFmpegFrameRecorder;
 import org.bytedeco.javacv.FrameGrabber;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 
-import java.io.ByteArrayOutputStream;
+import java.io.*;
-import java.io.IOException;
-import java.io.OutputStream;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -24,7 +28,10 @@ import java.util.concurrent.atomic.AtomicBoolean;
 
 @Slf4j
 @Service
+@RequiredArgsConstructor
 public class RecordService {
+    private final FfmpegSupportService ffmpegSupportService;
+    private final WvpProxyConfig wvpProxyConfig;
     /**
      * 写入 flv 响应头信息
      * @param response  HttpServletResponse 响应
@@ -66,6 +73,21 @@ public class RecordService {
      */
     @SneakyThrows
     public void record(ServletResponse response, String url, long time) {
+        if (wvpProxyConfig.getUseFfmpeg()) {
+            ffmpegRecord(response, url, time);
+        } else {
+            javaCVrecord(response, url, time);
+        }
+    }
+
+    /**
+     * 录制视频 并写入 异步响应
+     * @param response AsyncContext.getResponse 异步响应
+     * @param url 要录制的视频地址
+     * @param time 录制时长 (单位: 秒)
+     */
+    @SneakyThrows
+    public void javaCVrecord(ServletResponse response, String url, long time) {
         ByteArrayOutputStream stream = new ByteArrayOutputStream();
         OutputStream outputStream = response.getOutputStream();
         log.info("准备录制 url {}, time: {}", url, time);
@@ -120,8 +142,40 @@ public class RecordService {
         } catch (FFmpegFrameRecorder.Exception | FrameGrabber.Exception e) {
             throw new RuntimeException(e);
         } catch (IOException ignore){}
+        finally {
+            log.info("结束录制 {}", url);
+            stream.close();
+            outputStream.close();
+        }
+    }
 
-        log.info("结束录制 {}", url);
+    /**
-        outputStream.close();
+     * 录制视频 并写入 异步响应
+     * @param response AsyncContext.getResponse 异步响应
+     * @param url 要录制的视频地址
+     * @param time 录制时长 (单位: 秒)
+     */
+    @SneakyThrows
+    public void ffmpegRecord(ServletResponse response, String url, long time){
+        ServletOutputStream outputStream = response.getOutputStream();
+        ByteArrayOutputStream errorStream = new ByteArrayOutputStream();
+        PumpStreamHandler streamHandler = new PumpStreamHandler(outputStream, errorStream);
+        DefaultExecuteResultHandler defaultExecuteResultHandler = new DefaultExecuteResultHandler();
+        Executor executor = ffmpegSupportService.downloadToStream(url, time, TimeUnit.SECONDS,streamHandler,defaultExecuteResultHandler);
+        // executor.setStreamHandler(streamHandler);
+        log.info("开始录制 {}", url);
+        ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
+        AtomicBoolean record = new AtomicBoolean(true);
+        scheduledExecutorService.schedule(() -> {
+            log.info("到达结束时间, 结束录制 {}", url);
+            executor.getWatchdog().destroyProcess();
+            log.info("结束录制 {}", url);
+            // try {
+            //     outputStream.close();
+            // } catch (IOException e) {
+            //     throw new RuntimeException(e);
+            // }
+        }, time, TimeUnit.SECONDS);
+        defaultExecuteResultHandler.waitFor();
     }
 }

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

@@ -34,6 +34,7 @@ import lombok.RequiredArgsConstructor;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.http.MediaType;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 import org.springframework.util.CollectionUtils;
 
@@ -58,7 +59,7 @@ public class WvpService {
         response.setContentType("video/mp4");
         response.setHeader("Accept-Ranges", "none");
         response.setHeader("Connection", "close");
-        response.setHeader("Content-Disposition","attachment; filename=\"record.mp4\"");
+        // response.setHeader("Content-Disposition","attachment; filename=\"record.mp4\"");
     }
 
     @SneakyThrows
@@ -79,13 +80,13 @@ public class WvpService {
         String deviceId = wvpProxyDevice.getGbDeviceId();
         String channelId = wvpProxyDevice.getGbDeviceChannelId();
         log.info("设备编码 (deviceCode=>{}) 查询到的设备信息 国标id(gbDeviceId => {}), 通道(channelId => {})", deviceCode, deviceId, channelId);
-
+        request.startAsync();
-        Retryer<JsonResponse<?>> genericRetryer = RetryUtil.getDefaultGenericRetryer("调用 wvp api 查询设备历史");
+        AsyncContext asyncContext = request.getAsyncContext();
-        AsyncContext asyncContext = request.startAsync();
         asyncContext.setTimeout(0);
         asyncContext.start(() -> {
             HttpServletResponse asyncResponse = (HttpServletResponse) asyncContext.getResponse();
             try {
+                Retryer<JsonResponse<?>> genericRetryer = RetryUtil.getDefaultGenericRetryer("调用 wvp api 查询设备历史");
                 genericRetryer.call(() -> video(asyncResponse, deviceCode, deviceId, channelId, startTime, endTime));
             } catch (RetryException e) {
                 Attempt<?> failedAttempt = e.getLastFailedAttempt();
@@ -101,6 +102,7 @@ public class WvpService {
                 String reason = MessageFormat.format("调用 wvp api 查询设备({0})历史失败, 异常: {1}", deviceCode, e.getMessage());
                 writeErrorToResponse(asyncResponse, JsonResponse.error(reason));
             } finally {
+                log.info("asyncContext 结束");
                 asyncContext.complete();
             }
         });
@@ -202,7 +204,11 @@ public class WvpService {
         Retryer<JsonResponse<StreamContent>> playBackRetryer = RetryUtil
                 .<StreamContent>getDefaultRetryerBuilder("通过回放获取实时视频流下载", 100, TimeUnit.MILLISECONDS, 5)
                 .build();
-        JsonResponse<StreamContent> videoStreamResponse = playBackRetryer.call(() -> wvpProxyClient.playbackStart(token, deviceId, channelId, new GeneralTimeReq(startTime, endTime)));
+        JsonResponse<StreamContent> videoStreamResponse = playBackRetryer.call(() -> {
+            JsonResponse<StreamContent> streamContentJsonResponse = wvpProxyClient.playbackStart(token, deviceId, channelId, new GeneralTimeReq(startTime, endTime));
+            log.info("实时视频流下载 响应:{} ", streamContentJsonResponse);
+            return streamContentJsonResponse;
+        });
         StreamContent streamContent = videoStreamResponse.getData();
         String stream = streamContent.getStream();
         String streamUrl = streamContent.getFlv();

gb28181-wvp-proxy-starter/Dockerfile

@@ -6,6 +6,11 @@ ENV HOME_PATH /opt/gb28181-docking-platform-wvp-proxy/
 RUN mkdir -p $HOME_PATH
 WORKDIR $HOME_PATH
 
+RUN mkdir -p /usr/bin/ffmpeg
+ADD ./ffmpeg/ffmpeg /usr/bin/ffmpeg/ffmpeg
+ADD ./ffmpeg/ffprobe /usr/bin/ffmpeg/ffprobe
+ADD ./ffmpeg/model /usr/bin/ffmpeg/model
+
 ADD target/gb28181-wvp-proxy-starter.jar /opt/gb28181-docking-platform-wvp-proxy/starter.jar
 
 EXPOSE 18183

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

@@ -43,6 +43,22 @@ media:
 
 proxy:
   wvp:
-    url: http://10.10.10.20:18978
+    url: http://127.0.0.1:18978
     user: admin
-    passwd: admin
+    passwd: admi
+    use-ffmpeg: false
+    enable: true
+
+ffmpeg-support:
+  task:
+    max: 4
+  ffmpeg: D:\Soft\Captura\ffmpeg\ffmpeg.exe
+  ffprobe: D:\Soft\Captura\ffmpeg\ffprobe.exe
+  rtp:
+    input: -r -i http://10.10.10.200:5080/live/test.live.flv
+    #    input: -re -i
+    output: -vcodec copy -acodec copy -movflags empty_moov+frag_keyframe+default_base_moof -f mp4 # -rtsp_transport tcp
+  debug:
+    download: false
+    input: true
+    output: false

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

@@ -24,8 +24,8 @@ spring:
     username: root
     password: 123456a
     url: jdbc:mysql://192.168.1.241:3306/gb28181_docking_platform?createDatabaseIfNotExist=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai
-#  profiles:
+  profiles:
-#    active: local
+    active: local
   cloud:
     openfeign:
       httpclient:
@@ -50,3 +50,18 @@ proxy:
     user: admin
     passwd: admin
     use-wvp-assist: false
+    enable: true
+    use-ffmpeg: false
+
+ffmpeg-support:
+  task:
+    max: 4
+  ffmpeg: /usr/bin/ffmpeg/ffmpeg
+  ffprobe: /usr/bin/ffmpeg/ffprobe
+  rtp:
+    input: -i
+    output: -vcodec h264 -acodec aac -movflags empty_moov+frag_keyframe+default_base_moof -f mp4 # -rtsp_transport tcp
+  debug:
+    download: false
+    input: true
+    output: false

gb28181-wvp-proxy-starter/src/main/resources/logback.xml

@@ -0,0 +1,54 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<configuration scan="true" scanPeriod="60 seconds" debug="false">
+
+    <contextName>logback</contextName>
+    <!--定义日志文件的存储地址 勿在 LogBack 的配置中使用相对路径-->
+    <!--<property name="log.path" value="./log/business_Log" />-->
+
+    <!--输出到控制台-->
+    <appender name="console" class="ch.qos.logback.core.ConsoleAppender">
+<!--        <filter class="ch.qos.logback.classic.filter.ThresholdFilter">-->
+<!--            <level>INFO</level>-->
+<!--        </filter>-->
+<!--        <withJansi>true</withJansi>-->
+        <encoder>
+            <!--<pattern>%d %p (%file:%line\)- %m%n</pattern>-->
+            <!--格式化输出：%d:表示日期    %thread:表示线程名     %-5level:级别从左显示5个字符宽度  %msg:日志消息    %n:是换行符-->
+            <pattern>%red(%d{yyyy-MM-dd HH:mm:ss.SSS}) %green([%thread]) %highlight(%-5level) %yellow(at %class.%method) (%file:%line\) - %cyan(%msg%n)</pattern>
+            <!--<pattern>%d{yyyy-MM-dd HH:mm:ss} [%thread] %magenta(%-5level) %green([%-50.50class]) >>> %cyan(%msg) %n</pattern>-->
+            <charset>UTF-8</charset>
+        </encoder>
+    </appender>
+
+    <!--&lt;!&ndash;输出到文件&ndash;&gt;-->
+    <!--<appender name="file" class="ch.qos.logback.core.rolling.RollingFileAppender">-->
+    <!--    <filter class="ch.qos.logback.classic.filter.ThresholdFilter">-->
+    <!--        <level>INFO</level>-->
+    <!--    </filter>-->
+    <!--    <file>${log.path}/logback.log</file>-->
+    <!--    <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">-->
+    <!--        <fileNamePattern>${log.path}/logback-%d{yyyy-MM-dd-HH-mm}.log</fileNamePattern>-->
+    <!--        <maxHistory>365</maxHistory>-->
+    <!--        &lt;!&ndash;            <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">&ndash;&gt;-->
+    <!--        &lt;!&ndash;                <maxFileSize>100kB</maxFileSize>&ndash;&gt;-->
+    <!--        &lt;!&ndash;            </timeBasedFileNamingAndTriggeringPolicy>&ndash;&gt;-->
+    <!--    </rollingPolicy>-->
+    <!--    <encoder>-->
+    <!--        &lt;!&ndash;格式化输出：%d:表示日期    %thread:表示线程名     %-5level:级别从左显示5个字符宽度  %msg:日志消息    %n:是换行符&ndash;&gt;-->
+    <!--        <pattern>%d{yyyy-MM-dd HH:mm:ss} [%thread] %-5level %logger{36} - %msg%n</pattern>-->
+    <!--        <charset>UTF-8</charset>-->
+    <!--    </encoder>-->
+    <!--</appender>-->
+
+    <!--    如果appender里没有限定日志级别，那么root可以统一设置，如果没有配置那么控制台和文件不会输出任何日志，这里root的level不做限制-->
+    <root level="INFO">
+        <!--        允许控制台输出-->
+        <appender-ref ref="console" />
+        <!--&lt;!&ndash;        允许文件输出&ndash;&gt;-->
+        <!--<appender-ref ref="file" />-->
+    </root>
+
+    <logger name="cn.skcks.docking.gb28181.core.sip.logger" level="INFO" />
+    <logger name="cn.skcks.docking.gb28181" level="DEBUG" />
+</configuration>

package-docker.sh

@@ -2,6 +2,19 @@
 # 用于缓存打包过程下载的依赖
 mkdir repository
 curDir=`pwd`
+repository=${curDir}/repository
+
+if ! test -e ${curDir}/gb28181-wvp-proxy-starter/ffmpeg;then
+  xz -d ${curDir}/ffmpeg/ffmpeg-git-amd64-static.tar.xz
+  tar -xvf ${curDir}/ffmpeg/ffmpeg-git-amd64-static.tar -C ${curDir}/ffmpeg/
+  mv ${curDir}/ffmpeg/ffmpeg-git*-static/* ${curDir}/ffmpeg
+  rm -rf ${curDir}/ffmpeg/ffmpeg-git*-static
+  mkdir -p ${curDir}/gb28181-wvp-proxy-starter/ffmpeg
+  cp ${curDir}/ffmpeg/ffmpeg ${curDir}/gb28181-wvp-proxy-starter/ffmpeg
+  cp ${curDir}/ffmpeg/ffprobe ${curDir}/gb28181-wvp-proxy-starter/ffmpeg
+  cp -r ${curDir}/ffmpeg/model ${curDir}/gb28181-wvp-proxy-starter/ffmpeg
+fi
+
 docker run --name maven --rm \
 	-v ${curDir}:/usr/src/mymaven \
 	-v ${curDir}/repository:/root/.m2/repository \