forked from uav/ard-uav-server
ard-admin/src/main/resources/application-dev.yml
@@ -43,6 +43,7 @@ # 天宇tcp服务器地址 tcpserver: 59.110.237.99 tcpport: 8125 #流媒体配置 mediamtx: # 流媒体服务器地址 ard-work/src/main/java/com/ard/tianyu/component/netty/NettyTcpClient.java
File was renamed from ard-work/src/main/java/com/ard/tianyu/component/netty/TyNettyTcpClient.java @@ -21,9 +21,9 @@ @Component @Slf4j public class TyNettyTcpClient { public class NettyTcpClient { public static Map<String, TyBoxSession> channels = new HashMap<>(); public static Map<String, TcpClientSession> channels = new HashMap<>(); EventLoopGroup eventLoopGroup = new NioEventLoopGroup(); @@ -59,11 +59,11 @@ */ public void initConnection(String host, Integer port) { log.info("初始化连接天宇tcp服务器【{}:{}】", host, port); String currentBoxSn = ""; log.info("初始化连接TCP服务器【{}:{}】", host, port); String clientId = "clientId_1"; //判断是否已经存在连接，如果存在则关闭之前的连接 if (channels.containsKey(host + ":" + port)) { currentBoxSn = channels.get(host + ":" + port).getSubBoxSn(); clientId = channels.get(host + ":" + port).getClientId(); channels.get(host + ":" + port).getChannelFuture().channel().close(); } @@ -71,12 +71,12 @@ Bootstrap bootstrap = getBootstrap(); ChannelFuture future = bootstrap.connect(host, port).addListener((ChannelFuture futureListener) -> { if (futureListener.isSuccess()) { log.info("连接天宇TCP服务器成功【{}:{}】", host, port); log.info("连接TCP服务器成功【{}:{}】", host, port); TyTcpServerCache.putTcpServerCache(host + ":" + port, true); } else { final EventLoop eventLoop = futureListener.channel().eventLoop(); eventLoop.schedule(() -> { log.info("连接天宇TCP服务器失败【{}:{}】,正尝试重连...", host, port); log.info("连接TCP服务器失败【{}:{}】,正尝试重连...", host, port); TyTcpServerCache.putTcpServerCache(host + ":" + port, false); try { initConnection(host, port); @@ -87,7 +87,8 @@ } }); //将连接存入集合 channels.put(host + ":" + port, TyBoxSession.builder().subBoxSn(currentBoxSn).channelFuture(future).isReConnect(!currentBoxSn.equals("")).build()); channels.put(host + ":" + port, TcpClientSession.builder().ClientId(clientId).channelFuture(future).isReConnect(!clientId.isEmpty()).build()); } //创建天宇连接 ard-work/src/main/java/com/ard/tianyu/component/netty/TcpClientSession.java
File was renamed from ard-work/src/main/java/com/ard/tianyu/component/netty/TyBoxSession.java @@ -10,9 +10,9 @@ @Builder @AllArgsConstructor @NoArgsConstructor public class TyBoxSession { public class TcpClientSession { public String subBoxSn; public String ClientId; public ChannelFuture channelFuture;
@@ -1,7 +1,7 @@ package com.ard.tianyu.component.netty.handler; import com.ard.tianyu.domain.cache.TyTcpServerCache; import com.ard.tianyu.component.netty.TyNettyTcpClient; import com.ard.tianyu.component.netty.NettyTcpClient; import com.ard.tianyu.component.netty.command.BaseCmd; import com.ard.tianyu.service.ITyService; import io.netty.channel.ChannelHandler; @@ -25,7 +25,7 @@ private ITyService tyService; @Resource private TyNettyTcpClient tyNettyTcpClient; private NettyTcpClient nettyTcpClient; @Override public void channelRead0(ChannelHandlerContext ctx, BaseCmd command) { @@ -75,7 +75,7 @@ ctx.channel().eventLoop().schedule(() -> { log.error("天宇服务端链接不上，开始重连操作..."); try { tyNettyTcpClient.initConnection(host, port); nettyTcpClient.initConnection(host, port); } catch (Exception e) { log.error("天宇服务端连接异常: {}", e.getMessage()); } ard-work/src/main/java/com/ard/tianyu/component/websocket/handler/TyMessageHandler.java
@@ -3,6 +3,7 @@ import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.TypeReference; import com.ard.tianyu.component.websocket.WebSocketSessionManager; import com.ard.tianyu.domain.enums.MessageType; import com.ard.tianyu.domain.message.*; import com.ard.tianyu.service.control.UavControlService; import com.ard.tianyu.component.websocket.utils.WebSocketUtil; @@ -12,6 +13,7 @@ import org.springframework.web.socket.TextMessage; import org.springframework.web.socket.WebSocketSession; import org.springframework.web.socket.handler.TextWebSocketHandler; import javax.annotation.Resource; @@ -33,34 +35,39 @@ private UavControlManager uavControlManager; @Resource private UavAllManager uavAllManager; @Override public void handleTextMessage(WebSocketSession session, TextMessage message) { String userIdStr = (String) session.getAttributes().get("userId"); Long userId = Long.parseLong(userIdStr); String payload = message.getPayload(); log.info("【cmd】Received message: {}", payload); log.info("【cmd】size: {}", WebSocketSessionManager.getOnlineUserCount()); BaseMessage<?> baseMessage = JSON.parseObject(payload, BaseMessage.class); switch (baseMessage.getType()) { case "stream": handleStreamMessage(payload, userId); break; case "state": handleStateMessage(payload, userId); break; case "telemetry": handleTelemetryMessage(payload, userId); break; case "signal": handleSignalMessage(payload, userId); break; case "controller": handleControlMessage(payload, userId); break; case "all": handleAllMessage(payload, userId); break; try { String userIdStr = (String) session.getAttributes().get("userId"); Long userId = Long.parseLong(userIdStr); String payload = message.getPayload(); log.info("【cmd】Received message: {}", payload); log.info("【cmd】size: {}", WebSocketSessionManager.getOnlineUserCount()); BaseMessage<?> baseMessage = JSON.parseObject(payload, BaseMessage.class); MessageType messageType = MessageType.fromString(baseMessage.getType()); switch (messageType) { case STREAM: handleStreamMessage(payload, userId); break; case STATE: handleStateMessage(payload, userId); break; case TELEMETRY: handleTelemetryMessage(payload, userId); break; case SIGNAL: handleSignalMessage(payload, userId); break; case CONTROLLER: handleControlMessage(payload, userId); break; case ALL: handleAllMessage(payload, userId); break; } } catch (Exception e) { log.error("Error handling message: {}", message, e); } } @@ -215,6 +222,7 @@ } } } private void handleAllMessage(String payload, Long userId) { // 解析接收到的JSON消息，将其转换为BaseMessage对象 BaseMessage<UavControlMessage> controlMessage = JSON.parseObject(payload, @@ -238,6 +246,7 @@ } } } @Override public void afterConnectionEstablished(WebSocketSession session) { log.info("【cmd】Connected: {}", session.getId()); ard-work/src/main/java/com/ard/tianyu/domain/enums/MessageType.java
New file @@ -0,0 +1,29 @@ package com.ard.tianyu.domain.enums; public enum MessageType { STREAM("stream"), STATE("state"), TELEMETRY("telemetry"), SIGNAL("signal"), CONTROLLER("controller"), ALL("all"); private final String type; MessageType(String type) { this.type = type; } public String getType() { return type; } public static MessageType fromString(String type) { for (MessageType messageType : MessageType.values()) { if (messageType.type.equalsIgnoreCase(type)) { return messageType; } } throw new IllegalArgumentException("Unknown message type: " + type); } } ard-work/src/main/java/com/ard/tianyu/service/impl/TyServiceImpl.java
@@ -34,7 +34,7 @@ import com.ard.tianyu.domain.dto.signal.SignalInfo; import com.ard.tianyu.domain.dto.state.UavState; import com.ard.tianyu.domain.dto.telemetry.TelemetryData; import com.ard.tianyu.component.netty.TyNettyTcpClient; import com.ard.tianyu.component.netty.NettyTcpClient; import com.ard.tianyu.component.netty.command.control.AbsAngleControlCmd; import com.ard.tianyu.component.netty.command.control.AngleControlCmd; import com.ard.tianyu.component.netty.command.control.SpeedControlCmd; @@ -827,7 +827,7 @@ */ private void sendMessage(String key, BaseCmd msg) { try { ChannelFuture future = TyNettyTcpClient.channels.get(key).getChannelFuture(); ChannelFuture future = NettyTcpClient.channels.get(key).getChannelFuture(); Channel channel = future.channel(); if (channel.isActive()) { InetSocketAddress ipSocket = (InetSocketAddress) channel.remoteAddress();
