From 0f8870a794fe13d02ed29d01472fcb6bd236158f Mon Sep 17 00:00:00 2001 From: jaysunxiao Date: Wed, 12 Jan 2022 11:28:38 +0800 Subject: [PATCH] =?UTF-8?q?feat[net]:=20=E5=A2=9E=E5=8A=A0Jprotobuf?= =?UTF-8?q?=E7=9A=84=E6=9C=8D=E5=8A=A1=E5=99=A8=E5=90=AF=E5=8A=A8=E6=94=AF?= =?UTF-8?q?=E6=8C=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- net/pom.xml | 13 ++ .../core/jprotobuf/JProtobufTcpClient.java | 53 +++++++ .../core/jprotobuf/JProtobufTpcServer.java | 50 ++++++ .../jprotobuf/JProtobufTcpCodecHandler.java | 142 ++++++++++++++++++ .../net/packet/service/PacketService.java | 2 +- .../core/jprotobuf/client/TcpClientTest.java | 56 +++++++ .../server/JProtobufTcpController.java | 44 ++++++ .../server/JProtobufTcpServerTest.java | 40 +++++ .../jprotobuf/JProtobufHelloRequest.java | 47 ++++++ .../jprotobuf/JProtobufHelloResponse.java | 42 ++++++ net/src/test/resources/protocol.xml | 3 + .../zfoo/protocol/buffer/ByteBufUtils.java | 11 ++ 12 files changed, 502 insertions(+), 1 deletion(-) create mode 100644 net/src/main/java/com/zfoo/net/core/jprotobuf/JProtobufTcpClient.java create mode 100644 net/src/main/java/com/zfoo/net/core/jprotobuf/JProtobufTpcServer.java create mode 100644 net/src/main/java/com/zfoo/net/handler/codec/jprotobuf/JProtobufTcpCodecHandler.java create mode 100644 net/src/test/java/com/zfoo/net/core/jprotobuf/client/TcpClientTest.java create mode 100644 net/src/test/java/com/zfoo/net/core/jprotobuf/server/JProtobufTcpController.java create mode 100644 net/src/test/java/com/zfoo/net/core/jprotobuf/server/JProtobufTcpServerTest.java create mode 100644 net/src/test/java/com/zfoo/net/packet/jprotobuf/JProtobufHelloRequest.java create mode 100644 net/src/test/java/com/zfoo/net/packet/jprotobuf/JProtobufHelloResponse.java diff --git a/net/pom.xml b/net/pom.xml index 45a1ae06..88a0aaf5 100644 --- a/net/pom.xml +++ b/net/pom.xml @@ -134,6 +134,19 @@ ${netty.version} + + + com.baidu + jprotobuf + ${jprotobuf.version} + + + slf4j-api + org.slf4j + + + + org.javassist diff --git a/net/src/main/java/com/zfoo/net/core/jprotobuf/JProtobufTcpClient.java b/net/src/main/java/com/zfoo/net/core/jprotobuf/JProtobufTcpClient.java new file mode 100644 index 00000000..18dec41b --- /dev/null +++ b/net/src/main/java/com/zfoo/net/core/jprotobuf/JProtobufTcpClient.java @@ -0,0 +1,53 @@ +/* + * Copyright (C) 2020 The zfoo Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and limitations under the License. + */ + +package com.zfoo.net.core.jprotobuf; + +import com.zfoo.net.core.AbstractClient; +import com.zfoo.net.handler.ClientRouteHandler; +import com.zfoo.net.handler.codec.jprotobuf.JProtobufTcpCodecHandler; +import com.zfoo.net.handler.idle.ClientIdleHandler; +import com.zfoo.util.net.HostAndPort; +import io.netty.channel.Channel; +import io.netty.channel.ChannelInitializer; +import io.netty.channel.socket.SocketChannel; +import io.netty.handler.timeout.IdleStateHandler; + +/** + * @author jaysunxiao + * @version 3.0 + */ +public class JProtobufTcpClient extends AbstractClient { + + public JProtobufTcpClient(HostAndPort host) { + super(host); + } + + @Override + public ChannelInitializer channelChannelInitializer() { + return new ChannelHandlerInitializer(); + } + + + private static class ChannelHandlerInitializer extends ChannelInitializer { + @Override + protected void initChannel(SocketChannel channel) { + channel.pipeline().addLast(new IdleStateHandler(0, 0, 60)); + channel.pipeline().addLast(new ClientIdleHandler()); + channel.pipeline().addLast(new JProtobufTcpCodecHandler()); + channel.pipeline().addLast(new ClientRouteHandler()); + } + } + + +} diff --git a/net/src/main/java/com/zfoo/net/core/jprotobuf/JProtobufTpcServer.java b/net/src/main/java/com/zfoo/net/core/jprotobuf/JProtobufTpcServer.java new file mode 100644 index 00000000..d281ede8 --- /dev/null +++ b/net/src/main/java/com/zfoo/net/core/jprotobuf/JProtobufTpcServer.java @@ -0,0 +1,50 @@ +/* + * Copyright (C) 2020 The zfoo Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and limitations under the License. + */ + +package com.zfoo.net.core.jprotobuf; + +import com.zfoo.net.core.AbstractServer; +import com.zfoo.net.handler.ServerRouteHandler; +import com.zfoo.net.handler.codec.jprotobuf.JProtobufTcpCodecHandler; +import com.zfoo.net.handler.idle.ServerIdleHandler; +import com.zfoo.util.net.HostAndPort; +import io.netty.channel.ChannelInitializer; +import io.netty.channel.socket.SocketChannel; +import io.netty.handler.timeout.IdleStateHandler; + +/** + * @author jaysunxiao + * @version 3.0 + */ +public class JProtobufTpcServer extends AbstractServer { + + public JProtobufTpcServer(HostAndPort host) { + super(host); + } + + @Override + public ChannelInitializer channelChannelInitializer() { + return new ChannelHandlerInitializer(); + } + + + private static class ChannelHandlerInitializer extends ChannelInitializer { + @Override + protected void initChannel(SocketChannel channel) { + channel.pipeline().addLast(new IdleStateHandler(0, 0, 180)); + channel.pipeline().addLast(new ServerIdleHandler()); + channel.pipeline().addLast(new JProtobufTcpCodecHandler()); + channel.pipeline().addLast(new ServerRouteHandler()); + } + } +} diff --git a/net/src/main/java/com/zfoo/net/handler/codec/jprotobuf/JProtobufTcpCodecHandler.java b/net/src/main/java/com/zfoo/net/handler/codec/jprotobuf/JProtobufTcpCodecHandler.java new file mode 100644 index 00000000..b43d3203 --- /dev/null +++ b/net/src/main/java/com/zfoo/net/handler/codec/jprotobuf/JProtobufTcpCodecHandler.java @@ -0,0 +1,142 @@ +/* + * Copyright (C) 2020 The zfoo Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and limitations under the License. + */ + +package com.zfoo.net.handler.codec.jprotobuf; + +import com.baidu.bjf.remoting.protobuf.Codec; +import com.baidu.bjf.remoting.protobuf.ProtobufProxy; +import com.zfoo.net.packet.model.DecodedPacketInfo; +import com.zfoo.net.packet.model.EncodedPacketInfo; +import com.zfoo.net.packet.service.PacketService; +import com.zfoo.net.router.attachment.IAttachment; +import com.zfoo.net.util.SessionUtils; +import com.zfoo.protocol.IPacket; +import com.zfoo.protocol.ProtocolManager; +import com.zfoo.protocol.buffer.ByteBufUtils; +import com.zfoo.protocol.util.StringUtils; +import io.netty.buffer.ByteBuf; +import io.netty.channel.ChannelHandlerContext; +import io.netty.handler.codec.ByteToMessageCodec; +import io.netty.util.ReferenceCountUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.List; + +import static com.zfoo.protocol.ProtocolManager.protocols; + +/** + * header(4byte) + protocolId(2byte) + packet + * header = body(bytes.length) + protocolId.length(2byte) + * + * @author jaysunxiao + * @version 3.0 + */ +public class JProtobufTcpCodecHandler extends ByteToMessageCodec { + + private static final Logger logger = LoggerFactory.getLogger(JProtobufTcpCodecHandler.class); + + public static DecodedPacketInfo read(ByteBuf buffer) throws IOException { + var protocolId = ByteBufUtils.readShort(buffer); + var protocolRegistration = ProtocolManager.getProtocol(protocolId); + var protocolClass = protocols[protocolId].protocolConstructor().getDeclaringClass(); + + var protobufCodec = ProtobufProxy.create(protocolClass); + + var bytes = ByteBufUtils.readAllBytes(buffer); + var packet = protobufCodec.decode(bytes); + + // 解析包的附加包 + var hasAttachment = ByteBufUtils.tryReadBoolean(buffer); + IAttachment attachment = null; + if (hasAttachment) { + + } + return DecodedPacketInfo.valueOf((IPacket) packet, attachment); + } + + @Override + protected void decode(ChannelHandlerContext ctx, ByteBuf in, List out) { + // 不够读一个int + if (in.readableBytes() <= PacketService.PACKET_HEAD_LENGTH) { + return; + } + in.markReaderIndex(); + var length = in.readInt(); + + // 如果长度非法，则抛出异常断开连接 + if (length < 0) { + throw new IllegalArgumentException(StringUtils.format("[session:{}]的包头长度[length:{}]非法" + , SessionUtils.sessionInfo(ctx), length)); + } + + // ByteBuf里的数据太小 + if (in.readableBytes() < length) { + in.resetReaderIndex(); + return; + } + + ByteBuf tmpByteBuf = null; + try { + tmpByteBuf = in.readRetainedSlice(length); + DecodedPacketInfo packetInfo = read(tmpByteBuf); + out.add(packetInfo); + } catch (Exception e) { + logger.error("[session:{}]解码exception异常", SessionUtils.sessionInfo(ctx), e); + } catch (Throwable t) { + logger.error("[session:{}]解码throwable错误", SessionUtils.sessionInfo(ctx), t); + } finally { + ReferenceCountUtil.release(tmpByteBuf); + } + } + + @Override + protected void encode(ChannelHandlerContext ctx, EncodedPacketInfo packetInfo, ByteBuf out) { + try { + write(out, packetInfo.getPacket(), packetInfo.getAttachment()); + } catch (Exception e) { + logger.error("[session:{}][{}]编码exception异常", SessionUtils.sessionInfo(ctx), packetInfo.getPacket().getClass().getSimpleName(), e); + } catch (Throwable t) { + logger.error("[session:{}][{}]编码throwable错误", SessionUtils.sessionInfo(ctx), packetInfo.getPacket().getClass().getSimpleName(), t); + } + } + + public void write(ByteBuf buffer, IPacket packet, IAttachment attachment) throws IOException { + if (packet == null) { + logger.error("packet is null and can not be sent."); + return; + } + + // 预留写入包的长度，一个int字节大小 + buffer.writeInt(PacketService.PACKET_HEAD_LENGTH); + var protocolId = packet.protocolId(); + // 写入协议号 + ByteBufUtils.writeShort(buffer, protocolId); + + var protobufCodec = (Codec) ProtobufProxy.create(packet.getClass()); + byte[] bytes = protobufCodec.encode(packet); + + buffer.writeBytes(bytes); + + int length = buffer.readableBytes(); + + int packetLength = length - PacketService.PACKET_HEAD_LENGTH; + + buffer.writerIndex(0); + + buffer.writeInt(packetLength); + + buffer.writerIndex(length); + } +} diff --git a/net/src/main/java/com/zfoo/net/packet/service/PacketService.java b/net/src/main/java/com/zfoo/net/packet/service/PacketService.java index d7750e7b..3b45db7c 100644 --- a/net/src/main/java/com/zfoo/net/packet/service/PacketService.java +++ b/net/src/main/java/com/zfoo/net/packet/service/PacketService.java @@ -134,7 +134,7 @@ public class PacketService implements IPacketService { // 解析包体 var packet = ProtocolManager.read(buffer); // 解析包的附加包 - var hasAttachment = ByteBufUtils.readBoolean(buffer); + var hasAttachment = ByteBufUtils.tryReadBoolean(buffer); var attachment = hasAttachment ? ((IAttachment) ProtocolManager.read(buffer)) : null; return DecodedPacketInfo.valueOf(packet, attachment); } diff --git a/net/src/test/java/com/zfoo/net/core/jprotobuf/client/TcpClientTest.java b/net/src/test/java/com/zfoo/net/core/jprotobuf/client/TcpClientTest.java new file mode 100644 index 00000000..6f25c1ff --- /dev/null +++ b/net/src/test/java/com/zfoo/net/core/jprotobuf/client/TcpClientTest.java @@ -0,0 +1,56 @@ +/* + * Copyright (C) 2020 The zfoo Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and limitations under the License. + */ + +package com.zfoo.net.core.jprotobuf.client; + +import com.zfoo.net.NetContext; +import com.zfoo.net.core.jprotobuf.JProtobufTcpClient; +import com.zfoo.net.packet.jprotobuf.JProtobufHelloRequest; +import com.zfoo.net.packet.jprotobuf.JProtobufHelloResponse; +import com.zfoo.protocol.util.JsonUtils; +import com.zfoo.util.ThreadUtils; +import com.zfoo.util.net.HostAndPort; +import org.junit.Ignore; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.context.support.ClassPathXmlApplicationContext; + +/** + * @author jaysunxiao + * @version 3.0 + */ +@Ignore +public class TcpClientTest { + + private static final Logger logger = LoggerFactory.getLogger(TcpClientTest.class); + + @Test + public void startClient() throws Exception { + var context = new ClassPathXmlApplicationContext("config.xml"); + + var client = new JProtobufTcpClient(HostAndPort.valueOf("127.0.0.1:9000")); + var session = client.start(); + + for (int i = 0; i < 1000; i++) { + var ask = new JProtobufHelloRequest(); + ask.setMessage("Hello, this is jprotobuf client!"); + var answer = NetContext.getRouter().syncAsk(session, ask, JProtobufHelloResponse.class, null).packet(); + logger.info("同步请求收到结果[{}]", JsonUtils.object2String(answer)); + ThreadUtils.sleep(1000); + } + + ThreadUtils.sleep(Long.MAX_VALUE); + } + +} diff --git a/net/src/test/java/com/zfoo/net/core/jprotobuf/server/JProtobufTcpController.java b/net/src/test/java/com/zfoo/net/core/jprotobuf/server/JProtobufTcpController.java new file mode 100644 index 00000000..abe37db3 --- /dev/null +++ b/net/src/test/java/com/zfoo/net/core/jprotobuf/server/JProtobufTcpController.java @@ -0,0 +1,44 @@ +/* + * Copyright (C) 2020 The zfoo Authors + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and limitations under the License. + */ + +package com.zfoo.net.core.jprotobuf.server; + +import com.zfoo.net.NetContext; +import com.zfoo.net.packet.jprotobuf.JProtobufHelloRequest; +import com.zfoo.net.packet.jprotobuf.JProtobufHelloResponse; +import com.zfoo.net.router.receiver.PacketReceiver; +import com.zfoo.net.session.model.Session; +import com.zfoo.protocol.util.JsonUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Component; + +/** + * @author jaysunxiao + * @version 3.0 + */ +@Component +public class JProtobufTcpController { + + private static final Logger logger = LoggerFactory.getLogger(JProtobufTcpController.class); + + @PacketReceiver + public void atJProtobufHelloRequest(Session session, JProtobufHelloRequest ask) { + logger.info("receive [packet:{}] from client", JsonUtils.object2String(ask)); + + var answer = new JProtobufHelloResponse(); + answer.setMessage("Hello, this is the jprotobuf tcp server!"); + + NetContext.getRouter().send(session, answer); + } + +} diff --git a/net/src/test/java/com/zfoo/net/core/jprotobuf/server/JProtobufTcpServerTest.java b/net/src/test/java/com/zfoo/net/core/jprotobuf/server/JProtobufTcpServerTest.java new file mode 100644 index 00000000..9d98d456 --- /dev/null +++ b/net/src/test/java/com/zfoo/net/core/jprotobuf/server/JProtobufTcpServerTest.java @@ -0,0 +1,40 @@ +/* + * Copyright (C) 2020 The zfoo Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and limitations under the License. + */ + +package com.zfoo.net.core.jprotobuf.server; + +import com.zfoo.net.core.jprotobuf.JProtobufTpcServer; +import com.zfoo.util.ThreadUtils; +import com.zfoo.util.net.HostAndPort; +import org.junit.Ignore; +import org.junit.Test; +import org.springframework.context.support.ClassPathXmlApplicationContext; + +/** + * @author jaysunxiao + * @version 3.0 + */ +@Ignore +public class JProtobufTcpServerTest { + + @Test + public void startServer() { + var context = new ClassPathXmlApplicationContext("config.xml"); + + var server = new JProtobufTpcServer(HostAndPort.valueOf("127.0.0.1:9000")); + server.start(); + + ThreadUtils.sleep(Long.MAX_VALUE); + } + +} diff --git a/net/src/test/java/com/zfoo/net/packet/jprotobuf/JProtobufHelloRequest.java b/net/src/test/java/com/zfoo/net/packet/jprotobuf/JProtobufHelloRequest.java new file mode 100644 index 00000000..53afc5e0 --- /dev/null +++ b/net/src/test/java/com/zfoo/net/packet/jprotobuf/JProtobufHelloRequest.java @@ -0,0 +1,47 @@ +/* + * Copyright (C) 2020 The zfoo Authors + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and limitations under the License. + */ + +package com.zfoo.net.packet.jprotobuf; + +import com.baidu.bjf.remoting.protobuf.annotation.Protobuf; +import com.zfoo.protocol.IPacket; + +/** + * @author jaysunxiao + * @version 3.0 + */ +public class JProtobufHelloRequest implements IPacket { + + public static final transient short PROTOCOL_ID = 1500; + + @Protobuf(order = 1) + private String message; + + public static JProtobufHelloRequest valueOf(String message) { + var request = new JProtobufHelloRequest(); + request.message = message; + return request; + } + + @Override + public short protocolId() { + return PROTOCOL_ID; + } + + public String getMessage() { + return message; + } + + public void setMessage(String message) { + this.message = message; + } +} diff --git a/net/src/test/java/com/zfoo/net/packet/jprotobuf/JProtobufHelloResponse.java b/net/src/test/java/com/zfoo/net/packet/jprotobuf/JProtobufHelloResponse.java new file mode 100644 index 00000000..8bdc1710 --- /dev/null +++ b/net/src/test/java/com/zfoo/net/packet/jprotobuf/JProtobufHelloResponse.java @@ -0,0 +1,42 @@ +/* + * Copyright (C) 2020 The zfoo Authors + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and limitations under the License. + */ + +package com.zfoo.net.packet.jprotobuf; + +import com.baidu.bjf.remoting.protobuf.annotation.Protobuf; +import com.zfoo.protocol.IPacket; + +/** + * @author jaysunxiao + * @version 3.0 + */ +public class JProtobufHelloResponse implements IPacket { + + public static final transient short PROTOCOL_ID = 1501; + + @Protobuf(order = 1) + private String message; + + + @Override + public short protocolId() { + return PROTOCOL_ID; + } + + public String getMessage() { + return message; + } + + public void setMessage(String message) { + this.message = message; + } +} diff --git a/net/src/test/resources/protocol.xml b/net/src/test/resources/protocol.xml index 913f76e4..d2e6a651 100644 --- a/net/src/test/resources/protocol.xml +++ b/net/src/test/resources/protocol.xml @@ -66,6 +66,9 @@ + + + diff --git a/protocol/src/main/java/com/zfoo/protocol/buffer/ByteBufUtils.java b/protocol/src/main/java/com/zfoo/protocol/buffer/ByteBufUtils.java index 5ea19802..46027421 100644 --- a/protocol/src/main/java/com/zfoo/protocol/buffer/ByteBufUtils.java +++ b/protocol/src/main/java/com/zfoo/protocol/buffer/ByteBufUtils.java @@ -46,6 +46,10 @@ public abstract class ByteBufUtils { return byteBuf.readBoolean(); } + public static boolean tryReadBoolean(ByteBuf byteBuf) { + return byteBuf.isReadable() && readBoolean(byteBuf); + } + public static void writeBooleanBox(ByteBuf byteBuf, Boolean value) { byteBuf.writeBoolean(value != null && value); } @@ -71,6 +75,13 @@ public abstract class ByteBufUtils { return byteBuf.readByte(); } + public static byte[] readAllBytes(ByteBuf byteBuf) { + var readableBytes = byteBuf.readableBytes(); + var bytes = new byte[readableBytes]; + byteBuf.readBytes(bytes); + return bytes; + } + //---------------------------------short-------------------------------------- public static void writeShort(ByteBuf byteBuf, short value) { byteBuf.writeShort(value);