diff --git a/sec-beidou-fwd/src/main/java/com/imdroid/beidou_fwd/service/TCPClient.java b/sec-beidou-fwd/src/main/java/com/imdroid/beidou_fwd/service/TCPClient.java index d0032975..03303a96 100644 --- a/sec-beidou-fwd/src/main/java/com/imdroid/beidou_fwd/service/TCPClient.java +++ b/sec-beidou-fwd/src/main/java/com/imdroid/beidou_fwd/service/TCPClient.java @@ -22,10 +22,10 @@ public class TCPClient { //@Value("${xfz.server.port}") private int port; - private Bootstrap bootstrap; private EventLoopGroup group; private Channel channel; + ByteBuf sendBuffer; public void start() { new Thread(this::connect, "xfz-tcp-client").start(); @@ -34,6 +34,7 @@ public class TCPClient { public void init(String dest_addr, int dest_port) { host = dest_addr; port = dest_port; + sendBuffer = Unpooled.buffer(); //客户端需要一个事件循环组 group = new NioEventLoopGroup(); //创建客户端启动对象 @@ -45,7 +46,7 @@ public class TCPClient { @Override protected void initChannel(SocketChannel ch) throws Exception { //加入处理器 - ch.pipeline().addLast(new TcpMessageHandler()); + ch.pipeline().addLast(new TcpMessageHandler(TCPClient.this)); } }); } @@ -81,21 +82,26 @@ public class TCPClient { } } - public void writeAndFlush(String json) { - String str = "#" + json + "!"; - ByteBuf buf = Unpooled.buffer(); - buf.writeBytes(str.getBytes(StandardCharsets.UTF_8)); - channel.writeAndFlush(buf).addListener(future -> { - if (future.isSuccess()) { - logger.info("send {} to xfz server succeed.", str); - } else { - logger.info("send {} to xfz server failed.", str); - } - }); + sendBuffer.writeBytes(json.getBytes(StandardCharsets.UTF_8)); + if(channel.isActive()) flush(); + else connect(); } - public boolean isActive(){ - return channel.isActive(); + public void onConnectionActive(){ + flush(); + } + + private void flush(){ + if(sendBuffer.readableBytes()>0){ + channel.writeAndFlush(sendBuffer).addListener(future -> { + if (future.isSuccess()) { + logger.info("send to xfz server succeed."); + sendBuffer.clear(); + } else { + logger.info("send to xfz server failed."); + } + }); + } } } diff --git a/sec-beidou-fwd/src/main/java/com/imdroid/beidou_fwd/service/TcpMessageHandler.java b/sec-beidou-fwd/src/main/java/com/imdroid/beidou_fwd/service/TcpMessageHandler.java index 4b058d41..986c1505 100644 --- a/sec-beidou-fwd/src/main/java/com/imdroid/beidou_fwd/service/TcpMessageHandler.java +++ b/sec-beidou-fwd/src/main/java/com/imdroid/beidou_fwd/service/TcpMessageHandler.java @@ -15,6 +15,12 @@ public class TcpMessageHandler extends SimpleChannelInboundHandler { private final Logger logger = LoggerFactory.getLogger(this.getClass()); + private final TCPClient tcpClient; + + public TcpMessageHandler(TCPClient tcpClient) { + this.tcpClient = tcpClient; + } + @Override protected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf buf) throws Exception { if (logger.isDebugEnabled()) { @@ -27,6 +33,7 @@ public class TcpMessageHandler extends SimpleChannelInboundHandler { @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { logger.info("xfz tcp channel active"); + tcpClient.onConnectionActive(); } @Override diff --git a/sec-beidou-fwd/src/main/java/com/imdroid/beidou_fwd/task/GXXfzForwarder.java b/sec-beidou-fwd/src/main/java/com/imdroid/beidou_fwd/task/GXXfzForwarder.java index 72a14d6b..205d1ebb 100644 --- a/sec-beidou-fwd/src/main/java/com/imdroid/beidou_fwd/task/GXXfzForwarder.java +++ b/sec-beidou-fwd/src/main/java/com/imdroid/beidou_fwd/task/GXXfzForwarder.java @@ -82,26 +82,11 @@ public class GXXfzForwarder extends Forwarder{ data.setZ(NumberUtils.scale(locationRecord.getRb562d() * 0.001, 5)); sendNum++; } - String json = GsonUtil.toJson(xfzTcpMessage); + String json = "#" + GsonUtil.toJson(xfzTcpMessage) + "!"; logger.info("project " + projectId + ": push calculation result to XFZ"); logger.info(json); - if(!xfzTcpClient.isActive()) { - xfzTcpClient.connect(); - try { - Thread.sleep(5000); - } - catch (Exception e){ - - } - } - if(xfzTcpClient.isActive()) { - xfzTcpClient.writeAndFlush(json); - return sendNum; - } - else{ - logger.info("TCP disconnected!"); - return 0; - } + xfzTcpClient.writeAndFlush(json); + return sendNum; } }