forked from uav/ard-uav-server
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); }
@@ -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;
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; @@ -65,39 +66,35 @@ 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) -> { 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); 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(); //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("服务端链接不上，开始重连操作..."); log.error("与天宇服务{}:{}连接断开!", host, port); ctx.channel().eventLoop().schedule(() -> { log.error("天宇服务端链接不上，开始重连操作..."); try { tyNettyTcpClient.initOneConnection(host, port); } catch (Exception e) { e.printStackTrace(); } 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,34 +114,28 @@ 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()); } } @@ -155,7 +148,7 @@ @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());
