From d154dd8c12e8ef2381787c6078f1c539fc8b3696 Mon Sep 17 00:00:00 2001 From: jaysunxiao Date: Wed, 20 Oct 2021 17:55:41 +0800 Subject: [PATCH] =?UTF-8?q?ref[net]:=20=E9=87=8D=E6=9E=84net=EF=BC=8C?= =?UTF-8?q?=E4=B8=BB=E8=A6=81=E4=BF=AE=E6=94=B9=E7=B1=BB=E5=90=8D=E7=A7=B0?= =?UTF-8?q?=EF=BC=8C=E8=B0=83=E6=95=B4=E5=8C=85=E7=9A=84=E4=BD=8D=E7=BD=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 2 +- doc/FAQ.md | 3 +- .../java/com/zfoo/net/consumer/Consumer.java | 30 ++-- .../balancer/IConsumerLoadBalancer.java | 6 +- .../ShortestTimeConsumerLoadBalancer.java | 6 +- .../zfoo/net/handler/BaseRouteHandler.java | 2 +- .../zfoo/net/handler/GatewayRouteHandler.java | 22 +-- .../handler/codec/http/HttpCodecHandler.java | 6 +- .../handler/codec/tcp/TcpCodecHandler.java | 2 +- .../handler/codec/udp/UdpCodecHandler.java | 10 +- .../websocket/WebSocketCodecHandler.java | 2 +- .../net/packet/model/DecodedPacketInfo.java | 15 +- .../net/packet/model/EncodedPacketInfo.java | 15 +- .../net/packet/service/IPacketService.java | 4 +- .../net/packet/service/PacketService.java | 16 +- .../java/com/zfoo/net/router/IRouter.java | 8 +- .../main/java/com/zfoo/net/router/Router.java | 140 +++++++++--------- .../zfoo/net/router/answer/AsyncAnswer.java | 14 +- .../zfoo/net/router/answer/ISyncAnswer.java | 4 +- .../zfoo/net/router/answer/SyncAnswer.java | 10 +- .../attachment/AttachmentType.java} | 30 ++-- .../attachment/GatewayAttachment.java} | 27 ++-- .../attachment/HttpAttachment.java} | 13 +- .../attachment/IAttachment.java} | 7 +- .../attachment}/NoAnswerAttachment.java | 9 +- .../attachment/SignalAttachment.java} | 31 ++-- .../attachment/UdpAttachment.java} | 13 +- .../net/router/receiver/EnhanceUtils.java | 6 +- .../net/router/receiver/IPacketReceiver.java | 4 +- .../receiver/PacketReceiverDefinition.java | 6 +- .../zfoo/net/router/route/PacketRoute.java | 26 ++-- .../{PacketSignal.java => SignalBridge.java} | 38 ++--- .../ConsistentHashTaskDispatch.java | 8 +- .../net/task/model/PacketReceiverTask.java | 18 +-- .../gateway/GatewayProviderController.java | 4 +- .../udp/client/UdpClientPacketController.java | 4 +- .../net/core/udp/client/UdpClientTest.java | 4 +- .../udp/server/UdpServerPacketController.java | 4 +- .../com/zfoo/net/protocol/ProtocolTest.java | 39 +++-- .../zfoo/net/router/PacketSignalArray.java | 12 +- .../com/zfoo/net/router/PacketSignalMap.java | 10 +- ...tSignalTest.java => SignalBridgeTest.java} | 14 +- net/src/test/resources/protocol.xml | 10 +- 43 files changed, 325 insertions(+), 329 deletions(-) rename net/src/main/java/com/zfoo/net/{packet/model/PacketAttachmentType.java => router/attachment/AttachmentType.java} (61%) rename net/src/main/java/com/zfoo/net/{packet/model/GatewayPacketAttachment.java => router/attachment/GatewayAttachment.java} (80%) rename net/src/main/java/com/zfoo/net/{packet/model/HttpPacketAttachment.java => router/attachment/HttpAttachment.java} (88%) rename net/src/main/java/com/zfoo/net/{packet/model/IPacketAttachment.java => router/attachment/IAttachment.java} (86%) rename net/src/main/java/com/zfoo/net/{packet/model => router/attachment}/NoAnswerAttachment.java (88%) rename net/src/main/java/com/zfoo/net/{packet/model/SignalPacketAttachment.java => router/attachment/SignalAttachment.java} (77%) rename net/src/main/java/com/zfoo/net/{packet/model/UdpPacketAttachment.java => router/attachment/UdpAttachment.java} (80%) rename net/src/main/java/com/zfoo/net/router/route/{PacketSignal.java => SignalBridge.java} (54%) rename net/src/test/java/com/zfoo/net/router/{PacketSignalTest.java => SignalBridgeTest.java} (91%) diff --git a/README.md b/README.md index 18b533f1..910a21e1 100644 --- a/README.md +++ b/README.md @@ -23,7 +23,7 @@ Ⅰ. zfoo简介🚩 ----------- -- **性能炸裂，天生异步，Actor设计思想，无锁化设计，基于Spring的MVC式用法的万能RPC框架** +- **极致性能，天生异步，Actor设计思想，无锁化设计，基于Spring的MVC式用法的万能RPC框架** - **极致序列化**，原生集成的目前二进制序列化和反序列化速度最快的 [zfoo protocol](protocol/README.md) 作为网络通讯协议 - **高可拓展性**，单台服务器部署，集群部署，注册中心加集群部署，网关加集群部署，随意搭配 - **上能做游戏服务器框架，下能做应用网站服务器框架** diff --git a/doc/FAQ.md b/doc/FAQ.md index 248f1edf..1c4f1e62 100644 --- a/doc/FAQ.md +++ b/doc/FAQ.md @@ -16,6 +16,7 @@ - 并不是所有编程语言都有interface的，你用了interface的话其它编程语言就没法用了，就无法跨平台了 - 基于packet的容易扩展，可以轻易跨平台 -#### 接下来的计划 +#### TODO： - 参看一下akka的actor实现，进一步优化net的线程模型，统一异步回调的argument参数 +- protocol需要支持更多的语言，python，go，C++ diff --git a/net/src/main/java/com/zfoo/net/consumer/Consumer.java b/net/src/main/java/com/zfoo/net/consumer/Consumer.java index d3a52182..740b0533 100644 --- a/net/src/main/java/com/zfoo/net/consumer/Consumer.java +++ b/net/src/main/java/com/zfoo/net/consumer/Consumer.java @@ -15,15 +15,15 @@ package com.zfoo.net.consumer; import com.zfoo.net.NetContext; import com.zfoo.net.packet.common.Error; -import com.zfoo.net.packet.model.NoAnswerAttachment; -import com.zfoo.net.packet.model.SignalPacketAttachment; import com.zfoo.net.router.Router; import com.zfoo.net.router.answer.AsyncAnswer; import com.zfoo.net.router.answer.SyncAnswer; +import com.zfoo.net.router.attachment.NoAnswerAttachment; +import com.zfoo.net.router.attachment.SignalAttachment; import com.zfoo.net.router.exception.ErrorResponseException; import com.zfoo.net.router.exception.NetTimeOutException; import com.zfoo.net.router.exception.UnexpectedProtocolException; -import com.zfoo.net.router.route.PacketSignal; +import com.zfoo.net.router.route.SignalBridge; import com.zfoo.protocol.IPacket; import com.zfoo.protocol.util.JsonUtils; import com.zfoo.protocol.util.StringUtils; @@ -67,19 +67,19 @@ public class Consumer implements IConsumer { // 下面的代码逻辑同Router的syncAsk，如果修改的话，记得一起修改 - var clientAttachment = new SignalPacketAttachment(); + var clientSignalAttachment = new SignalAttachment(); var executorConsistentHash = (argument == null) ? RandomUtils.randomInt() : HashUtils.fnvHash(argument); - clientAttachment.setExecutorConsistentHash(executorConsistentHash); + clientSignalAttachment.setExecutorConsistentHash(executorConsistentHash); try { - PacketSignal.addSignalAttachment(clientAttachment); + SignalBridge.addSignalAttachment(clientSignalAttachment); // load balancer之前调用 - loadBalancer.beforeLoadBalancer(session, packet, clientAttachment); + loadBalancer.beforeLoadBalancer(session, packet, clientSignalAttachment); - NetContext.getRouter().send(session, packet, clientAttachment); + NetContext.getRouter().send(session, packet, clientSignalAttachment); - IPacket responsePacket = clientAttachment.getResponseFuture().get(Router.DEFAULT_TIMEOUT, TimeUnit.MILLISECONDS); + IPacket responsePacket = clientSignalAttachment.getResponseFuture().get(Router.DEFAULT_TIMEOUT, TimeUnit.MILLISECONDS); if (responsePacket.protocolId() == Error.errorProtocolId()) { throw new ErrorResponseException((Error) responsePacket); @@ -88,16 +88,16 @@ public class Consumer implements IConsumer { throw new UnexpectedProtocolException(StringUtils.format("client expect protocol:[{}], but found protocol:[{}]" , answerClass, responsePacket.getClass().getName())); } - var syncAnswer = new SyncAnswer<>((T) responsePacket, clientAttachment); + var syncAnswer = new SyncAnswer<>((T) responsePacket, clientSignalAttachment); // load balancer之后调用 - loadBalancer.afterLoadBalancer(session, packet, clientAttachment); + loadBalancer.afterLoadBalancer(session, packet, clientSignalAttachment); return syncAnswer; } catch (TimeoutException e) { throw new NetTimeOutException(StringUtils.format("syncRequest timeout exception, ask:[{}], attachment:[{}]" - , JsonUtils.object2String(packet), JsonUtils.object2String(clientAttachment))); + , JsonUtils.object2String(packet), JsonUtils.object2String(clientSignalAttachment))); } finally { - PacketSignal.removeSignalAttachment(clientAttachment); + SignalBridge.removeSignalAttachment(clientSignalAttachment); } } @@ -108,10 +108,10 @@ public class Consumer implements IConsumer { var asyncAnswer = NetContext.getRouter().asyncAsk(session, packet, answerClass, argument); // load balancer之前调用 - loadBalancer.beforeLoadBalancer(session, packet, asyncAnswer.getFutureAttachment()); + loadBalancer.beforeLoadBalancer(session, packet, asyncAnswer.getSignalAttachment()); // load balancer之后调用 - asyncAnswer.thenAccept(responsePacket -> loadBalancer.afterLoadBalancer(session, packet, asyncAnswer.getFutureAttachment())); + asyncAnswer.thenAccept(responsePacket -> loadBalancer.afterLoadBalancer(session, packet, asyncAnswer.getSignalAttachment())); return asyncAnswer; } diff --git a/net/src/main/java/com/zfoo/net/consumer/balancer/IConsumerLoadBalancer.java b/net/src/main/java/com/zfoo/net/consumer/balancer/IConsumerLoadBalancer.java index a9aed868..85f3f92a 100644 --- a/net/src/main/java/com/zfoo/net/consumer/balancer/IConsumerLoadBalancer.java +++ b/net/src/main/java/com/zfoo/net/consumer/balancer/IConsumerLoadBalancer.java @@ -13,7 +13,7 @@ package com.zfoo.net.consumer.balancer; -import com.zfoo.net.packet.model.SignalPacketAttachment; +import com.zfoo.net.router.attachment.SignalAttachment; import com.zfoo.net.session.model.Session; import com.zfoo.protocol.IPacket; import org.springframework.lang.Nullable; @@ -33,10 +33,10 @@ public interface IConsumerLoadBalancer { */ Session loadBalancer(IPacket packet, @Nullable Object argument); - default void beforeLoadBalancer(Session session, IPacket packet, SignalPacketAttachment attachment) { + default void beforeLoadBalancer(Session session, IPacket packet, SignalAttachment attachment) { } - default void afterLoadBalancer(Session session, IPacket packet, SignalPacketAttachment attachment) { + default void afterLoadBalancer(Session session, IPacket packet, SignalAttachment attachment) { } } diff --git a/net/src/main/java/com/zfoo/net/consumer/balancer/ShortestTimeConsumerLoadBalancer.java b/net/src/main/java/com/zfoo/net/consumer/balancer/ShortestTimeConsumerLoadBalancer.java index 16e5a8a4..6dda500e 100644 --- a/net/src/main/java/com/zfoo/net/consumer/balancer/ShortestTimeConsumerLoadBalancer.java +++ b/net/src/main/java/com/zfoo/net/consumer/balancer/ShortestTimeConsumerLoadBalancer.java @@ -13,7 +13,7 @@ package com.zfoo.net.consumer.balancer; -import com.zfoo.net.packet.model.SignalPacketAttachment; +import com.zfoo.net.router.attachment.SignalAttachment; import com.zfoo.net.session.model.AttributeType; import com.zfoo.net.session.model.Session; import com.zfoo.protocol.IPacket; @@ -74,13 +74,13 @@ public class ShortestTimeConsumerLoadBalancer extends AbstractConsumerLoadBalanc } @Override - public void beforeLoadBalancer(Session session, IPacket packet, SignalPacketAttachment attachment) { + public void beforeLoadBalancer(Session session, IPacket packet, SignalAttachment attachment) { // 因为要通过最短响应时间来路由分发消息，这里使用更精确的时间 attachment.setTimestamp(TimeUtils.currentTimeMillis()); } @Override - public void afterLoadBalancer(Session session, IPacket packet, SignalPacketAttachment attachment) { + public void afterLoadBalancer(Session session, IPacket packet, SignalAttachment attachment) { var map = (Map) session.getAttribute(AttributeType.RESPONSE_TIME); if (map == null) { map = new ConcurrentHashMap<>(); diff --git a/net/src/main/java/com/zfoo/net/handler/BaseRouteHandler.java b/net/src/main/java/com/zfoo/net/handler/BaseRouteHandler.java index af3943e1..68eb8824 100644 --- a/net/src/main/java/com/zfoo/net/handler/BaseRouteHandler.java +++ b/net/src/main/java/com/zfoo/net/handler/BaseRouteHandler.java @@ -65,7 +65,7 @@ public class BaseRouteHandler extends ChannelInboundHandlerAdapter { return; } DecodedPacketInfo decodedPacketInfo = (DecodedPacketInfo) msg; - NetContext.getRouter().receive(session, decodedPacketInfo.getPacket(), decodedPacketInfo.getPacketAttachment()); + NetContext.getRouter().receive(session, decodedPacketInfo.getPacket(), decodedPacketInfo.getAttachment()); } @Override diff --git a/net/src/main/java/com/zfoo/net/handler/GatewayRouteHandler.java b/net/src/main/java/com/zfoo/net/handler/GatewayRouteHandler.java index 317910fa..6785391d 100644 --- a/net/src/main/java/com/zfoo/net/handler/GatewayRouteHandler.java +++ b/net/src/main/java/com/zfoo/net/handler/GatewayRouteHandler.java @@ -22,9 +22,9 @@ import com.zfoo.net.packet.common.Heartbeat; import com.zfoo.net.packet.common.Ping; import com.zfoo.net.packet.common.Pong; import com.zfoo.net.packet.model.DecodedPacketInfo; -import com.zfoo.net.packet.model.GatewayPacketAttachment; -import com.zfoo.net.packet.model.IPacketAttachment; -import com.zfoo.net.packet.model.SignalPacketAttachment; +import com.zfoo.net.router.attachment.GatewayAttachment; +import com.zfoo.net.router.attachment.IAttachment; +import com.zfoo.net.router.attachment.SignalAttachment; import com.zfoo.net.session.model.AttributeType; import com.zfoo.net.session.model.Session; import com.zfoo.net.util.SessionUtils; @@ -48,7 +48,7 @@ public class GatewayRouteHandler extends ServerRouteHandler { private static final Logger logger = LoggerFactory.getLogger(GatewayRouteHandler.class); - private BiFunction packetFilter; + private final BiFunction packetFilter; public GatewayRouteHandler(BiFunction packetFilter) { this.packetFilter = packetFilter; @@ -78,33 +78,33 @@ public class GatewayRouteHandler extends ServerRouteHandler { , SessionUtils.sessionInfo(ctx), JsonUtils.object2String(packet))); } - var signalAttachment = (SignalPacketAttachment) decodedPacketInfo.getPacketAttachment(); - var gatewayPacketAttachment = new GatewayPacketAttachment(session, signalAttachment); + var signalAttachment = (SignalAttachment) decodedPacketInfo.getAttachment(); + var gatewayAttachment = new GatewayAttachment(session, signalAttachment); // 网关优先使用IGatewayLoadBalancer作为一致性hash的计算参数，然后才会使用客户端的session做参数 if (packet instanceof IGatewayLoadBalancer) { var loadBalancerConsistentHashObject = ((IGatewayLoadBalancer) packet).loadBalancerConsistentHashObject(); - gatewayPacketAttachment.useExecutorConsistentHash(loadBalancerConsistentHashObject); - forwardingPacket(packet, gatewayPacketAttachment, loadBalancerConsistentHashObject); + gatewayAttachment.useExecutorConsistentHash(loadBalancerConsistentHashObject); + forwardingPacket(packet, gatewayAttachment, loadBalancerConsistentHashObject); return; } else { // 使用用户的uid做一致性hash var uid = (Long) session.getAttribute(AttributeType.UID); if (uid != null) { - forwardingPacket(packet, gatewayPacketAttachment, uid); + forwardingPacket(packet, gatewayAttachment, uid); return; } } // 再使用session的sid做一致性hash，因为每次客户端连接过来sid都会改变，所以客户端重写建立连接的话可能会被路由到其它的服务器 // 如果有特殊需求的话，可以考虑去重写网关的转发策略 var sid = session.getSid(); - forwardingPacket(packet, gatewayPacketAttachment, sid); + forwardingPacket(packet, gatewayAttachment, sid); } /** * 转发网关收到的包 */ - private void forwardingPacket(IPacket packet, IPacketAttachment attachment, Object argument) { + private void forwardingPacket(IPacket packet, IAttachment attachment, Object argument) { try { var consumerSession = ConsistentHashConsumerLoadBalancer.getInstance().loadBalancer(packet, argument); NetContext.getRouter().send(consumerSession, packet, attachment); diff --git a/net/src/main/java/com/zfoo/net/handler/codec/http/HttpCodecHandler.java b/net/src/main/java/com/zfoo/net/handler/codec/http/HttpCodecHandler.java index 898c7b55..7a38253d 100644 --- a/net/src/main/java/com/zfoo/net/handler/codec/http/HttpCodecHandler.java +++ b/net/src/main/java/com/zfoo/net/handler/codec/http/HttpCodecHandler.java @@ -15,7 +15,7 @@ package com.zfoo.net.handler.codec.http; import com.zfoo.net.packet.common.Message; import com.zfoo.net.packet.model.DecodedPacketInfo; import com.zfoo.net.packet.model.EncodedPacketInfo; -import com.zfoo.net.packet.model.HttpPacketAttachment; +import com.zfoo.net.router.attachment.HttpAttachment; import com.zfoo.protocol.IPacket; import com.zfoo.protocol.util.JsonUtils; import com.zfoo.protocol.util.StringUtils; @@ -37,7 +37,7 @@ public class HttpCodecHandler extends MessageToMessageCodec uriResolver; + private final Function uriResolver; public HttpCodecHandler(Function uriResolver) { super(); @@ -62,7 +62,7 @@ public class HttpCodecHandler extends MessageToMessageCodec list) { try { var packet = (IPacket) out.getPacket(); - var attachment = (HttpPacketAttachment) out.getPacketAttachment(); + var attachment = (HttpAttachment) out.getAttachment(); var protocolVersion = attachment.getFullHttpRequest().protocolVersion(); var httpResponseStatus = attachment.getHttpResponseStatus(); diff --git a/net/src/main/java/com/zfoo/net/handler/codec/tcp/TcpCodecHandler.java b/net/src/main/java/com/zfoo/net/handler/codec/tcp/TcpCodecHandler.java index 98508dad..2f3caf53 100644 --- a/net/src/main/java/com/zfoo/net/handler/codec/tcp/TcpCodecHandler.java +++ b/net/src/main/java/com/zfoo/net/handler/codec/tcp/TcpCodecHandler.java @@ -79,7 +79,7 @@ public class TcpCodecHandler extends ByteToMessageCodec { @Override protected void encode(ChannelHandlerContext ctx, EncodedPacketInfo packetInfo, ByteBuf out) { try { - NetContext.getPacketService().write(out, packetInfo.getPacket(), packetInfo.getPacketAttachment()); + NetContext.getPacketService().write(out, packetInfo.getPacket(), packetInfo.getAttachment()); } catch (Exception e) { logger.error("[session:{}][{}]编码exception异常", SessionUtils.sessionInfo(ctx), packetInfo.getPacket().getClass().getSimpleName(), e); throw e; diff --git a/net/src/main/java/com/zfoo/net/handler/codec/udp/UdpCodecHandler.java b/net/src/main/java/com/zfoo/net/handler/codec/udp/UdpCodecHandler.java index 21c8a7df..40c9dea0 100644 --- a/net/src/main/java/com/zfoo/net/handler/codec/udp/UdpCodecHandler.java +++ b/net/src/main/java/com/zfoo/net/handler/codec/udp/UdpCodecHandler.java @@ -15,8 +15,8 @@ package com.zfoo.net.handler.codec.udp; import com.zfoo.net.NetContext; import com.zfoo.net.packet.model.DecodedPacketInfo; import com.zfoo.net.packet.model.EncodedPacketInfo; -import com.zfoo.net.packet.model.UdpPacketAttachment; import com.zfoo.net.packet.service.PacketService; +import com.zfoo.net.router.attachment.UdpAttachment; import com.zfoo.net.util.SessionUtils; import com.zfoo.protocol.util.JsonUtils; import com.zfoo.protocol.util.StringUtils; @@ -68,7 +68,7 @@ public class UdpCodecHandler extends MessageToMessageCodec list) { try { var byteBuf = channelHandlerContext.alloc().ioBuffer(); - var udpPacketAttachment = (UdpPacketAttachment) out.getPacketAttachment(); + var udpAttachment = (UdpAttachment) out.getAttachment(); - NetContext.getPacketService().write(byteBuf, out.getPacket(), out.getPacketAttachment()); - list.add(new DatagramPacket(byteBuf, new InetSocketAddress(udpPacketAttachment.getHost(), udpPacketAttachment.getPort()))); + NetContext.getPacketService().write(byteBuf, out.getPacket(), out.getAttachment()); + list.add(new DatagramPacket(byteBuf, new InetSocketAddress(udpAttachment.getHost(), udpAttachment.getPort()))); } catch (Exception e) { logger.error("[{}]编码exception异常", JsonUtils.object2String(out), e); throw e; diff --git a/net/src/main/java/com/zfoo/net/handler/codec/websocket/WebSocketCodecHandler.java b/net/src/main/java/com/zfoo/net/handler/codec/websocket/WebSocketCodecHandler.java index 1a2c948d..a0468f27 100644 --- a/net/src/main/java/com/zfoo/net/handler/codec/websocket/WebSocketCodecHandler.java +++ b/net/src/main/java/com/zfoo/net/handler/codec/websocket/WebSocketCodecHandler.java @@ -87,7 +87,7 @@ public class WebSocketCodecHandler extends MessageToMessageCodec serverReceiveSignalPacketAttachment = new FastThreadLocal<>(); + private final FastThreadLocal serverReceiveSignalAttachmentThreadLocal = new FastThreadLocal<>(); @Override - public void receive(Session session, IPacket packet, @Nullable IPacketAttachment packetAttachment) { + public void receive(Session session, IPacket packet, @Nullable IAttachment attachment) { if (packet.protocolId() == Heartbeat.heartbeatProtocolId()) { logger.info("heartbeat"); return; } // 发送者（客户端）同步和异步消息的接收，发送者通过packetId判断重复 - if (packetAttachment != null) { - switch (packetAttachment.packetType()) { + if (attachment != null) { + switch (attachment.packetType()) { case SIGNAL_PACKET: - var signalPacketAttachment = (SignalPacketAttachment) packetAttachment; + var signalAttachment = (SignalAttachment) attachment; - if (signalPacketAttachment.isClient()) { - // 服务器收到signalPacketAttachment，不做任何处理 - signalPacketAttachment.setClient(false); + if (signalAttachment.isClient()) { + // 服务器收到signalAttachment，不做任何处理 + signalAttachment.setClient(false); } else { // 客户端收到服务器应答，客户端发送的时候isClient为true，服务器收到的时候将其设置为false - var attachment = (SignalPacketAttachment) PacketSignal.removeSignalAttachment(signalPacketAttachment); - if (attachment != null) { - attachment.getResponseFuture().complete(packet); + var removedAttachment = (SignalAttachment) SignalBridge.removeSignalAttachment(signalAttachment); + if (removedAttachment != null) { + removedAttachment.getResponseFuture().complete(packet); } else { - logger.error("client receives packet:[{}] and packetAttachment:[{}] from server, but clientPacketAttachmentMap has no attachment, perhaps timeout exception." - , JsonUtils.object2String(packet), JsonUtils.object2String(packetAttachment)); + logger.error("client receives packet:[{}] and attachment:[{}] from server, but clientAttachmentMap has no attachment, perhaps timeout exception." + , JsonUtils.object2String(packet), JsonUtils.object2String(attachment)); } return; } break; case GATEWAY_PACKET: - var gatewayPacketAttachment = (GatewayPacketAttachment) packetAttachment; - if (gatewayPacketAttachment.isClient()) { - gatewayPacketAttachment.setClient(false); + var gatewayAttachment = (GatewayAttachment) attachment; + if (gatewayAttachment.isClient()) { + gatewayAttachment.setClient(false); } else { - var gatewaySession = NetContext.getSessionManager().getServerSession(gatewayPacketAttachment.getSid()); + var gatewaySession = NetContext.getSessionManager().getServerSession(gatewayAttachment.getSid()); if (gatewaySession != null) { - var signalAttachment = gatewayPacketAttachment.getSignalPacketAttachment(); - if (signalAttachment != null) { - signalAttachment.setClient(false); + var signalAttachmentInGatewayAttachment = gatewayAttachment.getSignalAttachment(); + if (signalAttachmentInGatewayAttachment != null) { + signalAttachmentInGatewayAttachment.setClient(false); } // 网关授权，授权完成直接返回 if (AuthUidToGatewayCheck.getAuthProtocolId() == packet.protocolId()) { @@ -120,14 +120,14 @@ public class Router implements IRouter { gatewaySession.putAttribute(AttributeType.UID, uid); EventBus.asyncSubmit(AuthUidToGatewayEvent.valueOf(gatewaySession.getSid(), uid)); - NetContext.getRouter().send(session, AuthUidToGatewayConfirm.valueOf(uid), new GatewayPacketAttachment(gatewaySession, null)); + NetContext.getRouter().send(session, AuthUidToGatewayConfirm.valueOf(uid), new GatewayAttachment(gatewaySession, null)); return; } - send(gatewaySession, packet, signalAttachment); + send(gatewaySession, packet, signalAttachmentInGatewayAttachment); } else { - logger.error("gateway receives packet:[{}] and packetAttachment:[{}] from server" + + logger.error("gateway receives packet:[{}] and attachment:[{}] from server" + ", but serverSessionMap has no session[id:{}], perhaps client disconnected from gateway." - , JsonUtils.object2String(packet), JsonUtils.object2String(packetAttachment), gatewayPacketAttachment.getSid()); + , JsonUtils.object2String(packet), JsonUtils.object2String(attachment), gatewayAttachment.getSid()); } return; } @@ -138,11 +138,11 @@ public class Router implements IRouter { } // 正常发送消息的接收 - TaskBus.submit(new PacketReceiverTask(session, packet, packetAttachment)); + TaskBus.submit(new PacketReceiverTask(session, packet, attachment)); } @Override - public void send(Session session, IPacket packet, IPacketAttachment packetAttachment) { + public void send(Session session, IPacket packet, IAttachment attachment) { if (session == null) { logger.error("session is null and can not be sent."); return; @@ -152,7 +152,7 @@ public class Router implements IRouter { return; } - var packetInfo = EncodedPacketInfo.valueOf(packet, packetAttachment); + var packetInfo = EncodedPacketInfo.valueOf(packet, attachment); var channel = session.getChannel(); channel.writeAndFlush(packetInfo); @@ -160,34 +160,34 @@ public class Router implements IRouter { @Override public void send(Session session, IPacket packet) { - // 服务器异步返回的消息的发送会有signalPacketAttachment，验证返回的消息是否满足 - var serverSignalPacketAttachment = serverReceiveSignalPacketAttachment.get(); + // 服务器异步返回的消息的发送会有signalAttachment，验证返回的消息是否满足 + var serverSignalAttachment = serverReceiveSignalAttachmentThreadLocal.get(); - if (serverSignalPacketAttachment != null) { - if (serverSignalPacketAttachment.isClient()) { - // 客户端发送的时候不应该有serverSignalPacketAttachment - logger.error("client can not have serverSignalPacketAttachment:[{}] and packet:[{}]", serverSignalPacketAttachment, packet); + if (serverSignalAttachment != null) { + if (serverSignalAttachment.isClient()) { + // 客户端发送的时候不应该有serverSignalAttachment + logger.error("client can not have serverSignalAttachment:[{}] and packet:[{}]", serverSignalAttachment, packet); } else if (Error.errorProtocolId() == packet.protocolId()) { // 错误信息直接返回 } } - send(session, packet, serverSignalPacketAttachment); + send(session, packet, serverSignalAttachment); } @Override public SyncAnswer syncAsk(Session session, IPacket packet, @Nullable Class answerClass, @Nullable Object argument) throws Exception { - var clientAttachment = new SignalPacketAttachment(); + var clientSignalAttachment = new SignalAttachment(); var executorConsistentHash = (argument == null) ? RandomUtils.randomInt() : HashUtils.fnvHash(argument); - clientAttachment.setExecutorConsistentHash(executorConsistentHash); + clientSignalAttachment.setExecutorConsistentHash(executorConsistentHash); try { - PacketSignal.addSignalAttachment(clientAttachment); - send(session, packet, clientAttachment); + SignalBridge.addSignalAttachment(clientSignalAttachment); + send(session, packet, clientSignalAttachment); - IPacket responsePacket = clientAttachment.getResponseFuture().get(DEFAULT_TIMEOUT, TimeUnit.MILLISECONDS); + IPacket responsePacket = clientSignalAttachment.getResponseFuture().get(DEFAULT_TIMEOUT, TimeUnit.MILLISECONDS); if (responsePacket.protocolId() == Error.errorProtocolId()) { throw new ErrorResponseException((Error) responsePacket); @@ -197,30 +197,30 @@ public class Router implements IRouter { , answerClass, responsePacket.getClass().getName())); } - return new SyncAnswer<>((T) responsePacket, clientAttachment); + return new SyncAnswer<>((T) responsePacket, clientSignalAttachment); } catch (TimeoutException e) { throw new NetTimeOutException(StringUtils.format("syncRequest timeout exception, ask:[{}], attachment:[{}]" - , JsonUtils.object2String(packet), JsonUtils.object2String(clientAttachment))); + , JsonUtils.object2String(packet), JsonUtils.object2String(clientSignalAttachment))); } finally { - PacketSignal.removeSignalAttachment(clientAttachment); + SignalBridge.removeSignalAttachment(clientSignalAttachment); } } @Override public AsyncAnswer asyncAsk(Session session, IPacket packet, @Nullable Class answerClass, @Nullable Object argument) { - var clientAttachment = new SignalPacketAttachment(); + var clientSignalAttachment = new SignalAttachment(); var executorConsistentHash = (argument == null) ? RandomUtils.randomInt() : HashUtils.fnvHash(argument); - clientAttachment.setExecutorConsistentHash(executorConsistentHash); + clientSignalAttachment.setExecutorConsistentHash(executorConsistentHash); // 服务器在同步或异步的消息处理中，又调用了同步或异步的方法，这时候threadReceiverAttachment不为空 - var serverSignalPacketAttachment = serverReceiveSignalPacketAttachment.get(); + var serverSignalAttachment = serverReceiveSignalAttachmentThreadLocal.get(); try { var asyncAnswer = new AsyncAnswer(); - asyncAnswer.setFutureAttachment(clientAttachment); + asyncAnswer.setSignalAttachment(clientSignalAttachment); - clientAttachment.getResponseFuture() + clientSignalAttachment.getResponseFuture() .completeOnTimeout(null, DEFAULT_TIMEOUT, TimeUnit.MILLISECONDS) .thenApply(answer -> { if (answer == null) { @@ -238,11 +238,11 @@ public class Router implements IRouter { }) .whenCompleteAsync((answer, throwable) -> { try { - PacketSignal.removeSignalAttachment(clientAttachment); + SignalBridge.removeSignalAttachment(clientSignalAttachment); // 接收者在同步或异步的消息处理中，又调用了异步的方法，这时候threadServerAttachment不为空 - if (serverSignalPacketAttachment != null) { - serverReceiveSignalPacketAttachment.set(serverSignalPacketAttachment); + if (serverSignalAttachment != null) { + serverReceiveSignalAttachmentThreadLocal.set(serverSignalAttachment); } // 如果有异常的话，whenCompleteAsync的下一个thenAccept不会执行 @@ -262,21 +262,21 @@ public class Router implements IRouter { } catch (Throwable throwable1) { logger.error("异步回调方法[ask:{}][answer:{}]错误", packet.getClass().getSimpleName(), answer.getClass().getSimpleName(), throwable1); } finally { - if (serverSignalPacketAttachment != null) { - serverReceiveSignalPacketAttachment.set(null); + if (serverSignalAttachment != null) { + serverReceiveSignalAttachmentThreadLocal.set(null); } } }, TaskBus.executor(executorConsistentHash)); - PacketSignal.addSignalAttachment(clientAttachment); + SignalBridge.addSignalAttachment(clientSignalAttachment); // 等到上层调用whenComplete才会发送消息 - asyncAnswer.setAskCallback(() -> send(session, packet, clientAttachment)); + asyncAnswer.setAskCallback(() -> send(session, packet, clientSignalAttachment)); return asyncAnswer; } catch (Exception e) { - PacketSignal.removeSignalAttachment(clientAttachment); + SignalBridge.removeSignalAttachment(clientSignalAttachment); throw e; } } @@ -289,13 +289,13 @@ public class Router implements IRouter { * 接收者同时只能处理一个session的一个包，同一个发送者发送过来的包排队处理 */ @Override - public void atReceiver(Session session, IPacket packet, IPacketAttachment packetAttachment) { + public void atReceiver(Session session, IPacket packet, IAttachment attachment) { try { // 接收者（服务器）同步和异步消息的接收 - if (packetAttachment != null) { - switch (packetAttachment.packetType()) { + if (attachment != null) { + switch (attachment.packetType()) { case SIGNAL_PACKET: - serverReceiveSignalPacketAttachment.set((SignalPacketAttachment) packetAttachment); + serverReceiveSignalAttachmentThreadLocal.set((SignalAttachment) attachment); break; default: break; @@ -303,17 +303,17 @@ public class Router implements IRouter { } // 调用PacketReceiver - PacketRoute.submit(session, packet, packetAttachment); + PacketRoute.submit(session, packet, attachment); } catch (Exception e) { logger.error(StringUtils.format("e[uid:{}][sid:{}]未知exception异常", session.getAttribute(AttributeType.UID), session.getSid(), e.getMessage()), e); } catch (Throwable t) { logger.error(StringUtils.format("e[uid:{}][sid:{}]未知error错误", session.getAttribute(AttributeType.UID), session.getSid(), t.getMessage()), t); } finally { // 如果有服务器在处理同步或者异步消息的时候由于错误没有返回给客户端消息，则可能会残留serverAttachment，所以先移除 - if (packetAttachment != null) { - switch (packetAttachment.packetType()) { + if (attachment != null) { + switch (attachment.packetType()) { case SIGNAL_PACKET: - serverReceiveSignalPacketAttachment.set(null); + serverReceiveSignalAttachmentThreadLocal.set(null); break; default: break; diff --git a/net/src/main/java/com/zfoo/net/router/answer/AsyncAnswer.java b/net/src/main/java/com/zfoo/net/router/answer/AsyncAnswer.java index 9cdc136a..e39b6ae4 100644 --- a/net/src/main/java/com/zfoo/net/router/answer/AsyncAnswer.java +++ b/net/src/main/java/com/zfoo/net/router/answer/AsyncAnswer.java @@ -13,7 +13,7 @@ package com.zfoo.net.router.answer; -import com.zfoo.net.packet.model.SignalPacketAttachment; +import com.zfoo.net.router.attachment.SignalAttachment; import com.zfoo.net.task.model.SafeRunnable; import com.zfoo.protocol.IPacket; @@ -28,9 +28,9 @@ import java.util.function.Consumer; public class AsyncAnswer implements IAsyncAnswer { private T futurePacket; - private SignalPacketAttachment futureAttachment; - private List> consumerList = new ArrayList<>(2); + private final List> consumerList = new ArrayList<>(2); + private SignalAttachment signalAttachment; private Runnable askCallback; @@ -67,12 +67,12 @@ public class AsyncAnswer implements IAsyncAnswer { this.futurePacket = futurePacket; } - public SignalPacketAttachment getFutureAttachment() { - return futureAttachment; + public SignalAttachment getSignalAttachment() { + return signalAttachment; } - public void setFutureAttachment(SignalPacketAttachment futureAttachment) { - this.futureAttachment = futureAttachment; + public void setSignalAttachment(SignalAttachment signalAttachment) { + this.signalAttachment = signalAttachment; } public Runnable getAskCallback() { diff --git a/net/src/main/java/com/zfoo/net/router/answer/ISyncAnswer.java b/net/src/main/java/com/zfoo/net/router/answer/ISyncAnswer.java index a1709aa1..9ca909d8 100644 --- a/net/src/main/java/com/zfoo/net/router/answer/ISyncAnswer.java +++ b/net/src/main/java/com/zfoo/net/router/answer/ISyncAnswer.java @@ -13,7 +13,7 @@ package com.zfoo.net.router.answer; -import com.zfoo.net.packet.model.SignalPacketAttachment; +import com.zfoo.net.router.attachment.SignalAttachment; import com.zfoo.protocol.IPacket; /** @@ -30,6 +30,6 @@ public interface ISyncAnswer { /** * @return 同步和异步控制的附加包 */ - SignalPacketAttachment attachment(); + SignalAttachment attachment(); } diff --git a/net/src/main/java/com/zfoo/net/router/answer/SyncAnswer.java b/net/src/main/java/com/zfoo/net/router/answer/SyncAnswer.java index 8666ce5c..434c2a29 100644 --- a/net/src/main/java/com/zfoo/net/router/answer/SyncAnswer.java +++ b/net/src/main/java/com/zfoo/net/router/answer/SyncAnswer.java @@ -13,7 +13,7 @@ package com.zfoo.net.router.answer; -import com.zfoo.net.packet.model.SignalPacketAttachment; +import com.zfoo.net.router.attachment.SignalAttachment; import com.zfoo.protocol.IPacket; /** @@ -23,10 +23,10 @@ import com.zfoo.protocol.IPacket; public class SyncAnswer implements ISyncAnswer { - private T packet; - private SignalPacketAttachment attachment; + private final T packet; + private final SignalAttachment attachment; - public SyncAnswer(T packet, SignalPacketAttachment attachment) { + public SyncAnswer(T packet, SignalAttachment attachment) { this.packet = packet; this.attachment = attachment; } @@ -37,7 +37,7 @@ public class SyncAnswer implements ISyncAnswer { } @Override - public SignalPacketAttachment attachment() { + public SignalAttachment attachment() { return attachment; } diff --git a/net/src/main/java/com/zfoo/net/packet/model/PacketAttachmentType.java b/net/src/main/java/com/zfoo/net/router/attachment/AttachmentType.java similarity index 61% rename from net/src/main/java/com/zfoo/net/packet/model/PacketAttachmentType.java rename to net/src/main/java/com/zfoo/net/router/attachment/AttachmentType.java index 193f5328..f045675c 100644 --- a/net/src/main/java/com/zfoo/net/packet/model/PacketAttachmentType.java +++ b/net/src/main/java/com/zfoo/net/router/attachment/AttachmentType.java @@ -1,6 +1,5 @@ /* * 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 * @@ -11,7 +10,7 @@ * See the License for the specific language governing permissions and limitations under the License. */ -package com.zfoo.net.packet.model; +package com.zfoo.net.router.attachment; import java.util.HashMap; import java.util.Map; @@ -20,28 +19,28 @@ import java.util.Map; * @author jaysunxiao * @version 3.0 */ -public enum PacketAttachmentType { +public enum AttachmentType { /** * 带有同步或者异步信息的附加包 */ - SIGNAL_PACKET((byte) 0, SignalPacketAttachment.class), + SIGNAL_PACKET((byte) 0, SignalAttachment.class), /** * 带有网关信息的附加包 */ - GATEWAY_PACKET((byte) 1, GatewayPacketAttachment.class), + GATEWAY_PACKET((byte) 1, GatewayAttachment.class), /** * udp消息的附加包 */ - UDP_PACKET((byte) 2, UdpPacketAttachment.class), + UDP_PACKET((byte) 2, UdpAttachment.class), /** * http消息的附加包 */ - HTTP_PACKET((byte) 3, HttpPacketAttachment.class), + HTTP_PACKET((byte) 3, HttpAttachment.class), /** * 无返回消息的附加包 @@ -52,28 +51,29 @@ public enum PacketAttachmentType { ; - public static final Map map = new HashMap<>(values().length); + public static final Map map = new HashMap<>(values().length); static { - for (var packetType : PacketAttachmentType.values()) { + for (var packetType : AttachmentType.values()) { map.put(packetType.packetType, packetType); } } - public static PacketAttachmentType getPacketType(byte packetType) { - return map.getOrDefault(packetType, PacketAttachmentType.NO_ANSWER_PACKET); - } + private final byte packetType; public byte getPacketType() { return packetType; } - private byte packetType; - private Class clazz; + private final Class clazz; - PacketAttachmentType(byte packetType, Class clazz) { + AttachmentType(byte packetType, Class clazz) { this.packetType = packetType; this.clazz = clazz; } + public static AttachmentType getPacketType(byte packetType) { + return map.getOrDefault(packetType, AttachmentType.NO_ANSWER_PACKET); + } + } diff --git a/net/src/main/java/com/zfoo/net/packet/model/GatewayPacketAttachment.java b/net/src/main/java/com/zfoo/net/router/attachment/GatewayAttachment.java similarity index 80% rename from net/src/main/java/com/zfoo/net/packet/model/GatewayPacketAttachment.java rename to net/src/main/java/com/zfoo/net/router/attachment/GatewayAttachment.java index 421ebfdb..699f9e72 100644 --- a/net/src/main/java/com/zfoo/net/packet/model/GatewayPacketAttachment.java +++ b/net/src/main/java/com/zfoo/net/router/attachment/GatewayAttachment.java @@ -1,6 +1,5 @@ /* * 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 * @@ -11,7 +10,7 @@ * See the License for the specific language governing permissions and limitations under the License. */ -package com.zfoo.net.packet.model; +package com.zfoo.net.router.attachment; import com.zfoo.net.session.model.AttributeType; import com.zfoo.net.session.model.Session; @@ -24,7 +23,7 @@ import org.springframework.lang.Nullable; * @author jaysunxiao * @version 3.0 */ -public class GatewayPacketAttachment implements IPacketAttachment { +public class GatewayAttachment implements IAttachment { public static final transient short PROTOCOL_ID = 1; @@ -56,29 +55,29 @@ public class GatewayPacketAttachment implements IPacketAttachment { /** * 客户端发到网关的可能是一个带有同步或者异步的附加包，网关转发的时候需要把这个附加包给带上 */ - private SignalPacketAttachment signalPacketAttachment; + private SignalAttachment signalAttachment; - public GatewayPacketAttachment() { + public GatewayAttachment() { } - public GatewayPacketAttachment(Session session, @Nullable SignalPacketAttachment signalPacketAttachment) { + public GatewayAttachment(Session session, @Nullable SignalAttachment signalAttachment) { this.client = true; this.sid = session.getSid(); var uid = session.getAttribute(AttributeType.UID); this.uid = uid == null ? 0 : (long) uid; - this.signalPacketAttachment = signalPacketAttachment; + this.signalAttachment = signalAttachment; } - public GatewayPacketAttachment(long sid, long uid) { + public GatewayAttachment(long sid, long uid) { this.sid = sid; this.uid = uid; } @Override - public PacketAttachmentType packetType() { - return PacketAttachmentType.GATEWAY_PACKET; + public AttachmentType packetType() { + return AttachmentType.GATEWAY_PACKET; } @Override @@ -141,11 +140,11 @@ public class GatewayPacketAttachment implements IPacketAttachment { } - public SignalPacketAttachment getSignalPacketAttachment() { - return signalPacketAttachment; + public SignalAttachment getSignalAttachment() { + return signalAttachment; } - public void setSignalPacketAttachment(SignalPacketAttachment signalPacketAttachment) { - this.signalPacketAttachment = signalPacketAttachment; + public void setSignalAttachment(SignalAttachment signalAttachment) { + this.signalAttachment = signalAttachment; } } diff --git a/net/src/main/java/com/zfoo/net/packet/model/HttpPacketAttachment.java b/net/src/main/java/com/zfoo/net/router/attachment/HttpAttachment.java similarity index 88% rename from net/src/main/java/com/zfoo/net/packet/model/HttpPacketAttachment.java rename to net/src/main/java/com/zfoo/net/router/attachment/HttpAttachment.java index d6595ee2..59c026a6 100644 --- a/net/src/main/java/com/zfoo/net/packet/model/HttpPacketAttachment.java +++ b/net/src/main/java/com/zfoo/net/router/attachment/HttpAttachment.java @@ -1,6 +1,5 @@ /* * 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 * @@ -11,7 +10,7 @@ * See the License for the specific language governing permissions and limitations under the License. */ -package com.zfoo.net.packet.model; +package com.zfoo.net.router.attachment; import com.zfoo.util.math.HashUtils; import io.netty.handler.codec.http.FullHttpRequest; @@ -21,7 +20,7 @@ import io.netty.handler.codec.http.HttpResponseStatus; * @author jaysunxiao * @version 3.0 */ -public class HttpPacketAttachment implements IPacketAttachment { +public class HttpAttachment implements IAttachment { public static final transient short PROTOCOL_ID = 3; @@ -35,16 +34,16 @@ public class HttpPacketAttachment implements IPacketAttachment { private transient HttpResponseStatus httpResponseStatus; - public static HttpPacketAttachment valueOf(FullHttpRequest fullHttpRequest, HttpResponseStatus httpResponseStatus) { - var attachment = new HttpPacketAttachment(); + public static HttpAttachment valueOf(FullHttpRequest fullHttpRequest, HttpResponseStatus httpResponseStatus) { + var attachment = new HttpAttachment(); attachment.fullHttpRequest = fullHttpRequest; attachment.httpResponseStatus = httpResponseStatus; return attachment; } @Override - public PacketAttachmentType packetType() { - return PacketAttachmentType.HTTP_PACKET; + public AttachmentType packetType() { + return AttachmentType.HTTP_PACKET; } @Override diff --git a/net/src/main/java/com/zfoo/net/packet/model/IPacketAttachment.java b/net/src/main/java/com/zfoo/net/router/attachment/IAttachment.java similarity index 86% rename from net/src/main/java/com/zfoo/net/packet/model/IPacketAttachment.java rename to net/src/main/java/com/zfoo/net/router/attachment/IAttachment.java index 85990bb2..0cb56e29 100644 --- a/net/src/main/java/com/zfoo/net/packet/model/IPacketAttachment.java +++ b/net/src/main/java/com/zfoo/net/router/attachment/IAttachment.java @@ -1,6 +1,5 @@ /* * 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 * @@ -11,7 +10,7 @@ * See the License for the specific language governing permissions and limitations under the License. */ -package com.zfoo.net.packet.model; +package com.zfoo.net.router.attachment; import com.zfoo.protocol.IPacket; @@ -19,9 +18,9 @@ import com.zfoo.protocol.IPacket; * @author jaysunxiao * @version 3.0 */ -public interface IPacketAttachment extends IPacket { +public interface IAttachment extends IPacket { - PacketAttachmentType packetType(); + AttachmentType packetType(); /** * 用来确定这条消息在哪一个线程处理 diff --git a/net/src/main/java/com/zfoo/net/packet/model/NoAnswerAttachment.java b/net/src/main/java/com/zfoo/net/router/attachment/NoAnswerAttachment.java similarity index 88% rename from net/src/main/java/com/zfoo/net/packet/model/NoAnswerAttachment.java rename to net/src/main/java/com/zfoo/net/router/attachment/NoAnswerAttachment.java index f7df7884..458e2d63 100644 --- a/net/src/main/java/com/zfoo/net/packet/model/NoAnswerAttachment.java +++ b/net/src/main/java/com/zfoo/net/router/attachment/NoAnswerAttachment.java @@ -1,6 +1,5 @@ /* * 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 * @@ -11,7 +10,7 @@ * See the License for the specific language governing permissions and limitations under the License. */ -package com.zfoo.net.packet.model; +package com.zfoo.net.router.attachment; /** * 附加包对业务层透明，禁止在业务层使用 @@ -19,7 +18,7 @@ package com.zfoo.net.packet.model; * @author jaysunxiao * @version 3.0 */ -public class NoAnswerAttachment implements IPacketAttachment { +public class NoAnswerAttachment implements IAttachment { public static final transient short PROTOCOL_ID = 4; @@ -35,8 +34,8 @@ public class NoAnswerAttachment implements IPacketAttachment { } @Override - public PacketAttachmentType packetType() { - return PacketAttachmentType.NO_ANSWER_PACKET; + public AttachmentType packetType() { + return AttachmentType.NO_ANSWER_PACKET; } @Override diff --git a/net/src/main/java/com/zfoo/net/packet/model/SignalPacketAttachment.java b/net/src/main/java/com/zfoo/net/router/attachment/SignalAttachment.java similarity index 77% rename from net/src/main/java/com/zfoo/net/packet/model/SignalPacketAttachment.java rename to net/src/main/java/com/zfoo/net/router/attachment/SignalAttachment.java index d45cdaf2..eaef572f 100644 --- a/net/src/main/java/com/zfoo/net/packet/model/SignalPacketAttachment.java +++ b/net/src/main/java/com/zfoo/net/router/attachment/SignalAttachment.java @@ -1,6 +1,5 @@ /* * 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 * @@ -11,7 +10,7 @@ * See the License for the specific language governing permissions and limitations under the License. */ -package com.zfoo.net.packet.model; +package com.zfoo.net.router.attachment; import com.zfoo.protocol.IPacket; import com.zfoo.scheduler.util.TimeUtils; @@ -25,16 +24,16 @@ import java.util.concurrent.atomic.AtomicInteger; * @author jaysunxiao * @version 3.0 */ -public class SignalPacketAttachment implements IPacketAttachment { +public class SignalAttachment implements IAttachment { public static final transient short PROTOCOL_ID = 0; - private static final AtomicInteger ATOMIC_PACKET_ID = new AtomicInteger(0); + private static final AtomicInteger ATOMIC_ID = new AtomicInteger(0); /** - * 唯一标识一个packet， 唯一表示一个PacketAttachment，hashcode() and equals() 也通过packetId计算 + * 唯一标识一个packet， 唯一表示一个Attachment，hashcode() and equals() 也通过signalId计算 */ - private int packetId = ATOMIC_PACKET_ID.incrementAndGet(); + private int signalId = ATOMIC_ID.incrementAndGet(); /** * 用来在TaskBus中计算一致性hash的参数 @@ -56,13 +55,13 @@ public class SignalPacketAttachment implements IPacketAttachment { */ private transient CompletableFuture responseFuture = new CompletableFuture<>(); - public SignalPacketAttachment() { + public SignalAttachment() { } @Override - public PacketAttachmentType packetType() { - return PacketAttachmentType.SIGNAL_PACKET; + public AttachmentType packetType() { + return AttachmentType.SIGNAL_PACKET; } @Override @@ -92,21 +91,21 @@ public class SignalPacketAttachment implements IPacketAttachment { if (o == null || getClass() != o.getClass()) { return false; } - SignalPacketAttachment that = (SignalPacketAttachment) o; - return packetId == that.packetId; + SignalAttachment that = (SignalAttachment) o; + return signalId == that.signalId; } @Override public int hashCode() { - return packetId; + return signalId; } - public int getPacketId() { - return packetId; + public int getSignalId() { + return signalId; } - public void setPacketId(int packetId) { - this.packetId = packetId; + public void setSignalId(int signalId) { + this.signalId = signalId; } public int getExecutorConsistentHash() { diff --git a/net/src/main/java/com/zfoo/net/packet/model/UdpPacketAttachment.java b/net/src/main/java/com/zfoo/net/router/attachment/UdpAttachment.java similarity index 80% rename from net/src/main/java/com/zfoo/net/packet/model/UdpPacketAttachment.java rename to net/src/main/java/com/zfoo/net/router/attachment/UdpAttachment.java index 9ccecd40..c3a2d123 100644 --- a/net/src/main/java/com/zfoo/net/packet/model/UdpPacketAttachment.java +++ b/net/src/main/java/com/zfoo/net/router/attachment/UdpAttachment.java @@ -1,6 +1,5 @@ /* * 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 * @@ -11,7 +10,7 @@ * See the License for the specific language governing permissions and limitations under the License. */ -package com.zfoo.net.packet.model; +package com.zfoo.net.router.attachment; import com.zfoo.util.math.RandomUtils; @@ -19,23 +18,23 @@ import com.zfoo.util.math.RandomUtils; * @author jaysunxiao * @version 3.0 */ -public class UdpPacketAttachment implements IPacketAttachment { +public class UdpAttachment implements IAttachment { public static final transient short PROTOCOL_ID = 2; private String host; private int port; - public static UdpPacketAttachment valueOf(String host, int port) { - var attachment = new UdpPacketAttachment(); + public static UdpAttachment valueOf(String host, int port) { + var attachment = new UdpAttachment(); attachment.host = host; attachment.port = port; return attachment; } @Override - public PacketAttachmentType packetType() { - return PacketAttachmentType.UDP_PACKET; + public AttachmentType packetType() { + return AttachmentType.UDP_PACKET; } @Override diff --git a/net/src/main/java/com/zfoo/net/router/receiver/EnhanceUtils.java b/net/src/main/java/com/zfoo/net/router/receiver/EnhanceUtils.java index db4d0379..5c65f6bb 100644 --- a/net/src/main/java/com/zfoo/net/router/receiver/EnhanceUtils.java +++ b/net/src/main/java/com/zfoo/net/router/receiver/EnhanceUtils.java @@ -13,7 +13,7 @@ package com.zfoo.net.router.receiver; -import com.zfoo.net.packet.model.IPacketAttachment; +import com.zfoo.net.router.attachment.IAttachment; import com.zfoo.net.session.model.Session; import com.zfoo.protocol.IPacket; import com.zfoo.protocol.util.StringUtils; @@ -34,7 +34,7 @@ public abstract class EnhanceUtils { static { var classArray = new Class[]{ IPacket.class, - IPacketAttachment.class, + IAttachment.class, IPacketReceiver.class, Session.class }; @@ -73,7 +73,7 @@ public abstract class EnhanceUtils { enhanceClazz.addConstructor(constructor); // 定义类实现的接口方法 - CtMethod invokeMethod = new CtMethod(classPool.get(void.class.getCanonicalName()), "invoke", classPool.get(new String[]{Session.class.getCanonicalName(), IPacket.class.getCanonicalName(), IPacketAttachment.class.getCanonicalName()}), enhanceClazz); + CtMethod invokeMethod = new CtMethod(classPool.get(void.class.getCanonicalName()), "invoke", classPool.get(new String[]{Session.class.getCanonicalName(), IPacket.class.getCanonicalName(), IAttachment.class.getCanonicalName()}), enhanceClazz); invokeMethod.setModifiers(Modifier.PUBLIC + Modifier.FINAL); if (attachmentClazz == null) { // 强制类型转换 diff --git a/net/src/main/java/com/zfoo/net/router/receiver/IPacketReceiver.java b/net/src/main/java/com/zfoo/net/router/receiver/IPacketReceiver.java index 0ea90354..80384d6b 100644 --- a/net/src/main/java/com/zfoo/net/router/receiver/IPacketReceiver.java +++ b/net/src/main/java/com/zfoo/net/router/receiver/IPacketReceiver.java @@ -13,7 +13,7 @@ package com.zfoo.net.router.receiver; -import com.zfoo.net.packet.model.IPacketAttachment; +import com.zfoo.net.router.attachment.IAttachment; import com.zfoo.net.session.model.Session; import com.zfoo.protocol.IPacket; @@ -23,6 +23,6 @@ import com.zfoo.protocol.IPacket; */ public interface IPacketReceiver { - void invoke(Session session, IPacket packet, IPacketAttachment attachment); + void invoke(Session session, IPacket packet, IAttachment attachment); } diff --git a/net/src/main/java/com/zfoo/net/router/receiver/PacketReceiverDefinition.java b/net/src/main/java/com/zfoo/net/router/receiver/PacketReceiverDefinition.java index 98dabd53..cc12c2a5 100644 --- a/net/src/main/java/com/zfoo/net/router/receiver/PacketReceiverDefinition.java +++ b/net/src/main/java/com/zfoo/net/router/receiver/PacketReceiverDefinition.java @@ -13,7 +13,7 @@ package com.zfoo.net.router.receiver; -import com.zfoo.net.packet.model.IPacketAttachment; +import com.zfoo.net.router.attachment.IAttachment; import com.zfoo.net.session.model.Session; import com.zfoo.protocol.IPacket; import com.zfoo.protocol.util.ReflectionUtils; @@ -44,7 +44,7 @@ public class PacketReceiverDefinition implements IPacketReceiver { private Class packetClazz; /** - * 接收的包的附加包的Class类，如GatewayPacketAttachment + * 接收的包的附加包的Class类，如GatewayAttachment */ private Class attachmentClazz; @@ -57,7 +57,7 @@ public class PacketReceiverDefinition implements IPacketReceiver { } @Override - public void invoke(Session session, IPacket packet, IPacketAttachment attachment) { + public void invoke(Session session, IPacket packet, IAttachment attachment) { if (attachmentClazz == null) { ReflectionUtils.invokeMethod(bean, method, session, packet); } else { diff --git a/net/src/main/java/com/zfoo/net/router/route/PacketRoute.java b/net/src/main/java/com/zfoo/net/router/route/PacketRoute.java index 2be5a87d..1f272a65 100644 --- a/net/src/main/java/com/zfoo/net/router/route/PacketRoute.java +++ b/net/src/main/java/com/zfoo/net/router/route/PacketRoute.java @@ -14,11 +14,11 @@ package com.zfoo.net.router.route; import com.zfoo.event.model.event.IEvent; -import com.zfoo.net.packet.model.GatewayPacketAttachment; -import com.zfoo.net.packet.model.HttpPacketAttachment; -import com.zfoo.net.packet.model.IPacketAttachment; -import com.zfoo.net.packet.model.UdpPacketAttachment; import com.zfoo.net.packet.service.PacketService; +import com.zfoo.net.router.attachment.GatewayAttachment; +import com.zfoo.net.router.attachment.HttpAttachment; +import com.zfoo.net.router.attachment.IAttachment; +import com.zfoo.net.router.attachment.UdpAttachment; import com.zfoo.net.router.receiver.EnhanceUtils; import com.zfoo.net.router.receiver.IPacketReceiver; import com.zfoo.net.router.receiver.PacketReceiver; @@ -38,6 +38,8 @@ import org.slf4j.LoggerFactory; import java.lang.reflect.Modifier; /** + * 包的接收路线，服务器收到packet调用对应的Receiver + * * @author jaysunxiao * @version 3.0 */ @@ -57,14 +59,14 @@ public abstract class PacketRoute { * 发送者同时能发送多个包 * 接收者同时只能处理一个session的一个包，同一个发送者发送过来的包排队处理 */ - public static void submit(Session session, IPacket packet, IPacketAttachment packetAttachment) { + public static void submit(Session session, IPacket packet, IAttachment attachment) { var packetReceiver = (IPacketReceiver) packetReceiverList[packet.protocolId()].receiver(); if (packetReceiver == null) { throw new RuntimeException(StringUtils.format("no any packetReceiverDefinition found for this [packet:{}]", packet.getClass().getName())); } // 调用PacketReceiver - packetReceiver.invoke(session, packet, packetAttachment); + packetReceiver.invoke(session, packet, attachment); } @@ -92,8 +94,8 @@ public abstract class PacketRoute { AssertionUtils.isTrue(IPacket.class.isAssignableFrom(paramClazzs[1]) , "[class:{}] [method:{}],the second parameter must be IPacket type parameter Exception.", bean.getClass().getName(), method.getName()); - AssertionUtils.isTrue(paramClazzs.length != 3 || IPacketAttachment.class.isAssignableFrom(paramClazzs[2]) - , "[class:{}] [method:{}],the third parameter must be IPacketAttachment type parameter Exception.", bean.getClass().getName(), method.getName()); + AssertionUtils.isTrue(paramClazzs.length != 3 || IAttachment.class.isAssignableFrom(paramClazzs[2]) + , "[class:{}] [method:{}],the third parameter must be IAttachment type parameter Exception.", bean.getClass().getName(), method.getName()); var packetClazz = (Class) paramClazzs[1]; var attachmentClazz = paramClazzs.length == 3 ? paramClazzs[2] : null; @@ -114,11 +116,11 @@ public abstract class PacketRoute { // 如果以Ask结尾的请求，那么attachment不能为GatewayAttachment if (attachmentClazz != null) { if (packetName.endsWith(PacketService.NET_REQUEST_SUFFIX)) { - AssertionUtils.isTrue(attachmentClazz.equals(GatewayPacketAttachment.class) || attachmentClazz.equals(UdpPacketAttachment.class) || attachmentClazz.equals(HttpPacketAttachment.class) - , "[class:{}] [method:{}] [packet:{}] must use [attachment:{}]!", bean.getClass().getName(), methodName, packetName, GatewayPacketAttachment.class.getCanonicalName()); + AssertionUtils.isTrue(attachmentClazz.equals(GatewayAttachment.class) || attachmentClazz.equals(UdpAttachment.class) || attachmentClazz.equals(HttpAttachment.class) + , "[class:{}] [method:{}] [packet:{}] must use [attachment:{}]!", bean.getClass().getName(), methodName, packetName, GatewayAttachment.class.getCanonicalName()); } else if (packetName.endsWith(PacketService.NET_ASK_SUFFIX)) { - AssertionUtils.isTrue(!attachmentClazz.equals(GatewayPacketAttachment.class) - , "[class:{}] [method:{}] [packet:{}] can not match with [attachment:{}]!", bean.getClass().getName(), methodName, packetName, GatewayPacketAttachment.class.getCanonicalName()); + AssertionUtils.isTrue(!attachmentClazz.equals(GatewayAttachment.class) + , "[class:{}] [method:{}] [packet:{}] can not match with [attachment:{}]!", bean.getClass().getName(), methodName, packetName, GatewayAttachment.class.getCanonicalName()); } } diff --git a/net/src/main/java/com/zfoo/net/router/route/PacketSignal.java b/net/src/main/java/com/zfoo/net/router/route/SignalBridge.java similarity index 54% rename from net/src/main/java/com/zfoo/net/router/route/PacketSignal.java rename to net/src/main/java/com/zfoo/net/router/route/SignalBridge.java index 5daf6e87..35ea0ff6 100644 --- a/net/src/main/java/com/zfoo/net/router/route/PacketSignal.java +++ b/net/src/main/java/com/zfoo/net/router/route/SignalBridge.java @@ -12,7 +12,7 @@ package com.zfoo.net.router.route; -import com.zfoo.net.packet.model.SignalPacketAttachment; +import com.zfoo.net.router.attachment.SignalAttachment; import com.zfoo.protocol.util.JsonUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -22,61 +22,61 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicReferenceArray; /** - * 同步或异步的调用控制器 + * 同步或异步的调用控制器，同步和异步调用的信号沟通桥梁 * * @author jaysunxiao * @version 3.0 */ -public class PacketSignal { +public class SignalBridge { - private static final Logger logger = LoggerFactory.getLogger(PacketSignal.class); + private static final Logger logger = LoggerFactory.getLogger(SignalBridge.class); // equal with 32767 private static final int SIGNAL_MASK = 0B00000000_00000000_01111111_11111111; - private static AtomicReferenceArray signalPacketArray = new AtomicReferenceArray<>(SIGNAL_MASK + 1); + private static final AtomicReferenceArray signalAttachmentArray = new AtomicReferenceArray<>(SIGNAL_MASK + 1); /** * Session控制同步或异步的附加包，key：packetId */ - private static Map signalPacketAttachmentMap = new ConcurrentHashMap<>(1000); + private static final Map signalAttachmentMap = new ConcurrentHashMap<>(1000); - public static void addSignalAttachment(SignalPacketAttachment packetAttachment) { - var packetId = packetAttachment.getPacketId(); + public static void addSignalAttachment(SignalAttachment signalAttachment) { + var packetId = signalAttachment.getSignalId(); var hash = packetId & SIGNAL_MASK; - if (signalPacketArray.compareAndSet(hash, null, packetAttachment)) { + if (signalAttachmentArray.compareAndSet(hash, null, signalAttachment)) { return; } - signalPacketAttachmentMap.put(packetId, packetAttachment); + signalAttachmentMap.put(packetId, signalAttachment); } - public static SignalPacketAttachment removeSignalAttachment(SignalPacketAttachment packetAttachment) { - return removeSignalAttachment(packetAttachment.getPacketId()); + public static SignalAttachment removeSignalAttachment(SignalAttachment signalAttachment) { + return removeSignalAttachment(signalAttachment.getSignalId()); } - public static SignalPacketAttachment removeSignalAttachment(int packetId) { + public static SignalAttachment removeSignalAttachment(int packetId) { var hash = packetId & SIGNAL_MASK; - var attachment = signalPacketArray.get(hash); - if (attachment != null && attachment.getPacketId() == packetId && signalPacketArray.compareAndSet(hash, attachment, null)) { + var attachment = signalAttachmentArray.get(hash); + if (attachment != null && attachment.getSignalId() == packetId && signalAttachmentArray.compareAndSet(hash, attachment, null)) { return attachment; } - return signalPacketAttachmentMap.remove(packetId); + return signalAttachmentMap.remove(packetId); } public static void status() { var count = 0; for (int i = 0; i < SIGNAL_MASK + 1; i++) { - var value = signalPacketArray.get(i); + var value = signalAttachmentArray.get(i); if (value != null) { logger.info("signalPacketArray has attachment [index:{}][count:{}][value:{}]", i, ++count, JsonUtils.object2String(value)); } } - signalPacketAttachmentMap.forEach((key, value) -> { - logger.info("signalPacketAttachmentMap has attachment [key:{}][value:{}]", key, JsonUtils.object2String(value)); + signalAttachmentMap.forEach((key, value) -> { + logger.info("signalAttachmentMap has attachment [key:{}][value:{}]", key, JsonUtils.object2String(value)); }); } diff --git a/net/src/main/java/com/zfoo/net/task/dispatcher/ConsistentHashTaskDispatch.java b/net/src/main/java/com/zfoo/net/task/dispatcher/ConsistentHashTaskDispatch.java index c5d4515a..d350ced8 100644 --- a/net/src/main/java/com/zfoo/net/task/dispatcher/ConsistentHashTaskDispatch.java +++ b/net/src/main/java/com/zfoo/net/task/dispatcher/ConsistentHashTaskDispatch.java @@ -24,7 +24,7 @@ import java.util.concurrent.ExecutorService; */ public class ConsistentHashTaskDispatch extends AbstractTaskDispatch { - private static ConsistentHashTaskDispatch INSTANCE = new ConsistentHashTaskDispatch(); + private static final ConsistentHashTaskDispatch INSTANCE = new ConsistentHashTaskDispatch(); public static ConsistentHashTaskDispatch getINSTANCE() { return INSTANCE; @@ -32,13 +32,13 @@ public class ConsistentHashTaskDispatch extends AbstractTaskDispatch { @Override public ExecutorService getExecutor(PacketReceiverTask packetReceiverTask) { - var packetAttachment = packetReceiverTask.getPacketAttachment(); + var attachment = packetReceiverTask.getAttachment(); - if (packetAttachment == null) { + if (attachment == null) { return SessionIdTaskDispatch.getInstance().getExecutor(packetReceiverTask); } - return TaskBus.executor(packetAttachment.executorConsistentHash()); + return TaskBus.executor(attachment.executorConsistentHash()); } } diff --git a/net/src/main/java/com/zfoo/net/task/model/PacketReceiverTask.java b/net/src/main/java/com/zfoo/net/task/model/PacketReceiverTask.java index 5a513156..41faefcd 100644 --- a/net/src/main/java/com/zfoo/net/task/model/PacketReceiverTask.java +++ b/net/src/main/java/com/zfoo/net/task/model/PacketReceiverTask.java @@ -14,7 +14,7 @@ package com.zfoo.net.task.model; import com.zfoo.net.NetContext; -import com.zfoo.net.packet.model.IPacketAttachment; +import com.zfoo.net.router.attachment.IAttachment; import com.zfoo.net.session.model.Session; import com.zfoo.protocol.IPacket; @@ -26,17 +26,17 @@ public final class PacketReceiverTask implements Runnable { private Session session; private IPacket packet; - private IPacketAttachment packetAttachment; + private IAttachment attachment; - public PacketReceiverTask(Session session, IPacket packet, IPacketAttachment packetAttachment) { + public PacketReceiverTask(Session session, IPacket packet, IAttachment attachment) { this.session = session; this.packet = packet; - this.packetAttachment = packetAttachment; + this.attachment = attachment; } @Override public void run() { - NetContext.getRouter().atReceiver(session, packet, packetAttachment); + NetContext.getRouter().atReceiver(session, packet, attachment); } public Session getSession() { @@ -55,11 +55,11 @@ public final class PacketReceiverTask implements Runnable { this.packet = packet; } - public IPacketAttachment getPacketAttachment() { - return packetAttachment; + public IAttachment getAttachment() { + return attachment; } - public void setPacketAttachment(IPacketAttachment packetAttachment) { - this.packetAttachment = packetAttachment; + public void setAttachment(IAttachment attachment) { + this.attachment = attachment; } } diff --git a/net/src/test/java/com/zfoo/net/core/gateway/GatewayProviderController.java b/net/src/test/java/com/zfoo/net/core/gateway/GatewayProviderController.java index 2d66b75d..86e5f910 100644 --- a/net/src/test/java/com/zfoo/net/core/gateway/GatewayProviderController.java +++ b/net/src/test/java/com/zfoo/net/core/gateway/GatewayProviderController.java @@ -15,7 +15,7 @@ package com.zfoo.net.core.gateway; import com.zfoo.net.NetContext; import com.zfoo.net.packet.gateway.GatewayToProviderRequest; import com.zfoo.net.packet.gateway.GatewayToProviderResponse; -import com.zfoo.net.packet.model.GatewayPacketAttachment; +import com.zfoo.net.router.attachment.GatewayAttachment; import com.zfoo.net.router.receiver.PacketReceiver; import com.zfoo.net.session.model.Session; import com.zfoo.protocol.util.JsonUtils; @@ -34,7 +34,7 @@ public class GatewayProviderController { private static final Logger logger = LoggerFactory.getLogger(GatewayProviderController.class); @PacketReceiver - public void atGatewayToProviderRequest(Session session, GatewayToProviderRequest request, GatewayPacketAttachment gatewayAttachment) { + public void atGatewayToProviderRequest(Session session, GatewayToProviderRequest request, GatewayAttachment gatewayAttachment) { logger.info("provider receive [packet:{}] from client", JsonUtils.object2String(request)); var response = new GatewayToProviderResponse(); diff --git a/net/src/test/java/com/zfoo/net/core/udp/client/UdpClientPacketController.java b/net/src/test/java/com/zfoo/net/core/udp/client/UdpClientPacketController.java index 0180bd44..a19092b8 100644 --- a/net/src/test/java/com/zfoo/net/core/udp/client/UdpClientPacketController.java +++ b/net/src/test/java/com/zfoo/net/core/udp/client/UdpClientPacketController.java @@ -12,8 +12,8 @@ package com.zfoo.net.core.udp.client; -import com.zfoo.net.packet.model.UdpPacketAttachment; import com.zfoo.net.packet.udp.UdpHelloResponse; +import com.zfoo.net.router.attachment.UdpAttachment; import com.zfoo.net.router.receiver.PacketReceiver; import com.zfoo.net.session.model.Session; import com.zfoo.protocol.util.JsonUtils; @@ -31,7 +31,7 @@ public class UdpClientPacketController { private static final Logger logger = LoggerFactory.getLogger(UdpClientPacketController.class); @PacketReceiver - public void atUdpHelloResponse(Session session, UdpHelloResponse response, UdpPacketAttachment attachment) { + public void atUdpHelloResponse(Session session, UdpHelloResponse response, UdpAttachment attachment) { logger.info("udp client receive [packet:{}] from server", JsonUtils.object2String(response)); } diff --git a/net/src/test/java/com/zfoo/net/core/udp/client/UdpClientTest.java b/net/src/test/java/com/zfoo/net/core/udp/client/UdpClientTest.java index 6336abb1..ac6ac388 100644 --- a/net/src/test/java/com/zfoo/net/core/udp/client/UdpClientTest.java +++ b/net/src/test/java/com/zfoo/net/core/udp/client/UdpClientTest.java @@ -14,8 +14,8 @@ package com.zfoo.net.core.udp.client; import com.zfoo.net.NetContext; import com.zfoo.net.core.udp.UdpClient; -import com.zfoo.net.packet.model.UdpPacketAttachment; import com.zfoo.net.packet.udp.UdpHelloRequest; +import com.zfoo.net.router.attachment.UdpAttachment; import com.zfoo.util.ThreadUtils; import com.zfoo.util.net.HostAndPort; import org.junit.Ignore; @@ -40,7 +40,7 @@ public class UdpClientTest { var request = new UdpHelloRequest(); request.setMessage("Hello, this is the udp client!"); - NetContext.getRouter().send(session, request, UdpPacketAttachment.valueOf(hostAndPort.getHost(), hostAndPort.getPort())); + NetContext.getRouter().send(session, request, UdpAttachment.valueOf(hostAndPort.getHost(), hostAndPort.getPort())); ThreadUtils.sleep(Long.MAX_VALUE); } diff --git a/net/src/test/java/com/zfoo/net/core/udp/server/UdpServerPacketController.java b/net/src/test/java/com/zfoo/net/core/udp/server/UdpServerPacketController.java index 2981fd7e..5d7d02ee 100644 --- a/net/src/test/java/com/zfoo/net/core/udp/server/UdpServerPacketController.java +++ b/net/src/test/java/com/zfoo/net/core/udp/server/UdpServerPacketController.java @@ -13,9 +13,9 @@ package com.zfoo.net.core.udp.server; import com.zfoo.net.NetContext; -import com.zfoo.net.packet.model.UdpPacketAttachment; import com.zfoo.net.packet.udp.UdpHelloRequest; import com.zfoo.net.packet.udp.UdpHelloResponse; +import com.zfoo.net.router.attachment.UdpAttachment; import com.zfoo.net.router.receiver.PacketReceiver; import com.zfoo.net.session.model.Session; import com.zfoo.protocol.util.JsonUtils; @@ -33,7 +33,7 @@ public class UdpServerPacketController { private static final Logger logger = LoggerFactory.getLogger(UdpServerPacketController.class); @PacketReceiver - public void atUdpHelloRequest(Session session, UdpHelloRequest request, UdpPacketAttachment attachment) { + public void atUdpHelloRequest(Session session, UdpHelloRequest request, UdpAttachment attachment) { logger.info("udp server receive [packet:{}] from client", JsonUtils.object2String(request)); var response = new UdpHelloResponse(); diff --git a/net/src/test/java/com/zfoo/net/protocol/ProtocolTest.java b/net/src/test/java/com/zfoo/net/protocol/ProtocolTest.java index 27ea6f32..d4936c4e 100644 --- a/net/src/test/java/com/zfoo/net/protocol/ProtocolTest.java +++ b/net/src/test/java/com/zfoo/net/protocol/ProtocolTest.java @@ -16,9 +16,9 @@ package com.zfoo.net.protocol; import com.zfoo.net.NetContext; import com.zfoo.net.packet.*; import com.zfoo.net.packet.model.DecodedPacketInfo; -import com.zfoo.net.packet.model.SignalPacketAttachment; import com.zfoo.net.packet.service.IPacketService; import com.zfoo.net.packet.service.PacketService; +import com.zfoo.net.router.attachment.SignalAttachment; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import org.junit.Assert; @@ -39,11 +39,15 @@ public class ProtocolTest { private static final ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("config.xml"); private static final IPacketService packetService = NetContext.getPacketService(); - private static SignalPacketAttachment attachment = new SignalPacketAttachment(); + private static final SignalAttachment attachment = new SignalAttachment(); - private static ObjectA objectA0 = new ObjectA(); - private static ObjectA objectA1 = new ObjectA(); - private static ObjectB objectB = new ObjectB(); + private static final ObjectA objectA0 = new ObjectA(); + private static final ObjectA objectA1 = new ObjectA(); + private static final ObjectB objectB = new ObjectB(); + private static final List list = List.of(Integer.MIN_VALUE, -99, 0, 99, Integer.MAX_VALUE); + private static final Map mapWithIntegerAndString = Map.of(Integer.MIN_VALUE, "min", -99, "-99", 0, "0", 99, "99", Integer.MAX_VALUE, "max"); + private static final Map mapWithObject = Map.of(Integer.MIN_VALUE, objectA0, -99, objectA0, 0, objectA0, 99, objectA0, Integer.MAX_VALUE, objectA0); + private static final List> listWithMapWithObject = List.of(mapWithObject, mapWithObject, mapWithObject); static { objectA0.setA(Integer.MAX_VALUE); @@ -52,16 +56,9 @@ public class ProtocolTest { objectA1.setObjectB(objectB); objectB.setFlag(false); - attachment.setPacketId(Integer.MAX_VALUE); + attachment.setSignalId(Integer.MAX_VALUE); } - private static List list = List.of(Integer.MIN_VALUE, -99, 0, 99, Integer.MAX_VALUE); - - private static Map mapWithIntegerAndString = Map.of(Integer.MIN_VALUE, "min", -99, "-99", 0, "0", 99, "99", Integer.MAX_VALUE, "max"); - private static Map mapWithObject = Map.of(Integer.MIN_VALUE, objectA0, -99, objectA0, 0, objectA0, 99, objectA0, Integer.MAX_VALUE, objectA0); - private static List> listWithMapWithObject = List.of(mapWithObject, mapWithObject, mapWithObject); - - @Test public void testCMInt() { CM_Int cm = new CM_Int(); @@ -81,7 +78,7 @@ public class ProtocolTest { DecodedPacketInfo packetInfo = packetService.read(writeBuff); Assert.assertEquals(packetInfo.getPacket(), cm); - Assert.assertEquals(packetInfo.getPacketAttachment(), attachment); + Assert.assertEquals(packetInfo.getAttachment(), attachment); } @Test @@ -98,7 +95,7 @@ public class ProtocolTest { DecodedPacketInfo packetInfo = packetService.read(writeBuff); Assert.assertEquals(packetInfo.getPacket(), cm); - Assert.assertNull(packetInfo.getPacketAttachment()); + Assert.assertNull(packetInfo.getAttachment()); } @Test @@ -117,7 +114,7 @@ public class ProtocolTest { DecodedPacketInfo packetInfo = packetService.read(writeBuff); Assert.assertEquals(packetInfo.getPacket(), cm); - Assert.assertNull(packetInfo.getPacketAttachment()); + Assert.assertNull(packetInfo.getAttachment()); } @Test @@ -136,7 +133,7 @@ public class ProtocolTest { DecodedPacketInfo packetInfo = packetService.read(writeBuff); Assert.assertEquals(packetInfo.getPacket(), cm); - Assert.assertNull(packetInfo.getPacketAttachment()); + Assert.assertNull(packetInfo.getAttachment()); } @Test @@ -171,7 +168,7 @@ public class ProtocolTest { DecodedPacketInfo packetInfo = packetService.read(writeBuff); Assert.assertEquals(packetInfo.getPacket(), cm); - Assert.assertNull(packetInfo.getPacketAttachment()); + Assert.assertNull(packetInfo.getAttachment()); } @@ -201,7 +198,7 @@ public class ProtocolTest { DecodedPacketInfo packetInfo = packetService.read(writeBuff); Assert.assertEquals(packetInfo.getPacket(), cm); - Assert.assertNull(packetInfo.getPacketAttachment()); + Assert.assertNull(packetInfo.getAttachment()); } @Test @@ -220,7 +217,7 @@ public class ProtocolTest { DecodedPacketInfo packetInfo = packetService.read(writeBuff); Assert.assertEquals(packetInfo.getPacket(), cm); - Assert.assertNull(packetInfo.getPacketAttachment()); + Assert.assertNull(packetInfo.getAttachment()); } @Test @@ -248,7 +245,7 @@ public class ProtocolTest { DecodedPacketInfo packetInfo = packetService.read(writeBuff); Assert.assertEquals(packetInfo.getPacket(), cm); - Assert.assertNull(packetInfo.getPacketAttachment()); + Assert.assertNull(packetInfo.getAttachment()); } } diff --git a/net/src/test/java/com/zfoo/net/router/PacketSignalArray.java b/net/src/test/java/com/zfoo/net/router/PacketSignalArray.java index 23b13680..4ba9202e 100644 --- a/net/src/test/java/com/zfoo/net/router/PacketSignalArray.java +++ b/net/src/test/java/com/zfoo/net/router/PacketSignalArray.java @@ -33,12 +33,12 @@ public class PacketSignalArray { // equal with 16383 private static final int SIGNAL_MASK = 0B00000000_00000000_01111111_11111111; - private static AtomicReferenceArray signalPacketArray = new AtomicReferenceArray<>(SIGNAL_MASK + 1); + private static final AtomicReferenceArray signalPacketArray = new AtomicReferenceArray<>(SIGNAL_MASK + 1); /** * Session控制同步或异步的附加包，key：packetId */ - private static Map signalPacketAttachmentMap = new ConcurrentHashMap<>(1000); + private static final Map signalAttachmentMap = new ConcurrentHashMap<>(1000); public static void addSignalAttachment(int packetId) { var hash = packetId & SIGNAL_MASK; @@ -47,7 +47,7 @@ public class PacketSignalArray { return; } // logger.info("add [packetId:{}] [oldPacketId:{}]", packetId, signalPacketArray.get(hash)); - signalPacketAttachmentMap.put(packetId, packetId); + signalAttachmentMap.put(packetId, packetId); } @@ -59,7 +59,7 @@ public class PacketSignalArray { return; } // logger.info("remove [packetId:{}] [oldPacketId:{}]", packetId, oldPacketId); - signalPacketAttachmentMap.remove(packetId); + signalAttachmentMap.remove(packetId); } public static void status() { @@ -71,8 +71,8 @@ public class PacketSignalArray { } } - signalPacketAttachmentMap.forEach((key, value) -> { - logger.info("signalPacketAttachmentMap has attachment [key:{}][value:{}]", key, JsonUtils.object2String(value)); + signalAttachmentMap.forEach((key, value) -> { + logger.info("signalAttachmentMap has attachment [key:{}][value:{}]", key, JsonUtils.object2String(value)); }); } diff --git a/net/src/test/java/com/zfoo/net/router/PacketSignalMap.java b/net/src/test/java/com/zfoo/net/router/PacketSignalMap.java index 7f5dd850..5b77c698 100644 --- a/net/src/test/java/com/zfoo/net/router/PacketSignalMap.java +++ b/net/src/test/java/com/zfoo/net/router/PacketSignalMap.java @@ -29,20 +29,20 @@ public class PacketSignalMap { private static final Logger logger = LoggerFactory.getLogger(PacketSignalMap.class); - private static Map signalPacketAttachmentMap = new ConcurrentHashMap<>(1_0000); + private static final Map signalAttachmentMap = new ConcurrentHashMap<>(1_0000); public static void addSignalAttachment(int packetId) { - signalPacketAttachmentMap.put(packetId, packetId); + signalAttachmentMap.put(packetId, packetId); } public static void removeSignalAttachment(int packetId) { - signalPacketAttachmentMap.remove(packetId); + signalAttachmentMap.remove(packetId); } public static void status() { - signalPacketAttachmentMap.forEach((key, value) -> { - logger.info("signalPacketAttachmentMap has attachment [key:{}][value:{}]", key, JsonUtils.object2String(value)); + signalAttachmentMap.forEach((key, value) -> { + logger.info("signalAttachmentMap has attachment [key:{}][value:{}]", key, JsonUtils.object2String(value)); }); } diff --git a/net/src/test/java/com/zfoo/net/router/PacketSignalTest.java b/net/src/test/java/com/zfoo/net/router/SignalBridgeTest.java similarity index 91% rename from net/src/test/java/com/zfoo/net/router/PacketSignalTest.java rename to net/src/test/java/com/zfoo/net/router/SignalBridgeTest.java index 3f25498c..ddb6b9d8 100644 --- a/net/src/test/java/com/zfoo/net/router/PacketSignalTest.java +++ b/net/src/test/java/com/zfoo/net/router/SignalBridgeTest.java @@ -13,7 +13,7 @@ package com.zfoo.net.router; import com.zfoo.event.manager.EventBus; -import com.zfoo.net.router.route.PacketSignal; +import com.zfoo.net.router.route.SignalBridge; import com.zfoo.scheduler.util.TimeUtils; import com.zfoo.util.ThreadUtils; import org.junit.Ignore; @@ -28,13 +28,13 @@ import java.util.concurrent.atomic.AtomicInteger; * @version 3.0 */ @Ignore -public class PacketSignalTest { +public class SignalBridgeTest { - private AtomicInteger atomicInteger = new AtomicInteger(0); + private final AtomicInteger atomicInteger = new AtomicInteger(0); - private int executorSize = EventBus.EXECUTORS_SIZE / 2; - private int count = 100_0000; - private int totalIndex = 10; + private final int executorSize = EventBus.EXECUTORS_SIZE / 2; + private final int count = 100_0000; + private final int totalIndex = 10; @Test public void test() throws InterruptedException { @@ -63,7 +63,7 @@ public class PacketSignalTest { }); } countDownLatch.await(); - PacketSignal.status(); + SignalBridge.status(); System.out.println(TimeUtils.currentTimeMillis() - startTime); } diff --git a/net/src/test/resources/protocol.xml b/net/src/test/resources/protocol.xml index a4d0ce30..913f76e4 100644 --- a/net/src/test/resources/protocol.xml +++ b/net/src/test/resources/protocol.xml @@ -4,11 +4,11 @@ - - - - - + + + + +