zjyr
/
PlatformServer
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

1. 重写格式化代码

This commit is contained in:
HuangXin 2023-08-12 12:43:36 +08:00
parent b1e49391e8
commit 3eb0572969
5 changed files with 43 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/main/java/com/zjyr/beidouservice/adapter/impl/netty/ChannelInit.java
View File

@ -28,11 +28,9 @@ public class ChannelInit<T> extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel channel) {
channel.pipeline()
.addLast("idle", new IdleStateHandler(0, 0, 60, TimeUnit.SECONDS))
.addLast("decode", new YuanRongProtocolDecode())
.addLast("startFlag", new ProtocolStartEncode(2))
.addLast("encode", new YuanRongProtocolEncode<T>())
channel.pipeline().addLast("idle", new IdleStateHandler(0, 0, 60, TimeUnit.SECONDS)).addLast("decode",
new YuanRongProtocolDecode())
.addLast("startFlag", new ProtocolStartEncode(2)).addLast("encode", new YuanRongProtocolEncode<T>())
.addLast("message", messageHandler);
}
}

51
src/main/java/com/zjyr/beidouservice/adapter/impl/netty/MessageHandler.java
View File

@ -45,8 +45,13 @@ public class MessageHandler<T> extends SimpleChannelInboundHandler<BaseBinaryPro
if (evt instanceof IdleStateEvent idleStateEvent) {
if (idleStateEvent.state() == IdleState.ALL_IDLE) {
log.info("{}:: Trigger Heart Signe", ctx.channel()
.id());
log.info("{}:: Trigger Heart Signe", ctx.channel().id());
MessageContent<HeartProtocol> msgCont = MessageContent.<HeartProtocol>builder().msgBody(
new HeartProtocol()).build();
BaseBinaryProtocol<HeartProtocol> h = BaseBinaryProtocol.<HeartProtocol>builder().msgContent(msgCont)
.build();
ctx.writeAndFlush(h);
}
} else {
super.userEventTriggered(ctx, evt);
@ -56,34 +61,23 @@ public class MessageHandler<T> extends SimpleChannelInboundHandler<BaseBinaryPro
@Override
public void channelRead0(ChannelHandlerContext ctx, BaseBinaryProtocol message) throws Exception {
//获取到线程池eventLoop，添加线程，执行
ctx.channel()
.eventLoop()
.execute(() -> {
//长时间操作，不至于长时间的业务操作导致Handler阻塞
//log.info("{}:: Receive Message: {}", ctx.channel().id(), HelperUtils.bytesToHexString(message
// .getStart()));
});
ctx.channel().eventLoop().execute(() -> {
//长时间操作，不至于长时间的业务操作导致Handler阻塞
//log.info("{}:: Receive Message: {}", ctx.channel().id(), HelperUtils.bytesToHexString(message
// .getStart()));
});
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
InetSocketAddress sa = (InetSocketAddress) ctx.channel()
.remoteAddress();
InetSocketAddress sa = (InetSocketAddress) ctx.channel().remoteAddress();
//List<ControlDevice> list = controlDeviceMapper.selectAll();
log.info("{}:: Connected <-- {}:{}", ctx.channel()
.id(), sa.getAddress()
.getHostAddress(), sa.getPort());
controlDeviceMapper.addControlDevice(ControlDevice.builder()
.deviceType(ControlDeviceTypeName.DEVICE_SOCKET_TCP)
.deviceAddr(sa.getAddress()
.getHostAddress())
.build());
BaseBinaryProtocol<HeartProtocol> h = BaseBinaryProtocol.<HeartProtocol>builder()
.msgContent(MessageContent.<HeartProtocol>builder()
.msgBody(new HeartProtocol())
.build())
.build();
log.info("{}:: Connected <-- {}:{}", ctx.channel().id(), sa.getAddress().getHostAddress(), sa.getPort());
controlDeviceMapper.addControlDevice(ControlDevice.builder().deviceType(ControlDeviceTypeName.DEVICE_SOCKET_TCP)
.deviceAddr(sa.getAddress().getHostAddress()).build());
MessageContent<HeartProtocol> msgCont = MessageContent.<HeartProtocol>builder().msgBody(new HeartProtocol())
.build();
BaseBinaryProtocol<HeartProtocol> h = BaseBinaryProtocol.<HeartProtocol>builder().msgContent(msgCont).build();
ctx.writeAndFlush(h);
super.channelActive(ctx);
@ -91,11 +85,8 @@ public class MessageHandler<T> extends SimpleChannelInboundHandler<BaseBinaryPro
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
InetSocketAddress sa = (InetSocketAddress) ctx.channel()
.remoteAddress();
log.info("{}:: Disonnected <-- {}", ctx.channel()
.id(), sa.getAddress()
.getHostAddress());
InetSocketAddress sa = (InetSocketAddress) ctx.channel().remoteAddress();
log.info("{}:: Disonnected <-- {}", ctx.channel().id(), sa.getAddress().getHostAddress());
super.channelActive(ctx);
ctx.close();
}

4
src/main/java/com/zjyr/beidouservice/adapter/impl/netty/encode/ProtocolStartEncode.java
View File

@ -25,9 +25,7 @@ public class ProtocolStartEncode extends LengthFieldPrepender {
@Override
protected void encode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception {
out.add(ctx.alloc()
.buffer(BaseBinaryProtocol.START.length())
.writeBytes(BaseBinaryProtocol.START.getBytes()));
out.add(ctx.alloc().buffer(BaseBinaryProtocol.START.length()).writeBytes(BaseBinaryProtocol.START.getBytes()));
super.encode(ctx, msg, out);
}
}

47
src/main/java/com/zjyr/beidouservice/adapter/impl/netty/encode/YuanRongProtocolEncode.java
View File

@ -27,13 +27,11 @@ public class YuanRongProtocolEncode<T> extends MessageToByteEncoder<BaseBinaryPr
ByteBuf byteBuf) {
if (baseBinaryProtocol == null || baseBinaryProtocol.getMsgContent() == null ||
baseBinaryProtocol.getMsgContent()
.getMsgBody() == null) {
baseBinaryProtocol.getMsgContent().getMsgBody() == null) {
return;
}
T msgBody = baseBinaryProtocol.getMsgContent()
.getMsgBody();
T msgBody = baseBinaryProtocol.getMsgContent().getMsgBody();
byteBuf.writeByte(BaseBinaryProtocol.VERSIN); // 协议版本号
byteBuf.writeInt(0); // 接收设备主ID
@ -45,21 +43,15 @@ public class YuanRongProtocolEncode<T> extends MessageToByteEncoder<BaseBinaryPr
byteBuf.writeInt(0); // 消息状态码
if (msgBody instanceof HeartProtocol) {
byteBuf.writeByte(ControlCommandName.COMMAND_HEART.getValue()
.byteValue()); // 消息类型
byteBuf.writeByte(ControlCommandName.COMMAND_HEART.getValue().byteValue()); // 消息类型
byteBuf.writeShort(0); // 消息内容长度
} else if (msgBody instanceof SensorControlProtocol msg) {
byteBuf.writeByte(ControlCommandName.COMMAND_SENSOR.getValue()
.byteValue()); // 消息类型
byteBuf.writeByte(ControlCommandName.COMMAND_SENSOR.getValue().byteValue()); // 消息类型
if (!msg.getControllContents()
.isEmpty()) {
byteBuf.writeShort(13 + 4 * msg.getControllContents()
.size()); // 消息内容长度
byteBuf.writeByte(msg.getControlTunnel()
.getValue()
.byteValue());
if (!msg.getControllContents().isEmpty()) {
byteBuf.writeShort(13 + 4 * msg.getControllContents().size()); // 消息内容长度
byteBuf.writeByte(msg.getControlTunnel().getValue().byteValue());
byteBuf.writeBytes(msg.getTunnelAddr());
byteBuf.writeByte(msg.getMonth());
byteBuf.writeByte(msg.getDays());
@ -67,42 +59,31 @@ public class YuanRongProtocolEncode<T> extends MessageToByteEncoder<BaseBinaryPr
byteBuf.writeByte(msg.getMinute());
byteBuf.writeByte(msg.getProvince());
byteBuf.writeByte(msg.getCity());
byteBuf.writeByte(msg.getControlMode()
.getValue()
.byteValue());
byteBuf.writeByte(msg.getControlMode().getValue().byteValue());
byteBuf.writeByte(msg.getControlCommand());
byteBuf.writeByte(msg.getNControlInfo());
for (var c : msg.getControllContents()) {
byteBuf.writeByte(c.getDistrictsCode());
byteBuf.writeByte(c.getControlAction()
.getValue()
.byteValue());
byteBuf.writeByte(c.getControlAction().getValue().byteValue());
byteBuf.writeShort(c.getSensorId());
}
}
} else if (msgBody instanceof QuerySensorProtocol msg) {
byteBuf.writeByte(ControlCommandName.COMMAND_QUERY_SENSOR.getValue()
.byteValue()); // 消息类型
byteBuf.writeByte(ControlCommandName.COMMAND_QUERY_SENSOR.getValue().byteValue()); // 消息类型
if (!msg.getControllContents()
.isEmpty()) {
byteBuf.writeShort(2 + 4 * msg.getControllContents()
.size()); // 消息内容长度
if (!msg.getControllContents().isEmpty()) {
byteBuf.writeShort(2 + 4 * msg.getControllContents().size()); // 消息内容长度
byteBuf.writeByte(msg.getProvince());
byteBuf.writeByte(msg.getCity());
byteBuf.writeByte(msg.getNControlInfo());
for (var c : msg.getControllContents()) {
byteBuf.writeByte(c.getDistrictsCode());
byteBuf.writeByte(c.getControlAction()
.getValue()
.byteValue());
byteBuf.writeByte(c.getControlAction().getValue().byteValue());
byteBuf.writeShort(c.getSensorId());
}
}
} else {
log.info("XXXXX Encode: {}", baseBinaryProtocol.getMsgContent()
.getMsgBody()
.getClass());
log.info("XXXXX Encode: {}", baseBinaryProtocol.getMsgContent().getMsgBody().getClass());
}
log.info("\n{}", ByteBufUtil.prettyHexDump(byteBuf));

13
src/main/java/com/zjyr/beidouservice/adapter/impl/netty/impl/TcpServer.java
View File

@ -51,15 +51,10 @@ public class TcpServer implements ISocketServer {
*/
private void tcpServer(int port) {
try {
new ServerBootstrap().group(boosGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.localAddress(new InetSocketAddress(port))
.childHandler(channelInit)
.option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.TCP_NODELAY, false)
.childOption(ChannelOption.SO_KEEPALIVE, true)
.bind()
.sync();
new ServerBootstrap().group(boosGroup, workerGroup).channel(NioServerSocketChannel.class).localAddress(
new InetSocketAddress(port)).childHandler(channelInit).option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.TCP_NODELAY, false).childOption(ChannelOption.SO_KEEPALIVE,
true).bind().sync();
log.info("Netty TCP Server Beginning Listen: {}", port);
} catch (Exception e) {
log.error(e.getMessage());