优化

liusuyi

2024-10-22 ce309155d4a0d750f786ca242b36bba8fc0bbf14

优化

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

@@ -3,7 +3,6 @@
import com.ard.tianyu.http.TyClient;
import com.ard.tianyu.model.dto.TokenDTO;
import com.ard.tianyu.model.receiver.AccessTokenReceiver;
import com.ard.tianyu.model.receiver.BoxReceiver;
import com.ard.tianyu.model.receiver.TyHttpCommonReceiver;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
@@ -13,7 +12,6 @@

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;

/**
 * TianyuConfiguration
@@ -57,7 +55,7 @@
    }

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

    @Resource

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

@@ -6,8 +6,8 @@
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;
import com.ard.tianyu.model.receiver.PullStreamUrlReceiver;
import com.ard.tianyu.model.receiver.BoxInfoReceiver;
import com.ard.tianyu.model.receiver.BoxStreamUrlReceiver;
import com.ard.tianyu.service.ITyService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
@@ -39,14 +39,14 @@
    @ApiOperation("获取天宇盒子列表")
    @GetMapping("/getBoxList")
    public AjaxResult getBoxList() {
        List<BoxReceiver> BoxList = tyService.getBoxList();
        List<BoxInfoReceiver> BoxList = tyService.getBoxList();
        return AjaxResult.success(BoxList);
    }

    @ApiOperation("获取直播地址")
    @GetMapping("/getBoxList/{boxSn}")
    public AjaxResult getBoxLiveUrl(@PathVariable("boxSn") String boxSn) {
        PullStreamUrlReceiver boxLiveUrl = tyService.getBoxLiveUrl(boxSn);
        BoxStreamUrlReceiver boxLiveUrl = tyService.getBoxLiveUrl(boxSn);
        return AjaxResult.success(boxLiveUrl);
    }


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

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

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

    /**
     * 获取飞行历史

 ard-work/src/main/java/com/ard/tianyu/model/receiver/AccessTokenReceiver.java

@@ -3,8 +3,6 @@
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.Data;
import org.springframework.util.MultiValueMap;

import java.util.HashMap;
import java.util.Map;


 ard-work/src/main/java/com/ard/tianyu/model/receiver/BoxInfoReceiver.java

File was renamed from ard-work/src/main/java/com/ard/tianyu/model/receiver/BoxReceiver.java
@@ -6,7 +6,7 @@

@Data
@JsonIgnoreProperties(ignoreUnknown = true)
public class BoxReceiver {
public class BoxInfoReceiver {

    /**
     * 云盒SN编号

 ard-work/src/main/java/com/ard/tianyu/model/receiver/BoxStreamUrlReceiver.java

File was renamed from ard-work/src/main/java/com/ard/tianyu/model/receiver/PullStreamUrlReceiver.java
@@ -6,7 +6,7 @@

@Data
@JsonIgnoreProperties(ignoreUnknown = true)
public class PullStreamUrlReceiver {
public class BoxStreamUrlReceiver {

    @JsonProperty("rtsp")
    private String rtsp;

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

@@ -11,6 +11,7 @@
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;
@@ -45,59 +46,55 @@
                    @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);
                        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) {
    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);
            initOneConnection(host, port);
        }
    }

    public void initOneConnection(String host,Integer port){
    public void initOneConnection(String host, Integer port) {

        log.info("调用一次{}:{}",host,port);
        log.info("初始化连接天宇tcp服务器【{}:{}】", host, port);
        //判断该账号下，是否有盒子被控制
        //当前正在被控制的盒子
        String currentBoxSn = "";
        if(channels.containsKey(host+"/"+port))
        {
            currentBoxSn = channels.get(host+"/"+port).getSubBoxSn();
            channels.get(host+"/"+port).getFuture().channel().close();
        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) -> {
        ChannelFuture future = bootstrap.connect(host, port).addListener((ChannelFuture futureListener) -> {
            if (futureListener.isSuccess()) {
                log.info("连接天宇TCP服务端成功:"+host+":"+port);
                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();
                        }
                eventLoop.schedule(() -> {
                    log.info("连接天宇TCP服务器失败【{}:{}】,正尝试重连...", host, port);
                    try {
                        initOneConnection(host, port);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }, 5, TimeUnit.SECONDS);
            }
        });

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

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

@@ -10,9 +10,7 @@
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;
@@ -23,17 +21,17 @@
@ChannelHandler.Sharable
public class ClientHandler extends SimpleChannelInboundHandler<BaseCommand> {

    @Autowired
    ITyService tyService;
    @Resource
    private 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();
    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);
    }

@@ -47,7 +45,7 @@

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        log.error("发生异常: ", cause); // 使用日志记录异常
        ctx.close();
    }

@@ -58,9 +56,9 @@
            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);
                log.info("写空闲发送⼼跳!");
                HeartbeatCommand msg = new HeartbeatCommand(System.currentTimeMillis());
                ctx.writeAndFlush(msg);
            } else if (event.state() == IdleState.ALL_IDLE) {
                log.info("客户端所有操作超时");
            }
@@ -72,18 +70,18 @@
        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();
                }
        log.error("与天宇服务{}:{}连接断开!", host, port);

        ctx.channel().eventLoop().schedule(() -> {
            log.error("天宇服务端链接不上，开始重连操作...");
            try {
                tyNettyTcpClient.initOneConnection(host, port);
            } catch (Exception e) {
                log.error("天宇服务端连接异常: {}", e.getMessage());
            }
        }, 5, TimeUnit.SECONDS);

        super.channelInactive(ctx);
    }

}

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

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

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

@@ -27,13 +27,13 @@
     * 获取云盒列表
     * @return
     */
    List<BoxReceiver> getBoxList();
    List<BoxInfoReceiver> getBoxList();

    /**
     * 获取直播地址
     * @return
     */
    PullStreamUrlReceiver  getBoxLiveUrl(@Var("boxSn") String boxSn);
    BoxStreamUrlReceiver getBoxLiveUrl(@Var("boxSn") String boxSn);

    /**
     * 获取飞行历史

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

@@ -12,7 +12,6 @@
import com.ard.tianyu.netty.command.RegisterCommand;
import com.ard.tianyu.netty.TyNettyTcpClient;
import com.ard.tianyu.service.ITyService;
import com.ard.vtdu.util.ArdTool;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
@@ -44,7 +43,7 @@
    /**
     * 获取天宇token信息
     *
     * @return
     * @return AccessTokenReceiver
     */
    @Override
    public AccessTokenReceiver getAccessToken() {
@@ -54,31 +53,31 @@
    /**
     * 获取云盒列表
     *
     * @return
     * @return List<BoxInfoReceiver>
     */
    @Override
    public List<BoxReceiver> getBoxList() {
    public List<BoxInfoReceiver> getBoxList() {
        //请求云盒列表
        TyHttpCommonReceiver<List<BoxReceiver>> boxListReveiver = tyClient.getBoxList(tyConfiguration.getAccessToken().convertFromDTO());
        TyHttpCommonReceiver<List<BoxInfoReceiver>> boxListReveiver = tyClient.getBoxList(tyConfiguration.getAccessToken().convertFromDTO());
        return boxListReveiver.getContent();
    }

    /**
     * 获取直播地址
     *
     * @return
     * @return BoxStreamUrlReceiver
     */
    @Override
    public PullStreamUrlReceiver getBoxLiveUrl(String boxSn) {
    public BoxStreamUrlReceiver getBoxLiveUrl(String boxSn) {

        TyHttpCommonReceiver<PullStreamUrlReceiver> boxLiveUrl = tyClient.getBoxLiveUrl(boxSn, tyConfiguration.getAccessToken().convertFromDTO());
        TyHttpCommonReceiver<BoxStreamUrlReceiver> boxLiveUrl = tyClient.getBoxLiveUrl(boxSn, tyConfiguration.getAccessToken().convertFromDTO());
        return boxLiveUrl.getContent();
    }

    /**
     * 获取飞行历史
     *
     * @return
     * @return List<BoxHistoryReceiver>
     */
    @Override
    public List<BoxHistoryReceiver> getBoxHistory(BoxHistoryDTO boxHistoryDTO) {
@@ -89,7 +88,7 @@
    /**
     * 获取盒子图片
     *
     * @return
     * @return List<BoxPhotosReceiver>
     */
    @Override
    public List<BoxPhotosReceiver> getBoxPhotos(BoxPhotosDTO boxPhotosDTO) {
@@ -102,7 +101,7 @@
     */
    @Override
    public Boolean updateBoxSettings(BoxSettingsDTO boxSettingsDTO) {
        TyHttpCommonReceiver<String> result = tyClient.updateBoxSettings(boxSettingsDTO, tyConfiguration.getAccessToken().convertFromDTO());
        TyHttpCommonReceiver<?> result = tyClient.updateBoxSettings(boxSettingsDTO, tyConfiguration.getAccessToken().convertFromDTO());
        return result.getState() == 1;
    }

@@ -115,47 +114,41 @@
        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;
    public void sendMessage(String key, BaseCommand msg) {
        try {
            ChannelFuture future = TyNettyTcpClient.channels.get(key).getFuture();
            Channel channel = future.channel();
            if (future != null && channel.isActive()) {
            if (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 + "发送数据" + "，命令码：" + String.format("0x%02X", msg.getCmdCode() & 0xFF));
                log.info("向天宇服务器【{}:{}】发送数据，命令码：{}", host, port, String.format("0x%02X", msg.getCmdCode() & 0xFF));
            } else {
                log.error("与天宇服务器链接断开，无法发送数据！");
            }
        } catch (Exception e) {
            log.info("Failed to publish the message. {}", "转发信息");
            e.printStackTrace();
        } finally {
            return cmdResult;
            log.error("发送消息异常：{}", e.getMessage());
        }
    }

    @Override
    public void controlAngle(String boxSn) {
        ControlAngleCommand controlAngleCommand = new ControlAngleCommand(boxSn.getBytes(),(byte)0x00,(byte)0x00,(byte)0x05);
        ControlAngleCommand controlAngleCommand = new ControlAngleCommand(boxSn.getBytes(), (byte) 0x00, (byte) 0x00, (byte) 0x05);
        this.sendMessage(tyConfiguration.getIpPort(), controlAngleCommand);
    }

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

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

@@ -2,8 +2,8 @@

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.model.receiver.BoxInfoReceiver;
import com.ard.tianyu.model.receiver.BoxStreamUrlReceiver;
import com.ard.tianyu.service.ITyService;
import com.ard.vtdu.client.MediaClient;
import com.ard.vtdu.domain.*;
@@ -570,14 +570,14 @@
    public void syncVtdu() {
        log.info("定时同步MediaMTX任务开始");
        List<String> nameList = getNameList();
        List<BoxReceiver> boxList = tyService.getBoxList();
        List<BoxInfoReceiver> boxList = tyService.getBoxList();
        if (boxList == null || boxList.isEmpty()) {
            return;
        }
        List<ArdVtdu> ardVtduList = boxList.stream()
                .filter(boxReceiver -> boxReceiver.getOnLine() != 0)  // 过滤掉不在线的
                .map(boxReceiver -> {
                    PullStreamUrlReceiver boxLiveUrl = tyService.getBoxLiveUrl(boxReceiver.getBoxSn());
                    BoxStreamUrlReceiver boxLiveUrl = tyService.getBoxLiveUrl(boxReceiver.getBoxSn());
                    return ArdVtdu.builder().name(boxReceiver.getBoxSn()).rtspSource(boxLiveUrl.getRtmp()).build();
                })
                .collect(Collectors.toList());