feat: 增加rtsp转flv的视频流转换

(cherry picked from commit c7134b0b)

+package org.thingsboard.server.controller.yunteng;
+
+import io.swagger.annotations.Api;
+import lombok.RequiredArgsConstructor;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+import org.thingsboard.server.common.data.StringUtils;
+import org.thingsboard.server.dao.yunteng.service.rtsp.TkFLVService;
+
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+@RestController
+@RequestMapping("api/yt/rtsp")
+@Api(tags = {"rtsp流转换"})
+@RequiredArgsConstructor
+public class TkRtspConvertController {
+
+  private final TkFLVService tkFLVService;
+
+  @GetMapping("/openFlv")
+  public void open(
+      @RequestParam String url,@RequestParam String browserId, HttpServletResponse response, HttpServletRequest request) {
+    if (StringUtils.isNotEmpty(url)) {
+      tkFLVService.open(url,browserId, response, request);
+    }
+  }
+
+  @GetMapping("/closeFlv")
+  public void close(@RequestParam String url,@RequestParam String browserId) {
+    if (StringUtils.isNotEmpty(url)) {
+      tkFLVService.close(url,browserId);
+    }
+  }
+}

@@ -254,6 +254,16 @@
                 </exclusion>
                 </exclusion>
             </exclusions>
             </exclusions>
         </dependency>
         </dependency>
+        <dependency>
+            <groupId>org.bytedeco</groupId>
+            <artifactId>javacv-platform</artifactId>
+            <version>${javacv.platform.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>javax.xml.bind</groupId>
+            <artifactId>jaxb-api</artifactId>
+            <version>${jaxb.api.version}</version>
+        </dependency>
     </dependencies>
     </dependencies>
     <build>
     <build>
         <plugins>
         <plugins>

+package org.thingsboard.server.dao.yunteng.impl.rtsp;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.io.IOUtils;
+import org.bytedeco.ffmpeg.avcodec.AVPacket;
+import org.bytedeco.ffmpeg.global.avcodec;
+import org.bytedeco.javacv.FFmpegFrameGrabber;
+import org.bytedeco.javacv.FFmpegFrameRecorder;
+import org.thingsboard.server.dao.yunteng.service.rtsp.TkConverter;
+
+import javax.servlet.AsyncContext;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+@Slf4j
+public class TkAbstractBaseConvert implements TkConverter {
+
+  public volatile boolean runing = true;
+  /** 读流器 */
+  private FFmpegFrameGrabber grabber;
+  /** 转码器 */
+  private FFmpegFrameRecorder recorder;
+  /**
+   * 转FLV格式的头信息<br>
+   * 如果有第二个客户端播放首先要返回头信息
+   */
+  private byte[] headers;
+  /** 保存转换好的流 */
+  private ByteArrayOutputStream stream;
+  /** 流地址，h264,aac */
+  private String url;
+  /** 流输出 */
+  private List<AsyncContext> outEntitys;
+
+  /** key用于表示这个转换器 */
+  private String key;
+
+  /** 转换队列 */
+  private Map<String, TkConverter> factories;
+
+  public TkAbstractBaseConvert(
+      String url, String key, Map<String, TkConverter> factories, List<AsyncContext> outEntitys) {
+    this.url = url;
+    this.outEntitys = outEntitys;
+    this.key = key;
+    this.factories = factories;
+  }
+
+  /**
+   * 输出FLV视频流
+   *
+   * @param b
+   */
+  private void writeResponse(byte[] b, List<AsyncContext> outEntitys) {
+    Iterator<AsyncContext> it = outEntitys.iterator();
+    while (it.hasNext()) {
+      AsyncContext o = it.next();
+      try {
+        o.getResponse().getOutputStream().write(b);
+      } catch (Exception e) {
+        if (null != it && it.hasNext()) {
+          it.remove();
+          log.info("移除一个输出");
+        }
+      }
+    }
+  }
+
+  @Override
+  public void addOutputStreamEntity(String key, AsyncContext entity) throws IOException {
+    outEntitys.add(entity);
+    if (headers != null) {
+      entity.getResponse().getOutputStream().write(headers);
+      entity.getResponse().getOutputStream().flush();
+    }
+  }
+
+  @Override
+  public void convert() {
+    try {
+      grabber = new FFmpegFrameGrabber(url);
+      if ("rtsp".equals(url.substring(0, 4))) {
+        grabber.setOption("rtsp_transport", "tcp");
+        grabber.setOption("stimeout", "5000000");
+      }
+      grabber.startUnsafe();
+      stream = new ByteArrayOutputStream();
+      recorder =
+          new FFmpegFrameRecorder(
+              stream,
+              grabber.getImageWidth(),
+              grabber.getImageHeight(),
+              grabber.getAudioChannels());
+      recorder.setInterleaved(true);
+      recorder.setVideoOption("preset", "ultrafast");
+      recorder.setVideoOption("tune", "zerolatency");
+      double framerate = 25.0;
+      if (grabber.getFrameRate() > 0 && grabber.getFrameRate() < 100) {
+        framerate = grabber.getFrameRate();
+      }
+      recorder.setVideoOption("crf", String.valueOf(framerate));
+      recorder.setFrameRate(grabber.getFrameRate());
+      recorder.setSampleRate(grabber.getSampleRate());
+      recorder.setFormat("flv");
+      recorder.setTimestamp(grabber.getTimestamp());
+      log.info("this url:{} converter start", url);
+      if (avcodec.AV_CODEC_ID_H264 == grabber.getVideoCodec()) {
+        rtspConvert();
+      }
+      factories.put(key, this);
+    } catch (Exception e) {
+      log.error(e.getMessage(), e);
+    } finally {
+      closeConverter();
+      completeResponse();
+      log.info("this url:{} converter exit", url);
+      factories.remove(this.key);
+    }
+  }
+  private void rtspConvert()
+      throws FFmpegFrameGrabber.Exception, FFmpegFrameRecorder.Exception, InterruptedException {
+    // 来源视频H264格式,音频AAC格式
+    // 无须转码，更低的资源消耗，更低的延迟
+    recorder.setVideoBitrate(grabber.getVideoBitrate());
+    recorder.setVideoCodec(grabber.getVideoCodec());
+    recorder.start(grabber.getFormatContext());
+    if (headers == null) {
+      headers = stream.toByteArray();
+      stream.reset();
+      writeResponse(headers, outEntitys);
+    }
+    int nullNumber = 0;
+    long dts = 0;
+    long pts = 0;
+    int timebase;
+    while (runing) {
+      AVPacket k = grabber.grabPacket();
+      if (k != null) {
+        k.pts(pts);
+        k.dts(dts);
+        timebase = grabber.getFormatContext().streams(k.stream_index()).time_base().den();
+        pts += timebase / grabber.getFrameRate();
+        dts += timebase / grabber.getFrameRate();
+        try {
+          recorder.recordPacket(k);
+        } catch (Exception e) {
+        }
+        if (stream.size() > 0) {
+          byte[] b = stream.toByteArray();
+          stream.reset();
+          writeResponse(b, outEntitys);
+          if (outEntitys.isEmpty()) {
+            log.info("没有输出退出");
+            break;
+          }
+        }
+        avcodec.av_packet_unref(k);
+      } else {
+        nullNumber++;
+        if (nullNumber > 200) {
+          break;
+        }
+      }
+      Thread.sleep(5);
+    }
+  }
+
+  /** 退出转换 */
+  public void closeConverter() {
+    try {
+      IOUtils.close(grabber);
+      factories.remove(this.key);
+      IOUtils.close(recorder);
+      IOUtils.close(stream);
+    } catch (Exception e) {
+
+    }
+  }
+
+  /** 关闭异步响应 */
+  public void completeResponse() {
+    Iterator<AsyncContext> it = outEntitys.iterator();
+    while (it.hasNext()) {
+      AsyncContext o = it.next();
+      try{
+        o.complete();
+      }catch (Exception e){
+        log.info("请求已完成处理");
+      }
+    }
+  }
+
+  @Override
+  public void exit() {
+    this.runing = false;
+  }
+}

+package org.thingsboard.server.dao.yunteng.impl.rtsp;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import org.thingsboard.common.util.ThingsBoardThreadFactory;
+import org.thingsboard.server.dao.yunteng.service.rtsp.TkConverter;
+import org.thingsboard.server.dao.yunteng.service.rtsp.TkFLVService;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import javax.servlet.AsyncContext;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import java.io.IOException;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.List;
+import java.util.concurrent.*;
+
+@Service
+@Slf4j
+public class TkFLVServiceImpl implements TkFLVService {
+  private ConcurrentHashMap<String, TkConverter> converters = new ConcurrentHashMap<>();
+  private ExecutorService executorService;
+
+  @PostConstruct
+  public void init() {
+    executorService =
+        new ThreadPoolExecutor(
+            0,
+            50,
+            10L,
+            TimeUnit.SECONDS,
+            new SynchronousQueue<Runnable>(),
+            ThingsBoardThreadFactory.forName("rtsp-convert-flv"));
+  }
+
+  @PreDestroy
+  public void stop() {
+    if (executorService != null) {
+      executorService.shutdownNow();
+    }
+  }
+
+  @Override
+  public void open(
+      String url, String browserId, HttpServletResponse response, HttpServletRequest request) {
+    String key = md5(url + browserId);
+    AsyncContext async = request.startAsync();
+    async.setTimeout(900000000);
+    if (converters.containsKey(key)) {
+      TkConverter c = converters.get(key);
+      try {
+        c.addOutputStreamEntity(key, async);
+      } catch (IOException e) {
+        log.error(e.getMessage(), e);
+        throw new IllegalArgumentException(e.getMessage());
+      }
+    } else {
+      executorService.execute(
+          () -> {
+            List<AsyncContext> outs = List.of(async);
+            TkAbstractBaseConvert c = new TkAbstractBaseConvert(url, key, converters, outs);
+            c.convert();
+          });
+    }
+    response.setHeader("Connection", "keep-alive");
+
+    response.setStatus(HttpServletResponse.SC_OK);
+    try {
+      response.flushBuffer();
+    } catch (IOException e) {
+      log.error(e.getMessage(), e);
+    }
+  }
+
+  public String md5(String plainText) {
+    StringBuilder buf = null;
+    try {
+      MessageDigest md = MessageDigest.getInstance("MD5");
+      md.update(plainText.getBytes());
+      byte b[] = md.digest();
+      int i;
+      buf = new StringBuilder("");
+      for (int offset = 0; offset < b.length; offset++) {
+        i = b[offset];
+        if (i < 0) i += 256;
+        if (i < 16) buf.append("0");
+        buf.append(Integer.toHexString(i));
+      }
+    } catch (NoSuchAlgorithmException e) {
+      log.error(e.getMessage(), e);
+    }
+    return buf.toString();
+  }
+
+  public void close(String url, String browserId) {
+    //    executorService.execute(()->{
+    //
+    //    });
+    String key = md5(url + browserId);
+    if (converters.containsKey(key)) {
+      TkConverter c = converters.get(key);
+      c.exit();
+    }
+  }
+}

+package org.thingsboard.server.dao.yunteng.service.rtsp;
+
+import javax.servlet.AsyncContext;
+import java.io.IOException;
+
+public interface TkConverter {
+    /**
+     * 添加一个流输出
+     *
+     * @param entity
+     */
+    void addOutputStreamEntity(String key, AsyncContext entity) throws IOException;
+
+    /**
+     * 退出转换
+     */
+    void exit();
+
+    void convert();
+}

+package org.thingsboard.server.dao.yunteng.service.rtsp;
+
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+public interface TkFLVService {
+  /**
+   * 打开一个流地址
+   *
+   * @param url 流地址
+   * @param response 响应
+   * @param request 请求
+   */
+  void open(String url, String browserId,HttpServletResponse response, HttpServletRequest request);
+
+  /**
+   * 关闭视频流
+   *
+   * @param url 视频流地址
+   */
+  void close(String url,String browserId);
+}

@@ -141,6 +141,8 @@
         <jakarta.mail.version>2.0.1</jakarta.mail.version>
         <jakarta.mail.version>2.0.1</jakarta.mail.version>
         <io.minio.version>8.3.1</io.minio.version>
         <io.minio.version>8.3.1</io.minio.version>
         <com.alibaba.easyexcel.version>3.1.1</com.alibaba.easyexcel.version>
         <com.alibaba.easyexcel.version>3.1.1</com.alibaba.easyexcel.version>
+        <javacv.platform.version>1.5.9</javacv.platform.version>
+        <jaxb.api.version>2.3.1</jaxb.api.version>
     </properties>
     </properties>
 
 
     <modules>
     <modules>