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 02b4fdc8..d153c4d3 100644 --- a/net/src/main/java/com/zfoo/net/consumer/Consumer.java +++ b/net/src/main/java/com/zfoo/net/consumer/Consumer.java @@ -16,7 +16,6 @@ package com.zfoo.net.consumer; import com.zfoo.net.NetContext; import com.zfoo.net.consumer.balancer.AbstractConsumerLoadBalancer; import com.zfoo.net.consumer.balancer.IConsumerLoadBalancer; -import com.zfoo.net.packet.IPacket; import com.zfoo.net.packet.common.Error; import com.zfoo.net.router.Router; import com.zfoo.net.router.answer.AsyncAnswer; @@ -72,7 +71,7 @@ public class Consumer implements IConsumer { } @Override - public void send(IPacket packet, Object argument) { + public void send(Object packet, Object argument) { try { var loadBalancer = loadBalancer(ProtocolManager.moduleByProtocol(packet.getClass())); var session = loadBalancer.loadBalancer(packet, argument); @@ -84,7 +83,7 @@ public class Consumer implements IConsumer { } @Override - public SyncAnswer syncAsk(IPacket packet, Class answerClass, Object argument) throws Exception { + public SyncAnswer syncAsk(Object packet, Class answerClass, Object argument) throws Exception { var loadBalancer = loadBalancer(ProtocolManager.moduleByProtocol(packet.getClass())); var session = loadBalancer.loadBalancer(packet, argument); @@ -102,7 +101,7 @@ public class Consumer implements IConsumer { NetContext.getRouter().send(session, packet, clientSignalAttachment); - IPacket responsePacket = clientSignalAttachment.getResponseFuture().get(Router.DEFAULT_TIMEOUT, TimeUnit.MILLISECONDS); + Object responsePacket = clientSignalAttachment.getResponseFuture().get(Router.DEFAULT_TIMEOUT, TimeUnit.MILLISECONDS); if (responsePacket.getClass() == Error.class) { throw new ErrorResponseException((Error) responsePacket); @@ -123,7 +122,7 @@ public class Consumer implements IConsumer { } @Override - public AsyncAnswer asyncAsk(IPacket packet, Class answerClass, Object argument) { + public AsyncAnswer asyncAsk(Object packet, Class answerClass, Object argument) { var loadBalancer = loadBalancer(ProtocolManager.moduleByProtocol(packet.getClass())); var session = loadBalancer.loadBalancer(packet, argument); var asyncAnswer = NetContext.getRouter().asyncAsk(session, packet, answerClass, argument); diff --git a/net/src/main/java/com/zfoo/net/consumer/IConsumer.java b/net/src/main/java/com/zfoo/net/consumer/IConsumer.java index ce1ae1e0..27ef89e6 100644 --- a/net/src/main/java/com/zfoo/net/consumer/IConsumer.java +++ b/net/src/main/java/com/zfoo/net/consumer/IConsumer.java @@ -14,7 +14,6 @@ package com.zfoo.net.consumer; import com.zfoo.net.consumer.balancer.IConsumerLoadBalancer; -import com.zfoo.net.packet.IPacket; import com.zfoo.net.router.answer.AsyncAnswer; import com.zfoo.net.router.answer.SyncAnswer; import com.zfoo.protocol.registration.ProtocolModule; @@ -38,10 +37,10 @@ public interface IConsumer { * @param packet 需要发送的包 * @param argument 计算负载均衡的参数，比如用户的id */ - void send(IPacket packet, @Nullable Object argument); + void send(Object packet, @Nullable Object argument); - SyncAnswer syncAsk(IPacket packet, Class answerClass, @Nullable Object argument) throws Exception; + SyncAnswer syncAsk(Object packet, Class answerClass, @Nullable Object argument) throws Exception; - AsyncAnswer asyncAsk(IPacket packet, Class answerClass, @Nullable Object argument); + AsyncAnswer asyncAsk(Object packet, Class answerClass, @Nullable Object argument); } diff --git a/net/src/main/java/com/zfoo/net/consumer/balancer/AbstractConsumerLoadBalancer.java b/net/src/main/java/com/zfoo/net/consumer/balancer/AbstractConsumerLoadBalancer.java index 64c1d14b..fccac580 100644 --- a/net/src/main/java/com/zfoo/net/consumer/balancer/AbstractConsumerLoadBalancer.java +++ b/net/src/main/java/com/zfoo/net/consumer/balancer/AbstractConsumerLoadBalancer.java @@ -47,10 +47,6 @@ public abstract class AbstractConsumerLoadBalancer implements IConsumerLoadBalan return balancer; } - public List getSessionsByPacket(IPacket packet) { - return getSessionsByModule(ProtocolManager.moduleByProtocol(packet.getClass())); - } - public List getSessionsByModule(ProtocolModule module) { var list = new ArrayList(); NetContext.getSessionManager().forEachClientSession(new Consumer() { diff --git a/net/src/main/java/com/zfoo/net/consumer/balancer/ConsistentHashConsumerLoadBalancer.java b/net/src/main/java/com/zfoo/net/consumer/balancer/ConsistentHashConsumerLoadBalancer.java index 62b2e01b..82db63d1 100644 --- a/net/src/main/java/com/zfoo/net/consumer/balancer/ConsistentHashConsumerLoadBalancer.java +++ b/net/src/main/java/com/zfoo/net/consumer/balancer/ConsistentHashConsumerLoadBalancer.java @@ -14,7 +14,6 @@ package com.zfoo.net.consumer.balancer; import com.zfoo.net.NetContext; -import com.zfoo.net.packet.IPacket; import com.zfoo.net.session.Session; import com.zfoo.net.util.ConsistentHash; import com.zfoo.net.util.FastTreeMapIntLong; @@ -60,7 +59,7 @@ public class ConsistentHashConsumerLoadBalancer extends AbstractConsumerLoadBala * @return 调用的session */ @Override - public Session loadBalancer(IPacket packet, Object argument) { + public Session loadBalancer(Object packet, Object argument) { if (argument == null) { return RandomConsumerLoadBalancer.getInstance().loadBalancer(packet, argument); } 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 e331a151..77cd1df1 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,6 @@ package com.zfoo.net.consumer.balancer; -import com.zfoo.net.packet.IPacket; import com.zfoo.net.router.attachment.SignalAttachment; import com.zfoo.net.session.Session; import org.springframework.lang.Nullable; @@ -31,12 +30,12 @@ public interface IConsumerLoadBalancer { * @param argument 计算参数 * @return 一个服务提供者的session */ - Session loadBalancer(IPacket packet, @Nullable Object argument); + Session loadBalancer(Object packet, @Nullable Object argument); - default void beforeLoadBalancer(Session session, IPacket packet, SignalAttachment attachment) { + default void beforeLoadBalancer(Session session, Object packet, SignalAttachment attachment) { } - default void afterLoadBalancer(Session session, IPacket packet, SignalAttachment attachment) { + default void afterLoadBalancer(Session session, Object packet, SignalAttachment attachment) { } } diff --git a/net/src/main/java/com/zfoo/net/consumer/balancer/RandomConsumerLoadBalancer.java b/net/src/main/java/com/zfoo/net/consumer/balancer/RandomConsumerLoadBalancer.java index 86c13935..fdf39684 100644 --- a/net/src/main/java/com/zfoo/net/consumer/balancer/RandomConsumerLoadBalancer.java +++ b/net/src/main/java/com/zfoo/net/consumer/balancer/RandomConsumerLoadBalancer.java @@ -13,7 +13,6 @@ package com.zfoo.net.consumer.balancer; -import com.zfoo.net.packet.IPacket; import com.zfoo.net.session.Session; import com.zfoo.protocol.ProtocolManager; import com.zfoo.protocol.exception.RunException; @@ -37,7 +36,7 @@ public class RandomConsumerLoadBalancer extends AbstractConsumerLoadBalancer { } @Override - public Session loadBalancer(IPacket packet, Object argument) { + public Session loadBalancer(Object packet, Object argument) { var module = ProtocolManager.moduleByProtocol(packet.getClass()); var sessions = getSessionsByModule(module); diff --git a/net/src/main/java/com/zfoo/net/core/event/ServerExceptionEvent.java b/net/src/main/java/com/zfoo/net/core/event/ServerExceptionEvent.java index 5f3f3ed8..85d0c55e 100644 --- a/net/src/main/java/com/zfoo/net/core/event/ServerExceptionEvent.java +++ b/net/src/main/java/com/zfoo/net/core/event/ServerExceptionEvent.java @@ -13,8 +13,6 @@ package com.zfoo.net.core.event; import com.zfoo.event.model.IEvent; -import com.zfoo.net.packet.IPacket; -import com.zfoo.net.router.attachment.IAttachment; import com.zfoo.net.session.Session; /** @@ -24,11 +22,11 @@ import com.zfoo.net.session.Session; public class ServerExceptionEvent implements IEvent { private Session session; - private IPacket packet; - private IAttachment attachment; + private Object packet; + private Object attachment; private Exception exception; - public static ServerExceptionEvent valueOf(Session session, IPacket packet, IAttachment attachment, Exception exception) { + public static ServerExceptionEvent valueOf(Session session, Object packet, Object attachment, Exception exception) { var event = new ServerExceptionEvent(); event.session = session; event.packet = packet; @@ -45,19 +43,19 @@ public class ServerExceptionEvent implements IEvent { this.session = session; } - public IPacket getPacket() { + public Object getPacket() { return packet; } - public void setPacket(IPacket packet) { + public void setPacket(Object packet) { this.packet = packet; } - public IAttachment getAttachment() { + public Object getAttachment() { return attachment; } - public void setAttachment(IAttachment attachment) { + public void setAttachment(Object attachment) { this.attachment = attachment; } 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 index 3de9ea32..86b221ea 100644 --- 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 @@ -19,7 +19,6 @@ import com.zfoo.net.packet.DecodedPacketInfo; import com.zfoo.net.packet.EncodedPacketInfo; import com.zfoo.net.packet.IPacket; import com.zfoo.net.packet.PacketService; -import com.zfoo.net.router.attachment.IAttachment; import com.zfoo.protocol.ProtocolManager; import com.zfoo.protocol.buffer.ByteBufUtils; import com.zfoo.protocol.util.IOUtils; @@ -83,9 +82,9 @@ public class JProtobufTcpCodecHandler extends ByteToMessageCodec) ProtobufProxy.create(packet.getClass()); + var protobufCodec = (Codec) ProtobufProxy.create(packet.getClass()); byte[] bytes = protobufCodec.encode(packet); // header(4byte) + protocolId(2byte) buffer.writeInt(bytes.length + 2); diff --git a/net/src/main/java/com/zfoo/net/handler/codec/json/JsonPacket.java b/net/src/main/java/com/zfoo/net/handler/codec/json/JsonPacket.java index 60801c26..0aec4d84 100644 --- a/net/src/main/java/com/zfoo/net/handler/codec/json/JsonPacket.java +++ b/net/src/main/java/com/zfoo/net/handler/codec/json/JsonPacket.java @@ -13,9 +13,6 @@ package com.zfoo.net.handler.codec.json; -import com.zfoo.net.packet.IPacket; -import com.zfoo.net.router.attachment.IAttachment; - /** * @author godotg * @version 3.0 @@ -26,11 +23,11 @@ public class JsonPacket { private short attachmentId; - private IPacket packet; - private IAttachment attachment; + private Object packet; + private Object attachment; - public static JsonPacket valueOf(short protocolId, IPacket packet, short attachmentId, IAttachment attachment) { + public static JsonPacket valueOf(short protocolId, Object packet, short attachmentId, Object attachment) { var jsonPacket = new JsonPacket(); jsonPacket.protocolId = protocolId; jsonPacket.attachmentId = attachmentId; @@ -47,14 +44,6 @@ public class JsonPacket { this.protocolId = protocolId; } - public IPacket getPacket() { - return packet; - } - - public void setPacket(IPacket packet) { - this.packet = packet; - } - public short getAttachmentId() { return attachmentId; } @@ -63,11 +52,19 @@ public class JsonPacket { this.attachmentId = attachmentId; } - public IAttachment getAttachment() { + public Object getPacket() { + return packet; + } + + public void setPacket(Object packet) { + this.packet = packet; + } + + public Object getAttachment() { return attachment; } - public void setAttachment(IAttachment attachment) { + public void setAttachment(Object attachment) { this.attachment = attachment; } } diff --git a/net/src/main/java/com/zfoo/net/packet/EncodedPacketInfo.java b/net/src/main/java/com/zfoo/net/packet/EncodedPacketInfo.java index 81b217b8..01154b86 100644 --- a/net/src/main/java/com/zfoo/net/packet/EncodedPacketInfo.java +++ b/net/src/main/java/com/zfoo/net/packet/EncodedPacketInfo.java @@ -12,7 +12,6 @@ package com.zfoo.net.packet; -import com.zfoo.net.router.attachment.IAttachment; import org.springframework.lang.Nullable; /** @@ -26,58 +25,34 @@ public class EncodedPacketInfo { /** * 解码后的包 */ - private IPacket packet; + private Object packet; /** * 解码后的包的附加包 */ - private IAttachment attachment; + private Object attachment; - /** - * 长度 - */ - private int length; - /** - * 加密所用时间 - */ - private long encodedTime; - public static EncodedPacketInfo valueOf(IPacket packet, @Nullable IAttachment attachment) { + public static EncodedPacketInfo valueOf(Object packet, @Nullable Object attachment) { EncodedPacketInfo packetInfo = new EncodedPacketInfo(); packetInfo.packet = packet; packetInfo.attachment = attachment; return packetInfo; } - public IPacket getPacket() { + public Object getPacket() { return packet; } - public void setPacket(IPacket packet) { + public void setPacket(Object packet) { this.packet = packet; } - public IAttachment getAttachment() { + public Object getAttachment() { return attachment; } - public void setAttachment(IAttachment attachment) { + public void setAttachment(Object attachment) { this.attachment = attachment; } - - public int getLength() { - return length; - } - - public void setLength(int length) { - this.length = length; - } - - public long getEncodedTime() { - return encodedTime; - } - - public void setEncodedTime(long encodedTime) { - this.encodedTime = encodedTime; - } } diff --git a/net/src/main/java/com/zfoo/net/packet/IPacket.java b/net/src/main/java/com/zfoo/net/packet/IPacket.java index e7aa5d94..714555c3 100644 --- a/net/src/main/java/com/zfoo/net/packet/IPacket.java +++ b/net/src/main/java/com/zfoo/net/packet/IPacket.java @@ -15,12 +15,7 @@ package com.zfoo.net.packet; import com.zfoo.protocol.ProtocolManager; /** - * 所有协议类都必须实现这个接口，协议类必须是简单的javabean，不能继承任何其它的类，但是可以继承接口 - *

- * 现在IPacket的接口只是一个标识接口，继承IPacket的设计主要是为了让代码更优雅，容易理解一点，改为只继承Object也并没有很大工作量 - * 继承IPacket的设计还有跨语言层面上的考虑，极大的简化了实现其它语言的序列化和反序列化难度，统一了其它语言的代码实现 - *

- * 为了防止代码里Object满天飞，避免协议层和po层混用对象造成一些潜在的并发问题，zfoo强制要求协议类必须实现IPacket接口 + * The interface of IPacket is just an identification interface * * @author godotg * @version 3.0 @@ -28,9 +23,10 @@ import com.zfoo.protocol.ProtocolManager; public interface IPacket { /** - * 这个类的协议号，重写这个方法，使用多态获取协议号，可以微弱的提高一点性能 + * The protocol id of this class *

- * @return 协议号Id + * + * @return protocol id */ default short protocolId() { return ProtocolManager.protocolId(this.getClass()); diff --git a/net/src/main/java/com/zfoo/net/packet/IPacketService.java b/net/src/main/java/com/zfoo/net/packet/IPacketService.java index d375a6e5..e73eaee2 100644 --- a/net/src/main/java/com/zfoo/net/packet/IPacketService.java +++ b/net/src/main/java/com/zfoo/net/packet/IPacketService.java @@ -12,7 +12,6 @@ package com.zfoo.net.packet; -import com.zfoo.net.router.attachment.IAttachment; import io.netty.buffer.ByteBuf; import org.springframework.lang.Nullable; @@ -26,6 +25,6 @@ public interface IPacketService { DecodedPacketInfo read(ByteBuf buffer); - void write(ByteBuf buffer, IPacket packet, @Nullable IAttachment attachment); + void write(ByteBuf buffer, Object packet, @Nullable Object attachment); } diff --git a/net/src/main/java/com/zfoo/net/packet/PacketService.java b/net/src/main/java/com/zfoo/net/packet/PacketService.java index 176a9c43..02a838f2 100644 --- a/net/src/main/java/com/zfoo/net/packet/PacketService.java +++ b/net/src/main/java/com/zfoo/net/packet/PacketService.java @@ -156,7 +156,7 @@ public class PacketService implements IPacketService { } @Override - public void write(ByteBuf buffer, IPacket packet, IAttachment attachment) { + public void write(ByteBuf buffer, Object packet, Object attachment) { if (packet == null) { logger.error("packet is null and can not be sent."); diff --git a/net/src/main/java/com/zfoo/net/router/IRouter.java b/net/src/main/java/com/zfoo/net/router/IRouter.java index a329652a..017ddd9e 100644 --- a/net/src/main/java/com/zfoo/net/router/IRouter.java +++ b/net/src/main/java/com/zfoo/net/router/IRouter.java @@ -13,10 +13,8 @@ package com.zfoo.net.router; -import com.zfoo.net.packet.IPacket; import com.zfoo.net.router.answer.AsyncAnswer; import com.zfoo.net.router.answer.SyncAnswer; -import com.zfoo.net.router.attachment.IAttachment; import com.zfoo.net.session.Session; import org.springframework.lang.Nullable; @@ -30,13 +28,13 @@ public interface IRouter { * EN:send() and receive() are the entry points for sending and receiving messages, which can be called directly * CN:send()和receive()是消息的发送和接收的入口，可以直接调用 */ - void send(Session session, IPacket packet); + void send(Session session, Object packet); - void send(Session session, IPacket packet, @Nullable IAttachment attachment); + void send(Session session, Object packet, @Nullable Object attachment); - void receive(Session session, IPacket packet, @Nullable IAttachment attachment); + void receive(Session session, Object packet, @Nullable Object attachment); - void atReceiver(Session session, IPacket packet, @Nullable IAttachment attachment); + void atReceiver(Session session, Object packet, @Nullable Object attachment); /** * attention：syncAsk和asyncAsk只能客户端调用 @@ -56,8 +54,8 @@ public interface IRouter { * @return 服务器返回的消息Response * @throws Exception 如果超时或者其它异常 */ - SyncAnswer syncAsk(Session session, IPacket packet, @Nullable Class answerClass, @Nullable Object argument) throws Exception; + SyncAnswer syncAsk(Session session, Object packet, @Nullable Class answerClass, @Nullable Object argument) throws Exception; - AsyncAnswer asyncAsk(Session session, IPacket packet, @Nullable Class answerClass, @Nullable Object argument); + AsyncAnswer asyncAsk(Session session, Object packet, @Nullable Class answerClass, @Nullable Object argument); } diff --git a/net/src/main/java/com/zfoo/net/router/Router.java b/net/src/main/java/com/zfoo/net/router/Router.java index 5e2c8fb8..a8deb3ee 100644 --- a/net/src/main/java/com/zfoo/net/router/Router.java +++ b/net/src/main/java/com/zfoo/net/router/Router.java @@ -20,13 +20,11 @@ import com.zfoo.net.core.gateway.model.AuthUidToGatewayCheck; import com.zfoo.net.core.gateway.model.AuthUidToGatewayConfirm; import com.zfoo.net.core.gateway.model.AuthUidToGatewayEvent; import com.zfoo.net.packet.EncodedPacketInfo; -import com.zfoo.net.packet.IPacket; import com.zfoo.net.packet.common.Error; import com.zfoo.net.packet.common.Heartbeat; import com.zfoo.net.router.answer.AsyncAnswer; import com.zfoo.net.router.answer.SyncAnswer; 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.router.exception.ErrorResponseException; import com.zfoo.net.router.exception.NetTimeOutException; @@ -64,13 +62,13 @@ public class Router implements IRouter { * atReceiver会设置attachment，但是在方法调用完成会取消，不需要过多关注。 * asyncAsk会再次设置attachment，需要重点关注。 */ - private final FastThreadLocal serverReceiverAttachmentThreadLocal = new FastThreadLocal<>(); + private final FastThreadLocal serverReceiverAttachmentThreadLocal = new FastThreadLocal<>(); /** * 在服务端收到数据后，会调用这个方法. 这个方法在BaseRouteHandler.java的channelRead中被调用 */ @Override - public void receive(Session session, IPacket packet, @Nullable IAttachment attachment) { + public void receive(Session session, Object packet, @Nullable Object attachment) { if (packet.getClass() == Heartbeat.class) { logger.info("heartbeat"); return; @@ -78,65 +76,60 @@ public class Router implements IRouter { // 发送者（客户端）同步和异步消息的接收，发送者通过signalId判断重复 if (attachment != null) { - switch (attachment.packetType()) { - case SIGNAL_PACKET: - var signalAttachment = (SignalAttachment) attachment; + if (attachment.getClass() == SignalAttachment.class) { + var signalAttachment = (SignalAttachment) attachment; - if (signalAttachment.isClient()) { - // 服务器收到signalAttachment，不做任何处理 - signalAttachment.setClient(false); + if (signalAttachment.isClient()) { + // 服务器收到signalAttachment，不做任何处理 + signalAttachment.setClient(false); + } else { + // 客户端收到服务器应答，客户端发送的时候isClient为true，服务器收到的时候将其设置为false + var removedAttachment = (SignalAttachment) SignalBridge.removeSignalAttachment(signalAttachment); + if (removedAttachment != null) { + // 这里会让之前的CompletableFuture得到结果，从而像asyncAsk之类的回调到结果 + removedAttachment.getResponseFuture().complete(packet); } else { - // 客户端收到服务器应答，客户端发送的时候isClient为true，服务器收到的时候将其设置为false - var removedAttachment = (SignalAttachment) SignalBridge.removeSignalAttachment(signalAttachment); - if (removedAttachment != null) { - // 这里会让之前的CompletableFuture得到结果，从而像asyncAsk之类的回调到结果 - removedAttachment.getResponseFuture().complete(packet); - } else { - logger.error("client receives packet:[{}] and attachment:[{}] from server, but clientAttachmentMap has no attachment, perhaps timeout exception.", JsonUtils.object2String(packet), JsonUtils.object2String(attachment)); - } - // 注意：这个return，这样子，asyncAsk的结果就返回了。 - return; + logger.error("client receives packet:[{}] and attachment:[{}] from server, but clientAttachmentMap has no attachment, perhaps timeout exception.", JsonUtils.object2String(packet), JsonUtils.object2String(attachment)); } - break; - case GATEWAY_PACKET: - var gatewayAttachment = (GatewayAttachment) attachment; + // 注意：这个return，这样子，asyncAsk的结果就返回了。 + return; + } + } else if (attachment.getClass() == GatewayAttachment.class) { + var gatewayAttachment = (GatewayAttachment) attachment; - // 如：在网关监听到GatewaySessionInactiveEvent后，这时告诉home时，这个client参数设置的true - // 注意：此时并没有return，这样子网关的消息才能发给home，在home进行处理LogoutRequest消息的处理 - if (gatewayAttachment.isClient()) { - gatewayAttachment.setClient(false); - } else { - // 这里是：别的服务提供者提供授权给网关，比如：在玩家登录后，home服查到了玩家uid，然后发给Gateway服 - var gatewaySession = NetContext.getSessionManager().getServerSession(gatewayAttachment.getSid()); - if (gatewaySession != null) { - var signalAttachmentInGatewayAttachment = gatewayAttachment.getSignalAttachment(); - if (signalAttachmentInGatewayAttachment != null) { - signalAttachmentInGatewayAttachment.setClient(false); - } + // 如：在网关监听到GatewaySessionInactiveEvent后，这时告诉home时，这个client参数设置的true + // 注意：此时并没有return，这样子网关的消息才能发给home，在home进行处理LogoutRequest消息的处理 + if (gatewayAttachment.isClient()) { + gatewayAttachment.setClient(false); + } else { + // 这里是：别的服务提供者提供授权给网关，比如：在玩家登录后，home服查到了玩家uid，然后发给Gateway服 + var gatewaySession = NetContext.getSessionManager().getServerSession(gatewayAttachment.getSid()); + if (gatewaySession != null) { + var signalAttachmentInGatewayAttachment = gatewayAttachment.getSignalAttachment(); + if (signalAttachmentInGatewayAttachment != null) { + signalAttachmentInGatewayAttachment.setClient(false); + } - // 网关授权，授权完成直接返回 - // 注意：这个 AuthUidToGatewayCheck 是在home的LoginController中处理完登录后，把消息发给网关进行授权 - if (AuthUidToGatewayCheck.class == packet.getClass()) { - var uid = ((AuthUidToGatewayCheck) packet).getUid(); - if (uid <= 0) { - logger.error("错误的网关授权信息，uid必须大于0"); - return; - } - gatewaySession.setUid(uid); - EventBus.post(AuthUidToGatewayEvent.valueOf(gatewaySession.getSid(), uid)); - - NetContext.getRouter().send(session, AuthUidToGatewayConfirm.valueOf(uid), new GatewayAttachment(gatewaySession)); + // 网关授权，授权完成直接返回 + // 注意：这个 AuthUidToGatewayCheck 是在home的LoginController中处理完登录后，把消息发给网关进行授权 + if (AuthUidToGatewayCheck.class == packet.getClass()) { + var uid = ((AuthUidToGatewayCheck) packet).getUid(); + if (uid <= 0) { + logger.error("错误的网关授权信息，uid必须大于0"); return; } - send(gatewaySession, packet, gatewayAttachment.attachment()); - } else { - 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(attachment), gatewayAttachment.getSid()); + gatewaySession.setUid(uid); + EventBus.post(AuthUidToGatewayEvent.valueOf(gatewaySession.getSid(), uid)); + + NetContext.getRouter().send(session, AuthUidToGatewayConfirm.valueOf(uid), new GatewayAttachment(gatewaySession)); + return; } - return; + send(gatewaySession, packet, gatewayAttachment.attachment()); + } else { + 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(attachment), gatewayAttachment.getSid()); } - break; - default: - break; + return; + } } } @@ -146,7 +139,7 @@ public class Router implements IRouter { } @Override - public void send(Session session, IPacket packet, IAttachment attachment) { + public void send(Session session, Object packet, Object attachment) { if (session == null) { logger.error("session is null and can not be sent."); return; @@ -167,7 +160,7 @@ public class Router implements IRouter { } @Override - public void send(Session session, IPacket packet) { + public void send(Session session, Object packet) { // 服务器异步返回的消息的发送会有signalAttachment，验证返回的消息是否满足 var serverSignalAttachment = serverReceiverAttachmentThreadLocal.get(); send(session, packet, serverSignalAttachment); @@ -175,7 +168,7 @@ public class Router implements IRouter { @Override - public SyncAnswer syncAsk(Session session, IPacket packet, @Nullable Class answerClass, @Nullable Object argument) throws Exception { + public SyncAnswer syncAsk(Session session, Object packet, @Nullable Class answerClass, @Nullable Object argument) throws Exception { var clientSignalAttachment = new SignalAttachment(); var taskExecutorHash = TaskBus.calTaskExecutorHash(argument); clientSignalAttachment.setTaskExecutorHash(taskExecutorHash); @@ -186,7 +179,7 @@ public class Router implements IRouter { // 里面调用的依然是：send方法发送消息 send(session, packet, clientSignalAttachment); - IPacket responsePacket = clientSignalAttachment.getResponseFuture().get(DEFAULT_TIMEOUT, TimeUnit.MILLISECONDS); + Object responsePacket = clientSignalAttachment.getResponseFuture().get(DEFAULT_TIMEOUT, TimeUnit.MILLISECONDS); if (responsePacket.getClass() == Error.class) { throw new ErrorResponseException((Error) responsePacket); @@ -210,7 +203,7 @@ public class Router implements IRouter { * 2.这个argument的参数，只用于provider处哪个线程执行，其实就是hashId，如：工会业务，则传入guildId，回调回来后，一定会在发起者线程。 */ @Override - public AsyncAnswer asyncAsk(Session session, IPacket packet, @Nullable Class answerClass, @Nullable Object argument) { + public AsyncAnswer asyncAsk(Session session, Object packet, @Nullable Class answerClass, @Nullable Object argument) { var clientSignalAttachment = new SignalAttachment(); var taskExecutorHash = TaskBus.calTaskExecutorHash(argument); @@ -292,7 +285,7 @@ public class Router implements IRouter { * 接收者同时只能处理一个session的一个包，同一个发送者发送过来的包排队处理 */ @Override - public void atReceiver(Session session, IPacket packet, IAttachment attachment) { + public void atReceiver(Session session, Object packet, Object attachment) { try { // 接收者（服务器）同步和异步消息的接收 if (attachment != null) { 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 94d0cddd..2b911708 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,6 @@ package com.zfoo.net.router.answer; -import com.zfoo.net.packet.IPacket; import com.zfoo.net.router.attachment.SignalAttachment; import com.zfoo.protocol.util.ThreadUtils; @@ -25,7 +24,7 @@ import java.util.function.Consumer; * @author godotg * @version 3.0 */ -public class AsyncAnswer implements IAsyncAnswer { +public class AsyncAnswer implements IAsyncAnswer { private T futurePacket; diff --git a/net/src/main/java/com/zfoo/net/router/answer/IAsyncAnswer.java b/net/src/main/java/com/zfoo/net/router/answer/IAsyncAnswer.java index fe16da9a..798056d4 100644 --- a/net/src/main/java/com/zfoo/net/router/answer/IAsyncAnswer.java +++ b/net/src/main/java/com/zfoo/net/router/answer/IAsyncAnswer.java @@ -13,7 +13,6 @@ package com.zfoo.net.router.answer; -import com.zfoo.net.packet.IPacket; import java.util.function.Consumer; @@ -21,7 +20,7 @@ import java.util.function.Consumer; * @author godotg * @version 3.0 */ -public interface IAsyncAnswer { +public interface IAsyncAnswer { IAsyncAnswer thenAccept(Consumer consumer); 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 01778b7a..b677b2a8 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,14 +13,13 @@ package com.zfoo.net.router.answer; -import com.zfoo.net.packet.IPacket; import com.zfoo.net.router.attachment.SignalAttachment; /** * @author godotg * @version 3.0 */ -public interface ISyncAnswer { +public interface ISyncAnswer { /** * The return packet for the synchronization request 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 f0f11a52..060bf894 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,14 +13,13 @@ package com.zfoo.net.router.answer; -import com.zfoo.net.packet.IPacket; import com.zfoo.net.router.attachment.SignalAttachment; /** * @author godotg * @version 3.0 */ -public class SyncAnswer implements ISyncAnswer { +public class SyncAnswer implements ISyncAnswer { private final T packet; diff --git a/net/src/main/java/com/zfoo/net/router/attachment/SignalAttachment.java b/net/src/main/java/com/zfoo/net/router/attachment/SignalAttachment.java index 9e82e625..a44baeb0 100644 --- a/net/src/main/java/com/zfoo/net/router/attachment/SignalAttachment.java +++ b/net/src/main/java/com/zfoo/net/router/attachment/SignalAttachment.java @@ -13,7 +13,6 @@ package com.zfoo.net.router.attachment; import com.fasterxml.jackson.annotation.JsonIgnore; -import com.zfoo.net.packet.IPacket; import com.zfoo.protocol.anno.Protocol; import com.zfoo.scheduler.util.TimeUtils; @@ -60,7 +59,7 @@ public class SignalAttachment implements IAttachment { * CN:客户端收到服务器回复的时候回调的方法 */ @JsonIgnore - private transient CompletableFuture responseFuture = new CompletableFuture<>(); + private transient CompletableFuture responseFuture = new CompletableFuture<>(); public SignalAttachment() { } @@ -130,11 +129,11 @@ public class SignalAttachment implements IAttachment { } - public CompletableFuture getResponseFuture() { + public CompletableFuture getResponseFuture() { return responseFuture; } - public void setResponseFuture(CompletableFuture responseFuture) { + public void setResponseFuture(CompletableFuture responseFuture) { this.responseFuture = responseFuture; } } 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 b189f69f..db1287d6 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,8 +13,6 @@ package com.zfoo.net.router.receiver; -import com.zfoo.net.packet.IPacket; -import com.zfoo.net.router.attachment.IAttachment; import com.zfoo.net.session.Session; /** @@ -23,6 +21,6 @@ import com.zfoo.net.session.Session; */ public interface IPacketReceiver { - void invoke(Session session, IPacket packet, IAttachment attachment); + void invoke(Session session, Object packet, Object 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 42875935..87dee432 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,8 +13,6 @@ package com.zfoo.net.router.receiver; -import com.zfoo.net.packet.IPacket; -import com.zfoo.net.router.attachment.IAttachment; import com.zfoo.net.session.Session; import com.zfoo.protocol.util.ReflectionUtils; @@ -58,7 +56,7 @@ public class PacketReceiverDefinition implements IPacketReceiver { } @Override - public void invoke(Session session, IPacket packet, IAttachment attachment) { + public void invoke(Session session, Object packet, Object attachment) { if (attachmentClazz == null) { ReflectionUtils.invokeMethod(bean, method, session, packet); } else { diff --git a/net/src/main/java/com/zfoo/net/router/route/PacketBus.java b/net/src/main/java/com/zfoo/net/router/route/PacketBus.java index 777059d4..e1163b11 100644 --- a/net/src/main/java/com/zfoo/net/router/route/PacketBus.java +++ b/net/src/main/java/com/zfoo/net/router/route/PacketBus.java @@ -13,6 +13,7 @@ package com.zfoo.net.router.route; +import com.zfoo.net.anno.PacketReceiver; import com.zfoo.net.packet.IPacket; import com.zfoo.net.packet.PacketService; import com.zfoo.net.router.attachment.GatewayAttachment; @@ -20,7 +21,6 @@ import com.zfoo.net.router.attachment.IAttachment; import com.zfoo.net.router.attachment.SignalAttachment; import com.zfoo.net.router.receiver.EnhanceUtils; import com.zfoo.net.router.receiver.IPacketReceiver; -import com.zfoo.net.anno.PacketReceiver; import com.zfoo.net.router.receiver.PacketReceiverDefinition; import com.zfoo.net.session.Session; import com.zfoo.protocol.ProtocolManager; @@ -52,7 +52,7 @@ public abstract class PacketBus { /** * The routing of the message */ - public static void route(Session session, IPacket packet, IAttachment attachment) { + public static void route(Session session, Object packet, Object attachment) { var receiver = receiverMap.get(ProtocolManager.protocolId(packet.getClass())); if (receiver == null) { var name = packet.getClass().getSimpleName(); diff --git a/net/src/main/java/com/zfoo/net/task/PacketReceiverTask.java b/net/src/main/java/com/zfoo/net/task/PacketReceiverTask.java index f25ef531..5d15aa38 100644 --- a/net/src/main/java/com/zfoo/net/task/PacketReceiverTask.java +++ b/net/src/main/java/com/zfoo/net/task/PacketReceiverTask.java @@ -13,8 +13,6 @@ package com.zfoo.net.task; import com.zfoo.net.NetContext; -import com.zfoo.net.packet.IPacket; -import com.zfoo.net.router.attachment.IAttachment; import com.zfoo.net.session.Session; /** @@ -24,10 +22,10 @@ import com.zfoo.net.session.Session; public final class PacketReceiverTask implements Runnable { private Session session; - private IPacket packet; - private IAttachment attachment; + private Object packet; + private Object attachment; - public PacketReceiverTask(Session session, IPacket packet, IAttachment attachment) { + public PacketReceiverTask(Session session, Object packet, Object attachment) { this.session = session; this.packet = packet; this.attachment = attachment; @@ -38,6 +36,7 @@ public final class PacketReceiverTask implements Runnable { NetContext.getRouter().atReceiver(session, packet, attachment); } + public Session getSession() { return session; } @@ -46,19 +45,19 @@ public final class PacketReceiverTask implements Runnable { this.session = session; } - public IPacket getPacket() { + public Object getPacket() { return packet; } - public void setPacket(IPacket packet) { + public void setPacket(Object packet) { this.packet = packet; } - public IAttachment getAttachment() { + public Object getAttachment() { return attachment; } - public void setAttachment(IAttachment attachment) { + public void setAttachment(Object attachment) { this.attachment = attachment; } } diff --git a/net/src/main/java/com/zfoo/net/task/TaskBus.java b/net/src/main/java/com/zfoo/net/task/TaskBus.java index 8a2df6a1..3a79e90c 100644 --- a/net/src/main/java/com/zfoo/net/task/TaskBus.java +++ b/net/src/main/java/com/zfoo/net/task/TaskBus.java @@ -17,15 +17,14 @@ import com.zfoo.event.manager.EventBus; import com.zfoo.net.NetContext; 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.SignalAttachment; import com.zfoo.net.session.Session; import com.zfoo.protocol.collection.concurrent.CopyOnWriteHashMapLongObject; import com.zfoo.protocol.util.AssertionUtils; -import com.zfoo.protocol.util.StringUtils; -import com.zfoo.scheduler.manager.SchedulerBus; import com.zfoo.protocol.util.RandomUtils; +import com.zfoo.protocol.util.StringUtils; import com.zfoo.protocol.util.ThreadUtils; +import com.zfoo.scheduler.manager.SchedulerBus; import io.netty.util.concurrent.FastThreadLocalThread; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -134,23 +133,16 @@ public final class TaskBus { } } - private static void dispatchByAttachment(IAttachment attachment, PacketReceiverTask task) { - switch (attachment.packetType()) { - case SIGNAL_PACKET: - execute(((SignalAttachment) attachment).taskExecutorHash(), task); - break; - case GATEWAY_PACKET: - execute(((GatewayAttachment) attachment).taskExecutorHash(), task); - break; - case HTTP_PACKET: - execute(((HttpAttachment) attachment).taskExecutorHash(), task); - break; - case SIGNAL_ONLY_PACKET: - case NO_ANSWER_PACKET: - case UDP_PACKET: - dispatchBySession(task.getSession(), task); - break; - default: + private static void dispatchByAttachment(Object attachment, PacketReceiverTask task) { + var attachmentClass = attachment.getClass(); + if (attachmentClass == SignalAttachment.class) { + execute(((SignalAttachment) attachment).taskExecutorHash(), task); + } else if (attachmentClass == GatewayAttachment.class) { + execute(((GatewayAttachment) attachment).taskExecutorHash(), task); + } else if (attachmentClass == HttpAttachment.class) { + execute(((HttpAttachment) attachment).taskExecutorHash(), task); + } else { + dispatchBySession(task.getSession(), task); } }