增加：tcp连接注册和心跳

liusuyi

2024-10-22 5141265c6214e84d4ef8a48c6cead6c372cc3892

增加：tcp连接注册和心跳

 ard-admin/src/main/java/com/ard/ArdUavApplication.java

@@ -3,12 +3,14 @@
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
import org.springframework.scheduling.annotation.EnableScheduling;

/**
 * 启动程序
 * 
 * @author ruoyi
 */
@EnableScheduling
@SpringBootApplication(exclude = { DataSourceAutoConfiguration.class })
public class ArdUavApplication
{

 ard-admin/src/main/resources/application-dev.yml

@@ -23,8 +23,19 @@

# 天宇服务器配置
tianyu:
  # 天宇服务器地址
  # 天宇http服务器地址
  endpoint: 112.98.106.22:12000
  accesskeyid: qmfmd3Ws7M4dMcylnrTU
  accesskeysecret: 5WBU2uEhHZaVefpZdaxpFh4LJW9fZZaa

  # 天宇tcp服务器地址
  tcpserver: 112.98.106.22
  tcpport: 13002
# netty配置
netty:
  tcp:
    enabled: true
    reConnectInterval: 30000
#流媒体配置
mediamtx:
  # 流媒体服务器地址
  host: 192.168.1.27

 ard-admin/src/main/resources/application.yml

@@ -89,7 +89,7 @@
  read-timeout: 3000           # 数据读取超时时间，单位为毫秒（默认为 timeout）
  max-retry-count: 0           # 请求失败后重试次数（默认为 0 次不重试）
  ssl-protocol: TLS            # 单向验证的HTTPS的默认TLS协议（默认为 TLS）
  log-enabled: true            # 打开或关闭日志（默认为 true）
  log-enabled: false            # 打开或关闭日志（默认为 true）
  log-request: true            # 打开/关闭Forest请求日志（默认为 true）
  log-response-status: true    # 打开/关闭Forest响应状态日志（默认为 true）
  log-response-content: true   # 打开/关闭Forest响应内容日志（默认为 false）

 ard-work/pom.xml

@@ -35,6 +35,14 @@
            <artifactId>forest-spring-boot-starter</artifactId>
            <version>1.5.33</version>
        </dependency>

        <!-- netty依赖 springboot2.x自动导入版本 -->
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
        </dependency>

        <!-- lombok -->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>

 ard-work/src/main/java/com/ard/tianyu/config/TyConfiguration.java

@@ -33,17 +33,20 @@

    private String accesskeysecret;

    private String tcpserver;

    private Integer tcpport;

    private long timestamp;
    public long getTimeStamp()
    {

    public long getTimeStamp() {
        timestamp = new Date().getTime();
        return timestamp;
    }

    private String encryptstr;

    public String getEncryptStr()
    {
    public String getEncryptStr() {
        String sb = new StringBuilder()
                .append(accesskeyid)
                .append(accesskeysecret)
@@ -53,13 +56,17 @@
        return encryptstr;
    }

    public String getIpPort() {
        return tcpserver + "/" + tcpport;
    }

    @Resource
    private TyClient tyClient;

    private AccessTokenReceiver accessToken;

    public AccessTokenReceiver getAccessToken() {
        if(accessToken != null)
        if (accessToken != null)
            return accessToken;
        // 准备请求的json
        TokenDTO gtd = TokenDTO.builder()

 ard-work/src/main/java/com/ard/tianyu/controller/UavController.java

@@ -3,6 +3,7 @@
import com.ard.common.core.domain.AjaxResult;
import com.ard.tianyu.model.dto.BoxHistoryDTO;
import com.ard.tianyu.model.dto.BoxPhotosDTO;
import com.ard.tianyu.model.dto.BoxSettingsDTO;
import com.ard.tianyu.model.receiver.BoxHistoryReceiver;
import com.ard.tianyu.model.receiver.BoxPhotosReceiver;
import com.ard.tianyu.model.receiver.BoxReceiver;
@@ -62,4 +63,10 @@
        List<BoxPhotosReceiver> boxPhotos = tyService.getBoxPhotos(boxPhotosDTO);
        return AjaxResult.success(boxPhotos);
    }
    @ApiOperation("修改云盒设置")
    @PostMapping("/updateBoxSettings")
    public AjaxResult updateBoxSettings(@RequestBody BoxSettingsDTO boxSettingsDTO) {
        Boolean state = tyService.updateBoxSettings(boxSettingsDTO);
        return AjaxResult.success(state);
    }
}

 ard-work/src/main/java/com/ard/tianyu/http/TyClient.java

@@ -2,6 +2,7 @@

import com.ard.tianyu.model.dto.BoxHistoryDTO;
import com.ard.tianyu.model.dto.BoxPhotosDTO;
import com.ard.tianyu.model.dto.BoxSettingsDTO;
import com.ard.tianyu.model.dto.TokenDTO;
import com.ard.tianyu.model.receiver.*;
import com.dtflys.forest.annotation.*;
@@ -19,19 +20,39 @@
        baseURL = "http://#{tianyu.endpoint}"    // 默认域名
)
public interface TyClient {

    /**
     * 获取云盒TOKEN
     */
    @Post(TyConst.POST_TOKEN)
    TyHttpCommonReceiver<AccessTokenReceiver> getToken(@JSONBody TokenDTO tokenDTO);

    /**
     * 获取云盒列表
     */
    @Get(TyConst.GET_BOX_LIST)
    TyHttpCommonReceiver<List<BoxReceiver>> getBoxList(@Header Map<String, String> headers);

    /**
     * 获取直播地址
     */
    @Get(TyConst.GET_BOX_LIVE_URL)
    TyHttpCommonReceiver<PullStreamUrlReceiver> getBoxLiveUrl(@Var("boxSn") String boxSn, @Header Map<String, String> headers);

    /**
     * 获取飞行历史
     */
    @Post(TyConst.POST_BOX_HISTORY)
    TyHttpCommonReceiver<List<BoxHistoryReceiver>> getBoxHistory(@JSONBody BoxHistoryDTO boxHistoryDTO, @Header Map<String, String> headers);

    /**
     * 获取任务照片
     */
    @Post(TyConst.POST_BOX_PHOTOS)
    TyHttpCommonReceiver<List<BoxPhotosReceiver>> getBoxPhotos(@JSONBody BoxPhotosDTO boxPhotosDTO, @Header Map<String, String> headers);

    /**
     * 修改云盒设置
     */
    @Post(TyConst.POST_BOX_UPDATE)
    TyHttpCommonReceiver updateBoxSettings(@JSONBody BoxSettingsDTO boxSettingsDTO, @Header Map<String, String> headers);
}

 ard-work/src/main/java/com/ard/tianyu/http/TyConst.java

@@ -5,7 +5,7 @@

    public static final String GET_BOX_LIST = "/eapi/box/list";
    public static final String POST_BOX_HISTORY = "/eapi/box/history";
    public static final String POST_BOX_NAME = "/eapi/box/name";
    public static final String POST_BOX_UPDATE = "/eapi/box/update";
    public static final String POST_UPDATE_LIVE = "/eapi/update/live";
    public static final String GET_BOX_LIVE_URL = "/eapi/box/pullUrl/{boxSn}";
    public static final String POST_BOX_PHOTOS = "/eapi/box/photos";

 ard-work/src/main/java/com/ard/tianyu/model/dto/BoxSettingsDTO.java

New file
@@ -0,0 +1,74 @@
package com.ard.tianyu.model.dto;

import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * BoxSettingsDTO
 *
 * @author liusuyi
 * @date 2024年10月22日11:54
 **/
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class BoxSettingsDTO {

    /**
     * 云盒SN编号
     */
    @JsonProperty("boxSn")
    private String boxSn;

    /**
     * 云盒名称(20个字符以内)
     */
    @JsonProperty("boxName")
    private String boxName;

    /**
     * 照片实时回传开关，0-关闭，1-开启
     */
    @JsonProperty("openImgRtp")
    private String openImgRtp;

    /**
     * 视频包排序开关，0-关闭，1-开启（排序会增加视频延时）
     */
    @JsonProperty("openVideoSort")
    private String openVideoSort;

    /**
     * 云端视频存储开关，0-关闭，1-开启
     */
    @JsonProperty("openVideoStorage")
    private String openVideoStorage;

    /**
     * 视频自动推流开关，0-关闭，1-开启
     */
    @JsonProperty("videoAutoPush")
    private String videoAutoPush;

    /**
     * 云盒固件自动升级开关，0-关闭，1-开启
     */
    @JsonProperty("autoUpdate")
    private String autoUpdate;

    /**
     * 云盒与平台失联xx秒后自动返航
     */
    @JsonProperty("lostTime")
    private String lostTime;

    /**
     * 低电量自动返航开关:0-关闭，1-智能计算，xx-指定的返航电量
     */
    @JsonProperty("minHomeBattery")
    private String minHomeBattery;
}

 ard-work/src/main/java/com/ard/tianyu/netty/NettyConfig.java

New file
@@ -0,0 +1,39 @@
package com.ard.tianyu.netty;

import com.ard.tianyu.config.TyConfiguration;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Component
@Slf4j
public class NettyConfig {

    @Resource
    private TyConfiguration tianyuConfiguration;

    @Resource
    private TyNettyTcpClient tyNettyTcpClient;

    @PostConstruct
    public void createTyConnect() {
        try {
            // 将Tcp访问的IP和地址加入到字典
            Map<String,Object> mapParam =new HashMap<String, Object>();
            mapParam.put("IP",tianyuConfiguration.getTcpserver());
            mapParam.put("PORT",tianyuConfiguration.getTcpport());

            //要么从数据库查询信息，要么从配置文件读取信息
            List<Map<String, Object>> list = new ArrayList<>();
            list.add(mapParam);
            tyNettyTcpClient.initAllChannel(list);
        }catch (Exception e) {
            e.printStackTrace();
        }
    }
}

 ard-work/src/main/java/com/ard/tianyu/netty/TyBoxSession.java

New file
@@ -0,0 +1,21 @@
package com.ard.tianyu.netty;

import io.netty.channel.ChannelFuture;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;

@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class TyBoxSession {

    public String subBoxSn;

    public ChannelFuture future;

    public Boolean isReConnect;

}

 ard-work/src/main/java/com/ard/tianyu/netty/TyNettyTcpClient.java

New file
@@ -0,0 +1,103 @@
package com.ard.tianyu.netty;

import com.ard.tianyu.netty.handler.ClientHandler;
import com.ard.tianyu.netty.handler.MessageProtocolDecoder;
import com.ard.tianyu.netty.handler.MessageProtocolEncoder;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

@Component
@Slf4j
public class TyNettyTcpClient {

    public static Map<String, TyBoxSession> channels = new HashMap<>();

    EventLoopGroup eventLoopGroup = new NioEventLoopGroup();

    @Resource
    ClientHandler clientHandler;

    /**
     * 初始化Bootstrap
     */
    public Bootstrap getBootstrap(EventLoopGroup group) {
        if (null == group) {
            group = eventLoopGroup;
        }
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        ChannelPipeline pipeline = socketChannel.pipeline();
                        pipeline.addLast("heart",new IdleStateHandler(0, 3, 0, TimeUnit.SECONDS));
                        pipeline.addLast("encoder",new MessageProtocolEncoder());
                        pipeline.addLast("decoder",new MessageProtocolDecoder());
                        pipeline.addLast("handler",clientHandler);
                    }
                });
        return bootstrap;
    }

    //   获取所有连接
    public void initAllChannel(List<Map<String,Object>> portHosts) {
        for (Map<String, Object> map : portHosts) {
            String host = (String) map.get("IP");
            Integer port = (Integer) map.get("PORT");
            initOneConnection(host,port);
        }
    }

    public void initOneConnection(String host,Integer port){

        log.info("调用一次{}:{}",host,port);
        //判断该账号下，是否有盒子被控制
        //当前正在被控制的盒子
        String currentBoxSn = "";
        if(channels.containsKey(host+"/"+port))
        {
            currentBoxSn = channels.get(host+"/"+port).getSubBoxSn();
            channels.get(host+"/"+port).getFuture().channel().close();
        }

        //异步连接tcp服务端
        Bootstrap bootstrap = getBootstrap(null);
        ChannelFuture future = bootstrap.connect(host,port).addListener((ChannelFuture futureListener) -> {
            if (futureListener.isSuccess()) {
                log.info("连接天宇TCP服务端成功:"+host+":"+port);
            } else {
                final EventLoop eventLoop = futureListener.channel().eventLoop();
                eventLoop.schedule(new Runnable() {
                    @Override
                    public void run() {
                        log.info("连接天宇TCP服务端失败ip:{},port:{},正尝试重连...",host,port);
                        try {
                            initOneConnection(host,port);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }, 5, TimeUnit.SECONDS);
            }
        });

        //判断是否有盒子被订阅
        channels.put(host+"/"+port, TyBoxSession.builder().subBoxSn(currentBoxSn).future(future)
                .isReConnect(!currentBoxSn.equals("")).build());
    }
}

 ard-work/src/main/java/com/ard/tianyu/netty/command/BaseCommand.java

New file
@@ -0,0 +1,14 @@
package com.ard.tianyu.netty.command;

import io.netty.buffer.ByteBuf;
import lombok.Data;

@Data
public abstract class BaseCommand {
    private short dataHeader; // 帧头
    private short dataLength; // 数据长度
    private byte cmdCode; // 指令编号

    // 编码方法，子类需要实现
    public abstract void encode(ByteBuf out);
}

 ard-work/src/main/java/com/ard/tianyu/netty/command/HeartbeatCommand.java

New file
@@ -0,0 +1,22 @@
package com.ard.tianyu.netty.command;

import io.netty.buffer.ByteBuf;


public class HeartbeatCommand extends BaseCommand {

    private final long timestamp;

    public HeartbeatCommand(long timestamp) {
        setDataHeader((short)0x7479);
        setDataLength((short)0x09);
        setCmdCode((byte)0x02);
        this.timestamp = timestamp;
    }
    public void encode(ByteBuf out) {
        out.writeShort(getDataHeader());
        out.writeShort(getDataLength());
        out.writeByte(getCmdCode());
        out.writeLong(timestamp); // 写入时间戳
    }
}

 ard-work/src/main/java/com/ard/tianyu/netty/command/HeartbeatReplyCommand.java

New file
@@ -0,0 +1,26 @@
package com.ard.tianyu.netty.command;

import io.netty.buffer.ByteBuf;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

@Data
@NoArgsConstructor
public class HeartbeatReplyCommand extends BaseCommand {

    private long timestamp; // 时间戳

    public HeartbeatReplyCommand(long timestamp) {
        this.timestamp = timestamp;
        setDataHeader((short) 0x6A77); // 设置帧头
        setCmdCode((byte) 0x02); // 设置指令编号
        setDataLength((short) (13)); // 数据长度 (指令编号 + 时间戳)
    }
    public void encode(ByteBuf out) {
        out.writeShort(getDataHeader());
        out.writeShort(getDataLength());
        out.writeByte(getCmdCode());
        out.writeLong(timestamp); // 写入时间戳
    }
}

 ard-work/src/main/java/com/ard/tianyu/netty/command/PointFlyCommand.java

New file
@@ -0,0 +1,30 @@
package com.ard.tianyu.netty.command;// PointFlyCommand.java
import io.netty.buffer.ByteBuf;

public class PointFlyCommand extends BaseCommand {
    private double longitude;
    private double latitude;
    private float height;
    private float speed;

    public PointFlyCommand(double longitude, double latitude, float height, float speed) {
        this.longitude = longitude;
        this.latitude = latitude;
        this.height = height;
        this.speed = speed;
        setDataHeader((short) 0x7479);
        setCmdCode((byte) 0xD1);
        setDataLength((short) 17); // 数据长度计算
    }

    @Override
    public void encode(ByteBuf out) {
        out.writeShort(getDataHeader());    // 写入帧头
        out.writeShort(getDataLength());    // 写入数据长度
        out.writeByte(getCmdCode());        // 写入指令编号
        out.writeDouble(longitude);         // 写入经度
        out.writeDouble(latitude);          // 写入纬度
        out.writeFloat(height);             // 写入高度
        out.writeFloat(speed);              // 写入速度
    }
}

 ard-work/src/main/java/com/ard/tianyu/netty/command/RegisterCommand.java

New file
@@ -0,0 +1,24 @@
package com.ard.tianyu.netty.command;// RegisterMessage.java
import io.netty.buffer.ByteBuf;

public class RegisterCommand extends BaseCommand {
    private int userId;
    private byte[] accessToken;

    public RegisterCommand(int userId, byte[] accessToken) {
        setDataHeader((short) 0x7479); // 设置帧头
        setDataLength((short) (5 + accessToken.length)); // 数据长度
        setCmdCode((byte) 0x01); // 设置命令编号
        this.userId = userId;
        this.accessToken = accessToken;
    }

    @Override
    public void encode(ByteBuf out) {
        out.writeShort(getDataHeader());    // 写入帧头
        out.writeShort(getDataLength());    // 写入数据长度
        out.writeByte(getCmdCode());        // 写入指令编号
        out.writeInt(userId);               // 写入用户编号
        out.writeBytes(accessToken);        // 写入访问令牌
    }
}

 ard-work/src/main/java/com/ard/tianyu/netty/handler/ClientHandler.java

New file
@@ -0,0 +1,89 @@
package com.ard.tianyu.netty.handler;

import com.ard.tianyu.netty.TyNettyTcpClient;
import com.ard.tianyu.netty.command.BaseCommand;
import com.ard.tianyu.netty.command.HeartbeatCommand;
import com.ard.tianyu.service.ITyService;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;


@Component
@Slf4j
@ChannelHandler.Sharable
public class ClientHandler extends SimpleChannelInboundHandler<BaseCommand> {

    @Autowired
    ITyService tyService;

    @Resource
    private TyNettyTcpClient tyNettyTcpClient;

    @Override
    public void channelRead0(ChannelHandlerContext ctx, BaseCommand command){
        InetSocketAddress ipSocket = (InetSocketAddress) ctx.channel().remoteAddress();
        String host = ipSocket.getHostString();
        int port = ipSocket.getPort();
        //tyService.processMessage(host + "/" + port, msg);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        //当客户端connect成功后，可以在最后一个handler等待用户输入
        log.info("连接成功建立！");
        tyService.registerMessage();
        super.channelActive(ctx);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object obj) {
        if (obj instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) obj;
            if (event.state() == IdleState.READER_IDLE) {
                log.info("客户端读超时");
            } else if (event.state() == IdleState.WRITER_IDLE) {
//                HeartbeatCommand msg = new HeartbeatCommand(System.currentTimeMillis());
//                log.info("发送⼼跳!");//1652345657305  1652345707631
//                ctx.writeAndFlush(msg);
            } else if (event.state() == IdleState.ALL_IDLE) {
                log.info("客户端所有操作超时");
            }
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        InetSocketAddress ipSocket = (InetSocketAddress) ctx.channel().remoteAddress();
        int port = ipSocket.getPort();
        String host = ipSocket.getHostString();
        log.error("与天宇设备" + host + ":" + port + "连接断开!");
        ctx.channel().eventLoop().schedule(new Runnable() {
            @Override
            public void run() {
                log.error("服务端链接不上，开始重连操作...");
                try {
                    tyNettyTcpClient.initOneConnection(host, port);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }, 5, TimeUnit.SECONDS);
        super.channelInactive(ctx);
    }
}

 ard-work/src/main/java/com/ard/tianyu/netty/handler/MessageProtocolDecoder.java

New file
@@ -0,0 +1,51 @@
package com.ard.tianyu.netty.handler;// MessageProtocolDecoder.java

import com.ard.tianyu.netty.command.BaseCommand;
import com.ard.tianyu.netty.command.HeartbeatReplyCommand;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import lombok.extern.slf4j.Slf4j;

import java.util.List;

@Slf4j
public class MessageProtocolDecoder extends ByteToMessageDecoder {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        // 检查可读字节数是否足够
        if (in.readableBytes() < 4) {
            return; // 不足以读取头部
        }

        in.markReaderIndex(); // 标记当前读取位置
        short dataHead = in.readShort();
        if(dataHead != 0x6A77){
            return;
        }
        short dataLength = in.readShort();

        if (in.readableBytes() < dataLength) {
            in.resetReaderIndex(); // 如果不足，重置读取位置
            return;
        }
        byte cmdCode = in.readByte();
        BaseCommand command = null;

        // 根据 cmdCode 创建相应消息对象
        if (cmdCode == 0x01) { // 注册回复消息
            byte success = in.readByte(); // 读取是否成功
            log.info("注册回复: {}", success == 0x01 ? "成功" : "失败");
        }
        if (cmdCode == 0x02) { // 注册回复消息
            long timestamp = in.readLong(); // 读取时间戳
            command = new HeartbeatReplyCommand(timestamp); // 创建心跳回复对象
            log.info("心跳回复: 时间戳 {}", timestamp);
        }
        if (command != null) {
            out.add(command); // 添加解码的命令对象到输出列表
        } else {
           // log.warn("无法识别的命令码: {}", cmdCode);
        }
    }
}

 ard-work/src/main/java/com/ard/tianyu/netty/handler/MessageProtocolEncoder.java

New file
@@ -0,0 +1,16 @@
package com.ard.tianyu.netty.handler;// MessageProtocolEncoder.java

import com.ard.tianyu.netty.command.BaseCommand;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;

public class MessageProtocolEncoder extends MessageToByteEncoder<BaseCommand> {
    @Override
    protected void encode(ChannelHandlerContext ctx, BaseCommand command, ByteBuf out) throws Exception {
        out.writeShort(command.getDataHeader()); // 写入帧头
        out.writeShort(command.getDataLength()); // 写入数据长度
        out.writeByte(command.getCmdCode()); // 写入指令编号
        command.encode(out); // 编码具体负载
    }
}

 ard-work/src/main/java/com/ard/tianyu/service/ITyService.java

@@ -2,12 +2,11 @@

import com.ard.tianyu.model.dto.BoxHistoryDTO;
import com.ard.tianyu.model.dto.BoxPhotosDTO;
import com.ard.tianyu.model.dto.BoxSettingsDTO;
import com.ard.tianyu.model.receiver.*;
import com.dtflys.forest.annotation.Header;
import com.dtflys.forest.annotation.Var;

import java.util.List;
import java.util.Map;

/**
 * ITyService
@@ -48,4 +47,15 @@
     */
    List<BoxPhotosReceiver> getBoxPhotos(BoxPhotosDTO boxPhotosDTO);

    /**
     * 修改云盒设置
     * @return
     */
    Boolean updateBoxSettings(BoxSettingsDTO boxSettingsDTO);

    /**
     * 注册
     * @return
     */
    void registerMessage();
}

 ard-work/src/main/java/com/ard/tianyu/service/impl/TyServiceImpl.java

@@ -4,12 +4,23 @@
import com.ard.tianyu.http.TyClient;
import com.ard.tianyu.model.dto.BoxHistoryDTO;
import com.ard.tianyu.model.dto.BoxPhotosDTO;
import com.ard.tianyu.model.dto.BoxSettingsDTO;
import com.ard.tianyu.model.receiver.*;
import com.ard.tianyu.netty.command.BaseCommand;
import com.ard.tianyu.netty.command.HeartbeatCommand;
import com.ard.tianyu.netty.command.RegisterCommand;
import com.ard.tianyu.netty.TyNettyTcpClient;
import com.ard.tianyu.service.ITyService;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Collections;
import java.net.InetSocketAddress;
import java.util.List;

/**
@@ -18,11 +29,13 @@
 * @author liusuyi
 * @date 2024年10月22日8:37
 **/
@Slf4j
@Service
public class TyServiceImpl implements ITyService {

    @Resource
    private TyClient tyClient;

    @Resource
    private TyConfiguration tyConfiguration;

@@ -81,4 +94,57 @@
        TyHttpCommonReceiver<List<BoxPhotosReceiver>> boxPhotos = tyClient.getBoxPhotos(boxPhotosDTO, tyConfiguration.getAccessToken().convertFromDTO());
        return boxPhotos.getContent();
    }
}

    /**
     * 修改云盒设置
     */
    @Override
    public Boolean updateBoxSettings(BoxSettingsDTO boxSettingsDTO) {
        TyHttpCommonReceiver<String> result = tyClient.updateBoxSettings(boxSettingsDTO, tyConfiguration.getAccessToken().convertFromDTO());
        return result.getState() == 1;
    }


    @Override
    public void registerMessage() {
        AccessTokenReceiver tyToken = tyConfiguration.getAccessToken();
        RegisterCommand registerMessage = new RegisterCommand(tyToken.getCompanyId(), tyToken.getAccessToken().getBytes());
        this.sendMessage(tyConfiguration.getIpPort(), registerMessage);
    }

    public boolean sendMessage(String key, BaseCommand msg) {
        //将从用户接收的信息发送到天宇 commandCache
        boolean cmdResult = false;
        try {
            ChannelFuture future = TyNettyTcpClient.channels.get(key).getFuture();
            Channel channel = future.channel();
            if (future != null && channel.isActive()) {
                InetSocketAddress ipSocket = (InetSocketAddress) channel.remoteAddress();
                String host = ipSocket.getHostString();
                int port = ipSocket.getPort();
                // 创建 ByteBuf 并编码消息
                ByteBuf buffer = Unpooled.buffer();
                msg.encode(buffer); // 确保对消息进行编码

                channel.writeAndFlush(buffer); // 发送编码后的数据
                cmdResult = true;
                log.info("向天宇服务器" + host + ":" + port + "发送数据" + "，命令码：" + msg.getCmdCode());
            } else {
                log.error("与天宇服务器链接断开，无法发送数据！");
            }
        } catch (Exception e) {
            log.info("Failed to publish the message. {}", "转发信息");
            e.printStackTrace();
        } finally {
            return cmdResult;
        }
    }

    @Scheduled(fixedRate = 5000)
    public void sendHeartBeat() {
        long timestamp = System.currentTimeMillis();
        log.debug("定时向天宇发送⼼跳,时间戳{}", timestamp);
        HeartbeatCommand heartbeatMessage = new HeartbeatCommand(timestamp);
        this.sendMessage(tyConfiguration.getIpPort(), heartbeatMessage);
    }
}

 ard-work/src/main/java/com/ard/vtdu/client/MediaClient.java

New file
@@ -0,0 +1,123 @@
package com.ard.vtdu.client;

import com.ard.vtdu.domain.*;
import com.dtflys.forest.annotation.*;

/**
 * @Description: mediamtx流媒体客户端
 * @ClassName: client
 * @Author: 刘苏义
 * @Date: 2023年07月06日9:51
 * @Version: 1.0
 **/
@BaseRequest(baseURL = "http://#{mediamtx.host}:9997")
public interface MediaClient {
    /**
     * 增加路径配置
     */
    @Post("/v3/config/paths/add/{name}")
    public String addPath(@Var("name") String name, @JSONBody Conf body);

    /**
     * 修改路径配置
     */
    @Patch("/v3/config/paths/patch/{name}")
    public String editPath(@Var("name") String name, @JSONBody Conf body);

    /**
     * 移除路径配置
     */
    @Delete("/v3/config/paths/delete/{name}")
    public String removePath(@Var("name") String name);

    /**
     * 获取路径配置详情
     */
    @Get(url = "/v3/config/paths/get/{name}")
    public Conf getPathConfig(@Var("name") String name);

    /**
     * 获取路径配置列表
     */
    @Get(url = "/v3/config/paths/list?page={pageNum}&itemsPerPage={pageSize}")
    public PathConfig getPathConfigs(@Var("pageNum") Integer pageNum, @Var("pageSize") Integer pageSize);

    /**
     * 查询所有路径
     */
    @Get("/v3/paths/list?page={pageNum}&itemsPerPage={pageSize}")
    public Paths paths(@Var("pageNum") Integer pageNum, @Var("pageSize") Integer pageSize);

    /**
     * 查询所有rtsp会话
     */
    @Get("/v3/rtspsessions/list?page={pageNum}&itemsPerPage={pageSize}")
    public RtspSessions rtspsessions(@Var("pageNum") Integer pageNum, @Var("pageSize") Integer pageSize);

    /**
     * 查询所有webrt会话
     */
    @Get("/v3/webrtcsessions/list?page={pageNum}&itemsPerPage={pageSize}")
    public WebrtcSessions webrtcsessions(@Var("pageNum") Integer pageNum, @Var("pageSize") Integer pageSize);

    /**
     * 查询所有rtmp会话
     */
    @Get("/v3/rtmpconns/list?page={pageNum}&itemsPerPage={pageSize}")
    public RtmpSessions rtmpsessions(@Var("pageNum") Integer pageNum, @Var("pageSize") Integer pageSize);

    /**
     * 按sessionId查询rtsp会话
     */
    @Get("/v3/rtspsessions/get/{sessionId}")
    public RtspSession getRtspsessionById(@Var("sessionId") String sessionId);

    /**
     * 按sessionId查询webrtc会话
     */
    @Get("/v3/webrtcsessions/get/{sessionId}")
    public WebrtcSession getWebrtcsessionById(@Var("sessionId") String sessionId);

    /**
     * 按sessionId查询rtmp会话
     */
    @Get("/v3/rtmpconns/get/{sessionId}")
    public RtmpSession getRtmpsessionById(@Var("sessionId") String sessionId);

    /**
     * 按sessionId删除rtsp会话
     */
    @Post("/v3/rtspsessions/kick/{sessionId}")
    public String kickRtspSessions(@Var("sessionId") String sessionId);

    /**
     * 按sessionId删除rtmp连接
     */
    @Post("/v3/rtmpconns/kick/{sessionId}")
    public String kickRtmpSessions(@Var("sessionId") String sessionId);

    /**
     * 按sessionId删除webrtc会话
     */
    @Post("/v3/webrtcsessions/kick/{sessionId}")
    public String kickWebrtcSessions(@Var("sessionId") String sessionId);

    /**
     * 录像分段列表
     */
    @Get("/v3/recordings/list/")
    public Paths recordingList(@Query("pageNum") Integer pageNum, @Query("pageSize") Integer pageSize);

    /**
     * 通过路径名获取录像分段列表
     */
    @Get("/v3/recordings/get/{name}")
    public Item recordingListByName(@Var("name") String name);

    /**
     * 通过路径名和分段录像开始时间删除录像分段
     */
    @Delete("/v3/recordings/deletesegment")
    public String deleteSegment(@Query("path") String name, @Query("start") String start);

}

 ard-work/src/main/java/com/ard/vtdu/controller/MediaController.java

New file
@@ -0,0 +1,216 @@
package com.ard.vtdu.controller;
import com.ard.common.core.controller.BaseController;
import com.ard.common.core.domain.AjaxResult;
import com.ard.common.core.page.TableDataInfo;
import com.ard.vtdu.domain.ArdVtdu;
import com.ard.vtdu.domain.Conf;
import com.ard.vtdu.service.IMediaService;
import com.github.xiaoymin.knife4j.annotations.ApiOperationSupport;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.*;
import javax.annotation.Resource;
import java.util.Arrays;

/**
 * @Description: mediaMTX流媒体业务
 * @ClassName: controller
 * @Author: 刘苏义
 * @Date: 2023年07月13日9:26
 * @Version: 1.0
 **/
@RestController
@Api(tags = "mediaMTX接口")
@RequestMapping("/mediamtx/media")
public class MediaController extends BaseController {

    @Resource
    private IMediaService mediaService;

    /**
     * 增加通道
     */
    @PostMapping()
    @ApiOperation("增加通道")
    @PreAuthorize("@ss.hasPermi('mediamtx:media:add')")
    @ApiOperationSupport(includeParameters = {"ardVtdu.name", "ardVtdu.rtspSource", "ardVtdu.mode", "ardVtdu.isCode", "ardVtdu.isRecord"}, order = 1)
    public AjaxResult addPath(@RequestBody ArdVtdu ardVtdu) {
        if (StringUtils.isEmpty(ardVtdu.getName())) {
            return AjaxResult.error("通道名称不能为空");
        }
        if (StringUtils.isEmpty(ardVtdu.getRtspSource())) {
            return AjaxResult.error("rtsp地址不能为空");
        }
        return AjaxResult.success(mediaService.addPath(ardVtdu));
    }


    /**
     * 修改通道
     */
    @ApiOperation("修改通道")
    @ApiOperationSupport(includeParameters = {"ardVtdu.name", "ardVtdu.rtspSource", "ardVtdu.mode", "ardVtdu.isCode", "ardVtdu.isRecord"}, order = 2)
    @PreAuthorize("@ss.hasPermi('mediamtx:media:edit')")
    @PutMapping
    public AjaxResult edit(@RequestBody ArdVtdu ardVtdu) {
        return AjaxResult.success(mediaService.editPath(ardVtdu));
    }

    /**
     * 移除通道
     */
    @DeleteMapping("/path/{names}")
    @PreAuthorize("@ss.hasPermi('mediamtx:media:remove')")
    @ApiOperation("移除通道")
    @ApiOperationSupport(order = 3)

    public AjaxResult removePath(@PathVariable String[] names) {
        mediaService.removePath(names);
        return AjaxResult.success();
    }

    /**
     * 移除Webrtc拉流
     */
    @ApiOperation("移除Webrtc拉流")
    @ApiOperationSupport(order = 4)
    @DeleteMapping("/webrtc/{sessionIds}")
    public AjaxResult removeWebrtcPullStreamSession(@PathVariable String[] sessionIds) {
        Arrays.stream(sessionIds).forEach(sessionId -> mediaService.kickWebrtcSession(sessionId));
        return AjaxResult.success();
    }

    /**
     * 移除Rtmp拉流
     */
    @ApiOperation("移除Rtmp拉流")
    @ApiOperationSupport(order = 4)
    @DeleteMapping("/rtmp/{sessionIds}")
    public AjaxResult removeRtmpPullStreamSession(@PathVariable String[] sessionIds) {
        Arrays.stream(sessionIds).forEach(sessionId -> mediaService.kickRtmpSession(sessionId));
        return AjaxResult.success();
    }

    /**
     * 移除Rtsp拉流
     */
    @ApiOperation("移除Rtsp拉流")
    @ApiOperationSupport(order = 4)
    @DeleteMapping("/rtsp/{sessionIds}")
    public AjaxResult removeRtspPullStreamSession(@PathVariable String[] sessionIds) {
        Arrays.stream(sessionIds).forEach(sessionId -> mediaService.kickRtspSession(sessionId));
        return AjaxResult.success();
    }


    /**
     * 获取当前通道列表
     */
    @GetMapping("/path/list")
    @ApiOperation("获取当前通道列表")
    @ApiOperationSupport(order = 5)
    public TableDataInfo getPaths(Integer pageNum, Integer pageSize) {
        startPage();
        return getDataTable(mediaService.paths(pageNum, pageSize));
    }

    /**
     * 按ID查询Webrtc拉流详情
     */
    @GetMapping("/getWebrtcSessionById")
    @ApiOperation("按ID查询Webrtc拉流详情")
    public AjaxResult getWebrtcSessionById(String sessionId) {
        return AjaxResult.success(mediaService.getWebrtcSessionById(sessionId));
    }

    /**
     * 获取推流列表
     */

    @PreAuthorize("@ss.hasPermi('mediamtx:media:list')")
    @GetMapping("/pushList")
    @ApiOperation("获取推流列表")
    @ApiOperationSupport(order = 6)
    public TableDataInfo getPushStreamList(Integer pageNum, Integer pageSize) {
        TableDataInfo tableDataInfo = mediaService.getPushStreamList(pageNum, pageSize);
        return tableDataInfo;
    }

    /**
     * 获取WebRtc拉流列表
     */

    @GetMapping("/getPullWebrtcStreamList")
    @ApiOperation("获取WebRtc拉流列表")
    @ApiOperationSupport(order = 7)
    public TableDataInfo getPullWebrtcStreamList(Integer pageNum, Integer pageSize) {
        return mediaService.getPullWebrtcStreamList(pageNum, pageSize);
    }

    /**
     * 获取rtsp拉流列表
     */

    @GetMapping("/getPullRtspStreamList")
    @ApiOperation("获取rtsp拉流列表")
    @ApiOperationSupport(order = 8)
    public TableDataInfo getPullRtspStreamList(Integer pageNum, Integer pageSize) {
        return mediaService.getPullRtspStreamList(pageNum, pageSize);

    }

    /**
     * 获取rtmp拉流列表
     */
    @GetMapping("/getPullRtmpStreamList")
    @ApiOperation("获取rtmp拉流列表")
    @ApiOperationSupport(order = 9)
    public TableDataInfo getPullRtmpStreamList(Integer pageNum, Integer pageSize) {
        return mediaService.getPullRtmpStreamList(pageNum, pageSize);
    }

    //配置录像
    @PostMapping("/setRecord")
    @ApiOperation("配置录像")
    @ApiOperationSupport(includeParameters = {"conf.name", "conf.sourceOnDemand", "conf.record"})
    public AjaxResult setRecord(@RequestBody Conf conf) {
        return AjaxResult.success(mediaService.setRecord(conf));
    }

    //获取全部录像分段列表
    @GetMapping("/getRecordingList")
    @ApiOperation("获取全部录像分段列表")
    public AjaxResult getRecordingList(Integer pageNum, Integer pageSize) {
        return AjaxResult.success(mediaService.recordingList(pageNum, pageSize));
    }

    //通过路径名获取录像分段列表
    @GetMapping("/getRecordingListByName/{name}")
    @ApiOperation("通过路径名获取录像分段列表")
    public AjaxResult getRecordingListByName(@PathVariable String name) {
        return AjaxResult.success(mediaService.recordingListByName(name));
    }

    //删除录像
    @DeleteMapping("/deleteRecording")
    @ApiOperation("删除录像")
    public AjaxResult deleteRecording(String name, String start) {
        return AjaxResult.success(mediaService.deleteSegment(name, start));
    }

    //获取通道配置
    @GetMapping("/{name}")
    @ApiOperation("获取通道配置")
    public AjaxResult getPathConfig(@PathVariable String name) {
        return AjaxResult.success(mediaService.getPathConfig(name));
    }

    //获取所有通道配置列表
    @GetMapping("/getRecordConfList")
    @ApiOperation("获取所有通道配置列表")
    public AjaxResult getRecordConfList(Integer pageNum, Integer pageSize) {
        return AjaxResult.success(mediaService.getPathConfigs(pageNum, pageSize));
    }
}

 ard-work/src/main/java/com/ard/vtdu/domain/ArdVtdu.java

New file
@@ -0,0 +1,74 @@
package com.ard.vtdu.domain;

import com.ard.common.annotation.Excel;
import com.ard.common.core.domain.BaseEntity;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * 流媒体管理对象 vtdu
 *
 * @author ard
 * @date 2023-08-29
 */
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class ArdVtdu extends BaseEntity {
    private static final long serialVersionUID = 1L;

    /**
     * 主键名称
     */
    @Excel(name = "名称")
    private String name;

    /**
     * rtsp源地址
     */
    @Excel(name = "rtsp源地址")
    private String rtspSource;

    /**
     * 是否转码
     * 0：不转码 1：转码
     */
    @Excel(name = "是否转码")
    @Builder.Default
    private Integer isCode = 0;

    /**
     * 编解码模式
     * 0：cpu 1：gpu
     */
    @Excel(name = "编解码模式")
    @Builder.Default
    private Integer mode = 0;
    /**
     * 是否录像
     * 0：不录像 1：录像
     */
    @Excel(name = "是否录像")
    @Builder.Default
    private Integer isRecord = 0;
    /**
     * rtsp播放地址
     */
    String rtspUrl;
    /**
     * rtmp播放地址
     */
    String rtmpUrl;
    /**
     * webrtc播放地址
     */
    String webrtcUrl;
    /**
     * 相机ID
     */
    String cameraId;

}

 ard-work/src/main/java/com/ard/vtdu/domain/Conf.java

New file
@@ -0,0 +1,46 @@
package com.ard.vtdu.domain;

import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * @Description:
 * @ClassName: MediaInfo
 * @Author: 刘苏义
 * @Date: 2023年07月13日9:32
 * @Version: 1.0
 **/
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class Conf {
    private String name;
    private String source;
    //是否按需发布
    private Boolean sourceOnDemand;
    private String sourceProtocol;
    private Integer maxReaders;
    private String runOnDemand;
    private Boolean runOnDemandRestart;
    private String runOnDemandStartTimeout;
    private String runOnDemandCloseAfter;
    private String runOnInit;
    private Boolean runOnInitRestart;
    //录像
    @Builder.Default
    private Boolean record = false;
    @Builder.Default
    private String recordPath = "./recordings/%path/%Y-%m-%d_%H-%M-%S-%f";
    @Builder.Default
    private String recordFormat = "fmp4";
    @Builder.Default
    private String recordPartDuration = "1s";
    @Builder.Default
    private String recordSegmentDuration = "0h5m0s";
    @Builder.Default
    private String recordDeleteAfter = "24h0m0s";

}

 ard-work/src/main/java/com/ard/vtdu/domain/Item.java

New file
@@ -0,0 +1,29 @@
package com.ard.vtdu.domain;

import com.fasterxml.jackson.annotation.JsonInclude;
import lombok.Data;

import java.util.Date;
import java.util.List;

/**
 * @Description:
 * @ClassName: Items
 * @Author: 刘苏义
 * @Date: 2023年07月13日10:23
 * @Version: 1.0
 **/
@Data
@JsonInclude(value = JsonInclude.Include.NON_NULL)
public class Item {
    private String name;
    private String confName;
    private Source source;
    private List<Readers> readers;
    private List<String> tracks;
    private Long bytesReceived;
    private Long bytesSent;
    private Date readyTime;
    private List<Segment> segments;
}


 ard-work/src/main/java/com/ard/vtdu/domain/PathConfig.java

New file
@@ -0,0 +1,20 @@
package com.ard.vtdu.domain;

import lombok.Data;

import java.util.List;

/**
 * ConfigPath
 *
 * @author liusuyi
 * @date 2024年09月26日10:04
 **/
@Data
public class PathConfig {
    private int itemCount;

    private int pageCount;

    private List<Conf> items;
}

 ard-work/src/main/java/com/ard/vtdu/domain/Paths.java

New file
@@ -0,0 +1,23 @@
package com.ard.vtdu.domain;

import lombok.Data;

import java.util.List;

/**
 * @Description: returns all paths.
 * @ClassName: Paths
 * @Author: 刘苏义
 * @Date: 2023年07月13日10:06
 * @Version: 1.0
 **/
@Data
public class Paths {

    private int itemCount;

    private int pageCount;

    private List<Item> items;

}

 ard-work/src/main/java/com/ard/vtdu/domain/Readers.java

New file
@@ -0,0 +1,16 @@
package com.ard.vtdu.domain;

import lombok.Data;

/**
 * @ClassName Readers
 * @Description:
 * @Author 刘苏义
 * @Date 2023/7/14 21:04
 * @Version 1.0
 */
@Data
public class Readers {
    private String type;
    private String id;
}

 ard-work/src/main/java/com/ard/vtdu/domain/RtmpSession.java

New file
@@ -0,0 +1,27 @@
package com.ard.vtdu.domain;

import com.fasterxml.jackson.annotation.JsonFormat;
import lombok.Data;

import java.util.Date;

/**
 * @Description:
 * @ClassName: WebrtcSession
 * @Author: 刘苏义
 * @Date: 2023年08月26日10:16:21
 **/
@Data
public class RtmpSession {
    private String path;
    private String id;
    private Date created;
    private String remoteAddr;
    private String state;
    private long bytesReceived;
    private long bytesSent;
    private String upStream;
    private String downStream;
    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss.SSS")
    private Date beginTime;
}

 ard-work/src/main/java/com/ard/vtdu/domain/RtmpSessions.java

New file
@@ -0,0 +1,21 @@
package com.ard.vtdu.domain;

import lombok.Data;

import java.util.List;

/**
 * @ClassName:Sessions
 * @Description:
 * @Author:ard
 * @Date:2024年07月12日13:34
 * @Version:1.0
 **/
@Data
public class RtmpSessions {
    private int itemCount;

    private int pageCount;

    private List<RtmpSession> items;
}

 ard-work/src/main/java/com/ard/vtdu/domain/RtspSession.java

New file
@@ -0,0 +1,29 @@
package com.ard.vtdu.domain;

import com.fasterxml.jackson.annotation.JsonFormat;
import lombok.Data;

import java.util.Date;

/**
 * @ClassName RtspSession
 * @Description:
 * @Author 刘苏义
 * @Date 2023/7/14 21:19
 * @Version 1.0
 */
@Data
public class RtspSession {
    private String id;
    private String path;
    private Date created;
    private String remoteAddr;
    private String state;
    private String transport;
    private long bytesReceived;
    private long bytesSent;
    private String upStream;
    private String downStream;
    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
    private Date beginTime;
}

 ard-work/src/main/java/com/ard/vtdu/domain/RtspSessions.java

New file
@@ -0,0 +1,21 @@
package com.ard.vtdu.domain;

import lombok.Data;

import java.util.List;

/**
 * @ClassName:Sessions
 * @Description:
 * @Author:ard
 * @Date:2024年07月12日13:34
 * @Version:1.0
 **/
@Data
public class RtspSessions {
    private int itemCount;

    private int pageCount;

    private List<RtspSession> items;
}

 ard-work/src/main/java/com/ard/vtdu/domain/Segment.java

New file
@@ -0,0 +1,22 @@
package com.ard.vtdu.domain;

import com.fasterxml.jackson.annotation.JsonFormat;
import lombok.Data;

import java.util.Date;

/**
 * Segment
 *
 * @author liusuyi
 * @date 2024年09月24日13:26
 **/
@Data
public class Segment {
    private String start;
    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss.SSS")
    private Date startTime;


}


 ard-work/src/main/java/com/ard/vtdu/domain/Source.java

New file
@@ -0,0 +1,16 @@
package com.ard.vtdu.domain;

import lombok.Data;

/**
 * @Description:
 * @ClassName: Source
 * @Author: 刘苏义
 * @Date: 2023年07月13日10:24
 * @Version: 1.0
 **/
@Data
public class Source {
    private String type;
    private String id;
}

 ard-work/src/main/java/com/ard/vtdu/domain/StreamInfo.java

New file
@@ -0,0 +1,85 @@
package com.ard.vtdu.domain;

import com.fasterxml.jackson.annotation.JsonFormat;
import com.fasterxml.jackson.annotation.JsonInclude;
import lombok.Data;

import java.util.Date;

/**
 * @Description:流信息
 * @ClassName: PushStreamInfo
 * @Author: 刘苏义
 * @Date: 2023年07月15日8:40:37
 * @Version: 1.0
 **/
@Data
@JsonInclude(value = JsonInclude.Include.NON_NULL)
public class StreamInfo {
    /**
     * 会话ID
     */
    String id;
    /**
     * 路径名称
     */
    String name;
    /**
     * 源rtsp地址
     */
    String rtspSource;
    /**
     * rtsp播放地址
     */
    String rtspUrl;
    /**
     * rtmp播放地址
     */
    String rtmpUrl;
    /**
     * webrtc播放地址
     */
    String webrtcUrl;
    /**
     * 传输方式
     */
    String protocol;
    /**
     * 上行流量
     */
    String upTraffic;
    /**
     * 下行流量
     */
    String downTraffic;
    /**
     * 拉流数量
     */
    Integer num;
    /**
     * 开始时间
     */
    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
    Date beginTime;
    /**
     * 远程主机
     */
    String remoteAddr;

    /**
     * 解码模式
     * 0-gpu硬解码 1-cpu软解码
     */
    Integer mode;

    /**
     * 是否转码
     * 0-不转码 1-转码
     */
    Integer isCode;
    /**
     * 媒体类型（webRTCSession/rtspSession）
     */
    String sessionType;

}

 ard-work/src/main/java/com/ard/vtdu/domain/WebrtcSession.java

New file
@@ -0,0 +1,31 @@
package com.ard.vtdu.domain;

import com.fasterxml.jackson.annotation.JsonFormat;
import lombok.Data;

import java.util.Date;

/**
 * @Description:
 * @ClassName: WebrtcSession
 * @Author: 刘苏义
 * @Date: 2023年08月26日10:16:21
 **/
@Data
public class WebrtcSession {
    private String id;
    private String path;
    private Date created;
    private String remoteAddr;
    private String peerConnectionEstablished;
    private String localCandidate;
    private String remoteCandidate;
    private String state;
    private long bytesReceived;
    private long bytesSent;
    private String upStream;
    private String downStream;
    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
    private Date beginTime;
}


 ard-work/src/main/java/com/ard/vtdu/domain/WebrtcSessions.java

New file
@@ -0,0 +1,21 @@
package com.ard.vtdu.domain;

import lombok.Data;

import java.util.List;

/**
 * @ClassName:Sessions
 * @Description:
 * @Author:ard
 * @Date:2024年07月12日13:34
 * @Version:1.0
 **/
@Data
public class WebrtcSessions {
    private int itemCount;

    private int pageCount;

    private List<WebrtcSession> items;
}

 ard-work/src/main/java/com/ard/vtdu/service/IMediaService.java

New file
@@ -0,0 +1,69 @@
package com.ard.vtdu.service;

import com.ard.common.core.page.TableDataInfo;
import com.ard.vtdu.domain.*;
import java.util.List;

public interface IMediaService {
    /**
     * 增加路径
     */
    ArdVtdu addPath(ArdVtdu ardVtdu);

    /**
     * 修改路径
     */
    ArdVtdu editPath(ArdVtdu ardVtdu);

    //获取通道配置
    Conf getPathConfig(String name);

    //获取通道配置列表
    PathConfig getPathConfigs(Integer pageNum, Integer pageSize);

    void removePath(String[] names);

    void removePath(String name);

    List<StreamInfo> paths(Integer pageNum, Integer pageSize);

    List<String> getNameList();

    boolean checkNameExist(String name);

    RtspSession getRtspSessionById(String sessionId);

    WebrtcSession getWebrtcSessionById(String sessionId);

    RtmpSession getRtmpSessionById(String sessionId);

    TableDataInfo getPushStreamList(Integer pageNum, Integer pageSize);

    TableDataInfo getPullWebrtcStreamList(Integer pageNum, Integer pageSize);

    TableDataInfo getPullRtspStreamList(Integer pageNum, Integer pageSize);

    TableDataInfo getPullRtmpStreamList(Integer pageNum, Integer pageSize);

    Boolean kickRtspSession(String sessionId);

    Boolean kickRtmpSession(String sessionId);

    Boolean kickWebrtcSession(String sessionId);

    void clearAll();

    //开启录像
    String setRecord(Conf conf);

    //获取全部录像分段列表
    Paths recordingList(Integer pageNum, Integer pageSize);

    //通过路径名获取录像分段列表
    Item recordingListByName(String name);

    //通过路径名和分段录像开始时间删除录像分段
    String deleteSegment(String name, String start);


}

 ard-work/src/main/java/com/ard/vtdu/service/impl/MediaServiceImpl.java

New file
@@ -0,0 +1,669 @@
package com.ard.vtdu.service.impl;

import com.ard.common.constant.HttpStatus;
import com.ard.common.core.page.TableDataInfo;
import com.ard.tianyu.model.receiver.BoxReceiver;
import com.ard.tianyu.model.receiver.PullStreamUrlReceiver;
import com.ard.tianyu.service.ITyService;
import com.ard.vtdu.client.MediaClient;
import com.ard.vtdu.domain.*;
import com.ard.vtdu.service.IMediaService;
import com.ard.vtdu.util.ArdTool;
import com.dtflys.forest.exceptions.ForestRuntimeException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.OffsetDateTime;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

/**
 * @Description: 流媒体业务
 * @ClassName: MediaService
 * @Author: 刘苏义
 * @Date: 2023年07月13日9:28
 * @Version: 1.0
 **/
@Service
@Slf4j
public class MediaServiceImpl implements IMediaService {

    @Resource
    private MediaClient mediaClient;
    @Resource
    private ITyService tyService;
    @Value("${mediamtx.host}")
    private String mediamtxHost;


    @PostConstruct
    public void initMediaMTX() {
        //清空mediamtx流媒体数据
        clearAll();
    }


    /**
     * @return
     * @Author 刘苏义
     * @Description 添加流媒体
     * name 相机ID
     * sourceUrl rtsp推流地址
     * isCode 0-不转码 1-转码
     * mode 1-gpu硬解码 0-cpu软解码
     * @Date 2024/7/3 8:45
     * @Param
     */
    @Override
    public ArdVtdu addPath(ArdVtdu ardVtdu) {
        try {
            String rtspUrl = "rtsp://" + mediamtxHost + ":8554/" + ardVtdu.getName();
            String rtmpUrl = "rtmp://" + mediamtxHost + ":1935/" + ardVtdu.getName();
            String webrtcUrl = "http://" + mediamtxHost + ":8889/" + ardVtdu.getName();

            Conf conf = new Conf();
            boolean isCodeEnabled = ardVtdu.getIsCode().equals(1);
            boolean isModeEabled = ardVtdu.getMode().equals(1);
            boolean isRecordEnabled = ardVtdu.getIsRecord().equals(1);
            if (isCodeEnabled) {
                conf.setSource("publisher");
                //默认软解码
                String cmd = "ffmpeg -rtsp_transport tcp -i " + ardVtdu.getRtspSource() + " -vcodec libx264 -preset ultrafast -r 25 -keyint_min 25 -g 60 -sc_threshold 0 -threads 6  -b:v 2048k -acodec opus  -strict -2 -f rtsp rtsp://localhost:$RTSP_PORT/$MTX_PATH";
                if (isModeEabled) {//硬解码
                    cmd = "ffmpeg -hwaccel cuvid -c:v hevc_cuvid  -rtsp_transport tcp  -i " + ardVtdu.getRtspSource() + " -c:v h264_nvenc  -r 25 -g 60 -sc_threshold 0 -threads 6  -b:v 2048k -bf 0 -acodec opus  -strict -2 -f rtsp rtsp://localhost:$RTSP_PORT/$MTX_PATH";
                }
                if (isRecordEnabled) {
                    //录像
                    conf.setRunOnInit(cmd);
                    conf.setRunOnInitRestart(true);
                    conf.setRunOnDemand("");
                    conf.setRecord(true);
                } else {
                    conf.setRunOnDemand(cmd);
                    conf.setRunOnDemandRestart(true);
                    conf.setRunOnDemandCloseAfter("5s");
                    conf.setRecord(false);
                    conf.setRunOnInit("");
                }
            } else {
                conf.setSource(ardVtdu.getRtspSource());
                conf.setRunOnDemandRestart(false);
                conf.setRunOnDemandCloseAfter("5s");
                conf.setSourceOnDemand(!isRecordEnabled);
                conf.setRecord(isRecordEnabled);
            }
            conf.setMaxReaders(100);
            conf.setSourceProtocol("tcp");

            if (!checkNameExist(ardVtdu.getName())) {
                mediaClient.addPath(ardVtdu.getName(), conf);
            } else {
                mediaClient.editPath(ardVtdu.getName(), conf);
            }
            ardVtdu.setCameraId(ardVtdu.getName().split("_")[0]);
            ardVtdu.setRtspUrl(rtspUrl);
            ardVtdu.setRtmpUrl(rtmpUrl);
            ardVtdu.setWebrtcUrl(webrtcUrl);
            return ardVtdu;
        } catch (ForestRuntimeException ex) {
            log.error("添加流媒体异常：" + ex.getMessage());
        }
        return null;
    }

    @Override
    public ArdVtdu editPath(ArdVtdu ardVtdu) {
        try {
            String rtspUrl = "rtsp://" + mediamtxHost + ":8554/" + ardVtdu.getName();
            String rtmpUrl = "rtmp://" + mediamtxHost + ":1935/" + ardVtdu.getName();
            String webrtcUrl = "http://" + mediamtxHost + ":8889/" + ardVtdu.getName();

            Conf conf = new Conf();
            boolean isCodeEnabled = ardVtdu.getIsCode().equals(1);
            boolean isModeEabled = ardVtdu.getMode().equals(1);
            boolean isRecordEnabled = ardVtdu.getIsRecord().equals(1);
            if (isCodeEnabled) {
                conf.setSource("publisher");
                //默认软解码
                String cmd = "ffmpeg -rtsp_transport tcp -i " + ardVtdu.getRtspSource() + " -vcodec libx264 -preset ultrafast -r 25 -keyint_min 25 -g 60 -sc_threshold 0 -threads 6  -b:v 2048k -acodec opus  -strict -2 -f rtsp rtsp://localhost:$RTSP_PORT/$MTX_PATH";
                if (isModeEabled) {//硬解码
                    cmd = "ffmpeg -hwaccel cuvid -c:v hevc_cuvid  -rtsp_transport tcp  -i " + ardVtdu.getRtspSource() + " -c:v h264_nvenc  -r 25 -g 60 -sc_threshold 0 -threads 6  -b:v 2048k -bf 0 -acodec opus  -strict -2 -f rtsp rtsp://localhost:$RTSP_PORT/$MTX_PATH";
                }
                if (isRecordEnabled) {
                    //录像
                    conf.setRunOnInit(cmd);
                    conf.setRunOnInitRestart(true);
                    conf.setRunOnDemand("");
                    conf.setRecord(true);
                } else {
                    conf.setRunOnDemand(cmd);
                    conf.setRunOnDemandRestart(true);
                    conf.setRunOnDemandCloseAfter("5s");
                    conf.setRecord(false);
                    conf.setRunOnInit("");
                }
            } else {
                conf.setSource(ardVtdu.getRtspSource());
                conf.setRunOnDemand("");
                conf.setRunOnDemandRestart(false);
                conf.setRunOnDemandCloseAfter("5s");
                conf.setSourceOnDemand(!isRecordEnabled);
                conf.setRecord(isRecordEnabled);
            }
            conf.setMaxReaders(100);
            conf.setSourceProtocol("tcp");

            if (checkNameExist(ardVtdu.getName())) {
                mediaClient.editPath(ardVtdu.getName(), conf);
            }
            ardVtdu.setRtspUrl(rtspUrl);
            ardVtdu.setRtmpUrl(rtmpUrl);
            ardVtdu.setWebrtcUrl(webrtcUrl);
            return ardVtdu;
        } catch (ForestRuntimeException ex) {
            log.error("修改流媒体异常：" + ex.getMessage());
        }
        return null;
    }

    @Override
    public void removePath(String[] names) {
        try {
            for (String name : names) {
                if (checkNameExist(name)) {
                    mediaClient.removePath(name);
                }
            }
        } catch (Exception ex) {
            log.error("批量移除流媒体异常：" + ex.getMessage());
        }
    }

    @Override
    public void removePath(String name) {
        try {
            if (checkNameExist(name)) {
                mediaClient.removePath(name);
            }
        } catch (ForestRuntimeException ex) {
            log.error("移除流媒体异常：" + ex.getMessage());
        }
    }

    @Override
    public List<StreamInfo> paths(Integer pageNum, Integer pageSize) {
        Paths paths = mediaClient.paths(pageNum - 1, pageSize);
        List<Item> items = paths.getItems();
        List<StreamInfo> pathInfoList = new ArrayList<>();
        for (Item item : items) {
            StreamInfo info = new StreamInfo();
            //ID
            String name = item.getName();
            info.setName(name);

            Conf conf = mediaClient.getPathConfig(name);
            String runOn = "";
            if (StringUtils.isNotEmpty(conf.getRunOnDemand())) {
                runOn = conf.getRunOnDemand();
            }
            //RTSP源地址
            Matcher matcher = Pattern.compile("rtsp://[^\\s\"]+").matcher(runOn);
            if (matcher.find()) {
                info.setRtspSource(matcher.group());
                info.setIsCode(1);
            } else {
                info.setRtspSource(conf.getSource());
                info.setIsCode(0);
            }
            //传输协议
            matcher = Pattern.compile("-rtsp_transport\\s+(\\w+)").matcher(runOn);
            if (matcher.find()) {
                info.setProtocol(matcher.group(1));
            }

            pathInfoList.add(info);
        }
        return pathInfoList;
    }

    @Override
    public RtspSession getRtspSessionById(String sessionId) {
        RtspSession rtspSession = mediaClient.getRtspsessionById(sessionId);
        return rtspSession;
    }

    @Override
    public WebrtcSession getWebrtcSessionById(String sessionId) {
        WebrtcSession webrtcSession = mediaClient.getWebrtcsessionById(sessionId);
        return webrtcSession;
    }

    @Override
    public RtmpSession getRtmpSessionById(String sessionId) {
        RtmpSession rtmpSession = mediaClient.getRtmpsessionById(sessionId);
        return rtmpSession;
    }

    /**
     * 获取推流列表
     * 刘苏义
     * 2023/8/29 9:37:05
     */
    @Override
    public TableDataInfo getPushStreamList(Integer pageNum, Integer pageSize) {
        TableDataInfo tableDataInfo = new TableDataInfo();
        List<StreamInfo> PushStreamInfoList = new ArrayList<>();
        Paths paths = mediaClient.paths(pageNum - 1, pageSize);
        int itemCount = paths.getItemCount();
        tableDataInfo.setTotal(itemCount);
        List<Item> items = paths.getItems();
        for (Item item : items) {
            StreamInfo info = new StreamInfo();
            //ID
            String name = item.getName();
            info.setName(name);
            Conf conf = mediaClient.getPathConfig(name);
            //RTMP播放地址
            String rtmpUrl = "rtmp://" + mediamtxHost + ":1935/" + name;
            info.setRtmpUrl(rtmpUrl);
            //RTSP播放地址
            String rtspUrl = "rtsp://" + mediamtxHost + ":8554/" + name;
            info.setRtspUrl(rtspUrl);
            //WEBRTC播放地址
            String webrtcUrl = "http://" + mediamtxHost + ":8889/" + name;
            info.setWebrtcUrl(webrtcUrl);
            Source source = item.getSource();
            if (source == null || source.getId().equals("")) {
                //会话ID
                info.setId("0");
                //下行流量
                long bytesReceived = item.getBytesReceived();
                String formatReceivedSize = ArdTool.formatFileSize(bytesReceived);
                info.setDownTraffic(formatReceivedSize);
                //上行流量
                long bytesSent = item.getBytesSent();
                String formatSentdSize = ArdTool.formatFileSize(bytesSent);
                info.setUpTraffic(formatSentdSize);
                info.setBeginTime(item.getReadyTime());
            } else {
                RtspSession rtspSession = getRtspSessionById(source.getId());
                //会话ID
                info.setId(rtspSession.getId());
                //开始推流时间
                info.setBeginTime(rtspSession.getCreated());
                //上行流量
                long bytesReceived = rtspSession.getBytesReceived();
                String formatReceivedSize = ArdTool.formatFileSize(bytesReceived);
                info.setUpTraffic(formatReceivedSize);
                //下行流量
                long bytesSent = rtspSession.getBytesSent();
                String formatSentSize = ArdTool.formatFileSize(bytesSent);
                info.setDownTraffic(formatSentSize);
                //推流服务器
                info.setRemoteAddr(rtspSession.getRemoteAddr());
            }
            //RTSP源地址
            String runOn = "";
            if (StringUtils.isNotEmpty(conf.getRunOnDemand())) {
                runOn = conf.getRunOnDemand();
            }
            Matcher matcher = Pattern.compile("rtsp://[^\\s\"]+").matcher(runOn);
            if (matcher.find()) {
                info.setRtspSource(matcher.group());
            } else {
                info.setRtspSource(conf.getSource());
            }
            //传输协议
            info.setProtocol(conf.getSourceProtocol());
            //拉流数量
            List<Readers> readers = item.getReaders();
            info.setNum(readers.size());

            PushStreamInfoList.add(info);
        }
        tableDataInfo.setRows(PushStreamInfoList);
        tableDataInfo.setCode(HttpStatus.SUCCESS);
        tableDataInfo.setMsg("查询成功");
        return tableDataInfo;
    }

    /**
     * 获取webrtc拉流列表
     * 刘苏义
     * 2023/8/29 9:37:05
     */
    @Override
    public TableDataInfo getPullWebrtcStreamList(Integer pageNum, Integer pageSize) {
        TableDataInfo tableDataInfo = new TableDataInfo();
        WebrtcSessions WebrtcSessions = mediaClient.webrtcsessions(pageNum - 1, pageSize);
        List<WebrtcSession> items = WebrtcSessions.getItems();
        items.forEach(
                webrtcSession -> {
                    webrtcSession.setUpStream(ArdTool.formatFileSize(webrtcSession.getBytesReceived()));
                    webrtcSession.setDownStream(ArdTool.formatFileSize(webrtcSession.getBytesSent()));
                    webrtcSession.setBeginTime(webrtcSession.getCreated());
                }
        );
        tableDataInfo.setTotal(WebrtcSessions.getItemCount());
        tableDataInfo.setRows(items);
        tableDataInfo.setCode(HttpStatus.SUCCESS);
        tableDataInfo.setMsg("查询成功");
        return tableDataInfo;
    }

    /**
     * 获取rtmp拉流列表
     * 刘苏义
     * 2023/8/29 9:37:05
     */
    @Override
    public TableDataInfo getPullRtmpStreamList(Integer pageNum, Integer pageSize) {
        TableDataInfo tableDataInfo = new TableDataInfo();
        RtmpSessions rtmpSessions = mediaClient.rtmpsessions(pageNum - 1, pageSize);
        List<RtmpSession> items = rtmpSessions.getItems();
        items.stream().forEach(
                webrtcSession -> {
                    webrtcSession.setUpStream(ArdTool.formatFileSize(webrtcSession.getBytesReceived()));
                    webrtcSession.setDownStream(ArdTool.formatFileSize(webrtcSession.getBytesSent()));
                    webrtcSession.setBeginTime(webrtcSession.getCreated());
                }
        );
        tableDataInfo.setTotal(rtmpSessions.getItemCount());
        tableDataInfo.setRows(items);
        tableDataInfo.setCode(HttpStatus.SUCCESS);
        tableDataInfo.setMsg("查询成功");
        return tableDataInfo;
    }

    /**
     * 获取rtsp拉流列表
     * 刘苏义
     * 2023/8/29 9:37:05
     */
    @Override
    public TableDataInfo getPullRtspStreamList(Integer pageNum, Integer pageSize) {
        TableDataInfo tableDataInfo = new TableDataInfo();
        RtspSessions rtspSessions = mediaClient.rtspsessions(pageNum - 1, pageSize);
        List<RtspSession> items = rtspSessions.getItems();
        items.stream().forEach(
                webrtcSession -> {
                    webrtcSession.setUpStream(ArdTool.formatFileSize(webrtcSession.getBytesReceived()));
                    webrtcSession.setDownStream(ArdTool.formatFileSize(webrtcSession.getBytesSent()));
                    webrtcSession.setBeginTime(webrtcSession.getCreated());
                }
        );
        tableDataInfo.setTotal(rtspSessions.getItemCount());
        tableDataInfo.setRows(items);
        tableDataInfo.setCode(HttpStatus.SUCCESS);
        tableDataInfo.setMsg("查询成功");
        return tableDataInfo;
    }

    /**
     * 踢出rtsp会话
     * 刘苏义
     * 2023/8/29 9:37:05
     */
    @Override
    public Boolean kickRtspSession(String sessionId) {
        try {
            mediaClient.kickRtspSessions(sessionId);
            return true;
        } catch (Exception ex) {
            return false;
        }
    }

    /**
     * 踢出rtmp会话
     * 刘苏义
     * 2023/8/29 9:37:05
     */
    @Override
    public Boolean kickRtmpSession(String sessionId) {
        try {
            mediaClient.kickRtmpSessions(sessionId);
            return true;
        } catch (Exception ex) {
            return false;
        }
    }

    /**
     * 踢出webrtc会话
     * 刘苏义
     * 2023/8/29 9:37:05
     */
    @Override
    public Boolean kickWebrtcSession(String sessionId) {
        try {
            mediaClient.kickWebrtcSessions(sessionId);
            return true;
        } catch (Exception ex) {
            return false;
        }
    }

    /**
     * 获取流媒体name列表
     * 刘苏义
     * 2023/10/13 14:19:07
     */
    @Override
    public List<String> getNameList() {
        List<String> nameList = new ArrayList<>();
        try {
            Paths paths = mediaClient.paths(0, 1000);
            List<Item> items = paths.getItems();
            for (Item item : items) {
                nameList.add(item.getName());
            }
        } catch (Exception ex) {
            log.error("获取流媒体name列表异常：" + ex.getMessage());
        }
        return nameList;
    }

    /**
     * 检查name是否存在
     * 刘苏义
     * 2023/10/19 15:18:45
     */
    @Override
    public boolean checkNameExist(String name) {
        boolean result = false;
        List<String> nameList = getNameList();
        if (nameList.contains(name)) {
            result = true;
        }
        return result;
    }

    /**
     * 清空所有流媒体信息
     *
     * @author 刘苏义
     * @date 2024/8/22 10:18
     */
    @Override
    public void clearAll() {
        List<String> nameList = getNameList();
        nameList.stream().forEach(name -> {
            removePath(name);
        });
    }

    /**
     * 开启录像
     */
    @Override
    public String setRecord(Conf conf) {
        return mediaClient.editPath(conf.getName(), conf);
    }

    /**
     * 获取全部录像分段列表
     */
    @Override
    public Paths recordingList(Integer pageNum, Integer pageSize) {
        Paths paths = mediaClient.recordingList(pageNum - 1, pageSize);
        paths.getItems().forEach(item -> {
            item.getSegments().forEach(segment -> {
                // 使用 OffsetDateTime 进行解析
                OffsetDateTime offsetDateTime = OffsetDateTime.parse(segment.getStart());
                // 转换为 java.util.Date
                Date date = Date.from(offsetDateTime.toInstant());
                segment.setStartTime(date);
            });
        });
        return paths;
    }

    /**
     * 指定路径名获取录像分段列表
     */
    @Override
    public Item recordingListByName(String name) {
        Item item = mediaClient.recordingListByName(name);
        item.getSegments().forEach(segment -> {
            // 使用 OffsetDateTime 进行解析
            OffsetDateTime offsetDateTime = OffsetDateTime.parse(segment.getStart());
            // 转换为 java.util.Date
            Date date = Date.from(offsetDateTime.toInstant());
            segment.setStartTime(date);
        });
        return item;
    }

    /**
     * 通过路径名和分段录像开始时间删除录像分段
     */
    @Override
    public String deleteSegment(String name, String start) {
        return mediaClient.deleteSegment(name, start);
    }

    //获取通道配置
    @Override
    public Conf getPathConfig(String name) {
        return mediaClient.getPathConfig(name);
    }

    //获取所有通道配置列表
    @Override
    public PathConfig getPathConfigs(Integer pageNum, Integer pageSize) {
        return mediaClient.getPathConfigs(pageNum - 1, pageSize);
    }

    //同步无人机推流到流媒体
    @Scheduled(cron = "0/20 * * * * ?")
    public void syncVtdu() {
        log.info("定时同步MediaMTX任务开始");
        List<String> nameList = getNameList();
        List<BoxReceiver> boxList = tyService.getBoxList();
        List<ArdVtdu> ardVtduList = boxList.stream()
                .filter(boxReceiver -> boxReceiver.getOnLine() != 0)  // 过滤掉不在线的
                .map(boxReceiver -> {
                    PullStreamUrlReceiver boxLiveUrl = tyService.getBoxLiveUrl(boxReceiver.getBoxSn());
                    return ArdVtdu.builder().name(boxReceiver.getBoxSn()).rtspSource(boxLiveUrl.getRtmp()).build();
                })
                .collect(Collectors.toList());
        asyncMediaMTX(ardVtduList, nameList);
    }

    /**
     * @return
     * @Author 刘苏义
     * @Description 同步本地vtdu库和流媒体中的数据
     * @Date 2024/7/10 15:26
     * @Param vtdus vtdu库的集合
     * @Param names 流媒体的name集合
     */
    public void asyncMediaMTX(List<ArdVtdu> vtdus, List<String> names) {
        //需要更新的数据,参数顺序注意
        sameList(vtdus, names).forEach(vtdu -> {
            Conf conf = getPathConfig(vtdu.getName());
            if (conf == null || conf.getSource() == null) {
                return;
            }
            String currentRtspSource = vtdu.getRtspSource();  // 提前获取需要比对的RTSP源
            String confRtspSource = null;

            // 未转码
            if (vtdu.getIsCode().equals(0)) {
                //获取source中的rtsp地址
                confRtspSource = conf.getSource();

            } else {
                // 转码
                if (vtdu.getIsRecord().equals(1)) {
                    //开启录像获取RunOnInit中的rtsp地址
                    confRtspSource = conf.getRunOnInit() != null ? matchRtsp(conf.getRunOnInit()) : null;
                } else {
                    //关闭录像获取RunOnDemand中的rtsp地址
                    confRtspSource = conf.getRunOnDemand() != null ? matchRtsp(conf.getRunOnDemand()) : null;
                }
            }
            // 比较 录像状态 是否一致
            boolean isRecordChanged = !Objects.equals(conf.getRecord() != null && conf.getRecord() ? 1 : 0, vtdu.getIsRecord());
            // 比较 RTSP 源是否一致
            boolean isRtspSourceChanged = !Objects.equals(confRtspSource, currentRtspSource);
            // 如果 RTSP 源或录制状态发生变化，则更新路径
            if (isRecordChanged || isRtspSourceChanged) {
                editPath(vtdu);
            }
        });
        //需要删除的数据
        diffListToDel(vtdus, names).forEach(name -> {
            removePath(name);
        });
        //需要新增的数据
        diffListToAdd(vtdus, names).forEach(vtdu -> {
            addPath(vtdu);
        });
    }

    //需要更新的（流媒体和vtdu相同）
    public List<ArdVtdu> sameList(List<ArdVtdu> vtduses, List<String> names) {
        return vtduses.stream().filter(vtdu -> names.contains(vtdu.getName())).collect(Collectors.toList());
    }

    //需要删除的（流媒体中多的）
    public List<String> diffListToDel(List<ArdVtdu> vtduses, List<String> names) {
        if (vtduses.size() >= names.size()) {
            List<String> nameList = vtduses.stream().map(ArdVtdu::getName)
                    .filter(item -> !names.contains(item))
                    .collect(Collectors.toList());
            return nameList;
        } else {
            List<String> nameList = names.stream().filter(item -> !vtduses.stream().map(ArdVtdu::getName).collect(Collectors.toList()).contains(item))
                    .collect(Collectors.toList());
            return nameList;
        }
    }

    //需要新增的（流媒体中少的）
    public List<ArdVtdu> diffListToAdd(List<ArdVtdu> vtduses, List<String> names) {
        return vtduses.stream().filter(vtdu -> !names.contains(vtdu.getName())).collect(Collectors.toList());
    }

    //匹配rtsp地址
    private String matchRtsp(String value) {
        Matcher matcher = Pattern.compile("rtsp://[^\\s\"]+").matcher(value);
        if (matcher.find()) {
            return matcher.group();
        } else {
            return "";
        }
    }
}

 ard-work/src/main/java/com/ard/vtdu/util/ArdTool.java

New file
@@ -0,0 +1,214 @@
package com.ard.vtdu.util;

import lombok.extern.slf4j.Slf4j;

import java.io.*;
import java.lang.reflect.Field;
import java.text.DecimalFormat;
import java.time.LocalTime;
import java.util.*;

/**
 * @Description: 工具方法
 * @ClassName: ArdTool
 * @Author: 刘苏义
 * @Date: 2023年06月02日14:37
 * @Version: 1.0
 **/
@Slf4j
public class ArdTool {

    /**
     * @描述 实体对象转Map
     * @参数 [entity]
     * @返回值 java.util.Map<java.lang.String, java.lang.Object>
     * @创建人 刘苏义
     * @创建时间 2023/6/2 14:35
     * @修改人和其它信息
     */
    public static Map<String, Object> convertEntityToMap(Object entity) {
        Map<String, Object> map = new HashMap<>();
        Field[] fields = entity.getClass().getDeclaredFields();
        for (Field field : fields) {
            field.setAccessible(true);
            try {
                Object value = field.get(entity);
                map.put(field.getName(), value);
            } catch (IllegalAccessException e) {
                e.printStackTrace();
            }
        }
        return map;
    }

    /**
     * 求Map<K,V>中最小 Value 对应的Key值
     *
     * @param map
     * @return
     */
    public static String getKeyByMinValue(Map<String, Double> map) {
        if (map == null || map.size() == 0) return null;
        List<Map.Entry<String, Double>> list = new ArrayList(map.entrySet());
        Collections.sort(list, (o1, o2) -> (o1.getValue().intValue() - o2.getValue().intValue()));
        String min = list.get(0).getKey();
//        String max = list.get(list.size() - 1).getKey();
        return min;
    }

    /**
     * @描述 通过日夜切换时间判断当前使用通道号
     * @参数 []
     * @返回值 java.lang.Integer
     * @创建人 刘苏义
     * @创建时间 2023/6/17 13:50
     * @修改人和其它信息
     */
    public static Integer getChannelBydayNightTime(String dayNightTime) {
        try {
            // 获取当前时间
            LocalTime currentTime = LocalTime.now();
            String[] dayNight = dayNightTime.split("-");
            // 解析时间字符串
            LocalTime startTime = LocalTime.parse(dayNight[0]);
            LocalTime endTime = LocalTime.parse(dayNight[1]);

            // 检查当前时间是否在给定的时间范围内
            boolean isWithinRange = !currentTime.isBefore(startTime) && !currentTime.isAfter(endTime);
            // 输出结果
            if (isWithinRange) {
                return 1;
            } else {
                return 2;
            }
        } catch (Exception ex) {
            log.error("光电切换获取通道异常：" + ex.getMessage());
            return 1;
        }
    }

    /**
     * 文件大小智能转换
     * 会将文件大小转换为最大满足单位
     *
     * @param size（文件大小，单位为B）
     * @return 文件大小
     */
    public static String formatFileSize(Long size) {
        String sizeName = null;
        if (1024 * 1024 > size && size >= 1024) {
            sizeName = String.format("%.2f", size.doubleValue() / 1024) + "KB";
        } else if (1024 * 1024 * 1024 > size && size >= 1024 * 1024) {
            sizeName = String.format("%.2f", size.doubleValue() / (1024 * 1024)) + "MB";
        } else if (size >= 1024 * 1024 * 1024) {
            sizeName = String.format("%.2f", size.doubleValue() / (1024 * 1024 * 1024)) + "GB";
        } else {
            sizeName = size.toString() + "B";
        }
        return sizeName;
    }

    /**
     * 文件大小智能转换
     * 会将文件大小转换为最大满足单位
     *
     * @param size（文件大小，单位为B）
     * @return 文件大小
     */
    public static String readableFileSize(long size) {
        if (size <= 0) {
            return "0";
        }
        final String[] units = new String[]{"B", "KB", "MB", "GB", "TB"};
        int digitGroups = (int) (Math.log10(size) / Math.log10(1024));
        return new DecimalFormat("#,###.##").format(size / Math.pow(1024, digitGroups)) + " " + units[digitGroups];
    }

    /**
     * byte数组转字符串
     * 刘苏义
     * 2023/10/18 8:42:59
     *
     * @param byteArray
     */
    public static String byteArrayToString(byte[] byteArray) {
        StringBuilder stringBuilder = new StringBuilder();
        for (byte b : byteArray) {
            stringBuilder.append(String.valueOf(b));
        }
        return stringBuilder.toString();
    }

    /**
     * 将数据写入txt文件
     * 刘苏义
     * 2024/3/8 12:58:39
     */
    public static void writeStringToFile(String data, String fileName) {

        try {
            File file = new File(fileName);
            File parentDir = file.getParentFile();
            if (!parentDir.exists()) {
                parentDir.mkdirs(); // 创建文件所在的目录，包括父目录
            }
            boolean fileExists = file.exists();
            boolean dataExists = false;
            if(fileExists) {
                // 读取文件内容
                BufferedReader reader = new BufferedReader(new FileReader(fileName));
                String line;
                while ((line = reader.readLine()) != null) {
                    if (line.equals(data)) {
                        dataExists = true;
                        break;
                    }
                }
                reader.close();
            }

            // 如果文件不存在或数据不存在，则追加到文件
            if (!fileExists || !dataExists) {
                FileWriter writer = new FileWriter(fileName, true);
                writer.write(data + System.lineSeparator());
                writer.close();
                if (!fileExists) {
                    log.debug("File created and data has been appended.");
                } else {
                    log.debug("Data has been appended to the file.");
                }
            } else {
                log.debug("Data already exists in the file. Not appending.");
            }
        } catch (IOException e) {
            log.error("An error occurred.");
            e.printStackTrace();
        }
    }

    /**
     *  通过反射对象填充
     * 刘苏义
     * 2024/4/2 11:50:57
     */
    public static void fillNullFields(Object source, Object target) {
        if (source == null || target == null) {
            return;
        }
        Field[] fields = source.getClass().getDeclaredFields();
        for (Field field : fields) {
            field.setAccessible(true);
            try {
                Object sourceValue = field.get(source);
                Object targetValue = field.get(target);
                if (targetValue == null && sourceValue != null) {
                    field.set(target, sourceValue);
                }
            } catch (IllegalAccessException e) {
                e.printStackTrace();
            }
        }
    }


}