tiennm99/zfoo
1
0
Fork 0
mirror of https://github.com/tiennm99/zfoo.git synced 2026-08-23 22:27:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

ref[net]: No need to inherit the IPacket interface anymore

Browse Source
This commit is contained in:
godotg
2023-09-09 09:30:29 +08:00
parent 2555c9b131
commit 7c2615fcdf
25 changed files with 137 additions and 209 deletions
Download Patch File Download Diff File Expand all files Collapse all files
net/src/main/java/com/zfoo/net/consumer/Consumer.java
+4 -5
View File
@@ -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 <T extends IPacket> SyncAnswer<T> syncAsk(IPacket packet, Class<T> answerClass, Object argument) throws Exception {
public <T> SyncAnswer<T> syncAsk(Object packet, Class<T> 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 <T extends IPacket> AsyncAnswer<T> asyncAsk(IPacket packet, Class<T> answerClass, Object argument) {
public <T> AsyncAnswer<T> asyncAsk(Object packet, Class<T> 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);
net/src/main/java/com/zfoo/net/consumer/IConsumer.java
+3 -4
View File
@@ -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);
<T extends IPacket> SyncAnswer<T> syncAsk(IPacket packet, Class<T> answerClass, @Nullable Object argument) throws Exception;
<T> SyncAnswer<T> syncAsk(Object packet, Class<T> answerClass, @Nullable Object argument) throws Exception;
<T extends IPacket> AsyncAnswer<T> asyncAsk(IPacket packet, Class<T> answerClass, @Nullable Object argument);
<T> AsyncAnswer<T> asyncAsk(Object packet, Class<T> answerClass, @Nullable Object argument);
}
net/src/main/java/com/zfoo/net/consumer/balancer/AbstractConsumerLoadBalancer.java
-4
View File
@@ -47,10 +47,6 @@ public abstract class AbstractConsumerLoadBalancer implements IConsumerLoadBalan
return balancer;
}
public List<Session> getSessionsByPacket(IPacket packet) {
return getSessionsByModule(ProtocolManager.moduleByProtocol(packet.getClass()));
}
public List<Session> getSessionsByModule(ProtocolModule module) {
var list = new ArrayList<Session>();
NetContext.getSessionManager().forEachClientSession(new Consumer<Session>() {
net/src/main/java/com/zfoo/net/consumer/balancer/ConsistentHashConsumerLoadBalancer.java
+1 -2
View File
@@ -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);
}
net/src/main/java/com/zfoo/net/consumer/balancer/IConsumerLoadBalancer.java
+3 -4
View File
@@ -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) {
}
}
net/src/main/java/com/zfoo/net/consumer/balancer/RandomConsumerLoadBalancer.java
+1 -2
View File
@@ -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);
net/src/main/java/com/zfoo/net/core/event/ServerExceptionEvent.java
+7 -9
View File
@@ -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;
}
net/src/main/java/com/zfoo/net/handler/codec/jprotobuf/JProtobufTcpCodecHandler.java
+2 -3
View File
@@ -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<EncodedPacketIn
return DecodedPacketInfo.valueOf((IPacket) packet, null);
}
public void write(ByteBuf buffer, IPacket packet, IAttachment attachment) throws IOException {
public void write(ByteBuf buffer, Object packet, Object attachment) throws IOException {
// 写入protobuf协议
var protobufCodec = (Codec<IPacket>) ProtobufProxy.create(packet.getClass());
var protobufCodec = (Codec<Object>) ProtobufProxy.create(packet.getClass());
byte[] bytes = protobufCodec.encode(packet);
// header(4byte) + protocolId(2byte)
buffer.writeInt(bytes.length + 2);
net/src/main/java/com/zfoo/net/handler/codec/json/JsonPacket.java
+13 -16
View File
@@ -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;
}
}
net/src/main/java/com/zfoo/net/packet/EncodedPacketInfo.java
+7 -32
View File
@@ -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;
}
}
net/src/main/java/com/zfoo/net/packet/IPacket.java
+4 -8
View File
@@ -15,12 +15,7 @@ package com.zfoo.net.packet;
import com.zfoo.protocol.ProtocolManager;
/**
* 所有协议类都必须实现这个接口，协议类必须是简单的javabean，不能继承任何其它的类，但是可以继承接口
* <p>
* 现在IPacket的接口只是一个标识接口，继承IPacket的设计主要是为了让代码更优雅，容易理解一点，改为只继承Object也并没有很大工作量
* 继承IPacket的设计还有跨语言层面上的考虑，极大的简化了实现其它语言的序列化和反序列化难度，统一了其它语言的代码实现
* <p>
* 为了防止代码里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
* <p>
* @return 协议号Id
*
* @return protocol id
*/
default short protocolId() {
return ProtocolManager.protocolId(this.getClass());
net/src/main/java/com/zfoo/net/packet/IPacketService.java
+1 -2
View File
@@ -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);
}
net/src/main/java/com/zfoo/net/packet/PacketService.java
+1 -1
View File
@@ -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.");
net/src/main/java/com/zfoo/net/router/IRouter.java
+6 -8
View File
@@ -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 如果超时或者其它异常
*/
<T extends IPacket> SyncAnswer<T> syncAsk(Session session, IPacket packet, @Nullable Class<T> answerClass, @Nullable Object argument) throws Exception;
<T> SyncAnswer<T> syncAsk(Session session, Object packet, @Nullable Class<T> answerClass, @Nullable Object argument) throws Exception;
<T extends IPacket> AsyncAnswer<T> asyncAsk(Session session, IPacket packet, @Nullable Class<T> answerClass, @Nullable Object argument);
<T> AsyncAnswer<T> asyncAsk(Session session, Object packet, @Nullable Class<T> answerClass, @Nullable Object argument);
}
net/src/main/java/com/zfoo/net/router/Router.java
+53 -60
View File
@@ -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<IAttachment> serverReceiverAttachmentThreadLocal = new FastThreadLocal<>();
private final FastThreadLocal<Object> 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 <T extends IPacket> SyncAnswer<T> syncAsk(Session session, IPacket packet, @Nullable Class<T> answerClass, @Nullable Object argument) throws Exception {
public <T> SyncAnswer<T> syncAsk(Session session, Object packet, @Nullable Class<T> 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 <T extends IPacket> AsyncAnswer<T> asyncAsk(Session session, IPacket packet, @Nullable Class<T> answerClass, @Nullable Object argument) {
public <T> AsyncAnswer<T> asyncAsk(Session session, Object packet, @Nullable Class<T> 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) {
net/src/main/java/com/zfoo/net/router/answer/AsyncAnswer.java
+1 -2
View File
@@ -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<T extends IPacket> implements IAsyncAnswer<T> {
public class AsyncAnswer<T> implements IAsyncAnswer<T> {
private T futurePacket;
net/src/main/java/com/zfoo/net/router/answer/IAsyncAnswer.java
+1 -2
View File
@@ -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<T extends IPacket> {
public interface IAsyncAnswer<T> {
IAsyncAnswer<T> thenAccept(Consumer<T> consumer);
net/src/main/java/com/zfoo/net/router/answer/ISyncAnswer.java
+1 -2
View File
@@ -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<T extends IPacket> {
public interface ISyncAnswer<T> {
/**
* The return packet for the synchronization request
net/src/main/java/com/zfoo/net/router/answer/SyncAnswer.java
+1 -2
View File
@@ -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<T extends IPacket> implements ISyncAnswer<T> {
public class SyncAnswer<T> implements ISyncAnswer<T> {
private final T packet;
net/src/main/java/com/zfoo/net/router/attachment/SignalAttachment.java
+3 -4
View File
@@ -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<IPacket> responseFuture = new CompletableFuture<>();
private transient CompletableFuture<Object> responseFuture = new CompletableFuture<>();
public SignalAttachment() {
}
@@ -130,11 +129,11 @@ public class SignalAttachment implements IAttachment {
}
public CompletableFuture<IPacket> getResponseFuture() {
public CompletableFuture<Object> getResponseFuture() {
return responseFuture;
}
public void setResponseFuture(CompletableFuture<IPacket> responseFuture) {
public void setResponseFuture(CompletableFuture<Object> responseFuture) {
this.responseFuture = responseFuture;
}
}
net/src/main/java/com/zfoo/net/router/receiver/IPacketReceiver.java
+1 -3
View File
@@ -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);
}
net/src/main/java/com/zfoo/net/router/receiver/PacketReceiverDefinition.java
+1 -3
View File
@@ -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 {
net/src/main/java/com/zfoo/net/router/route/PacketBus.java
+2 -2
View File
@@ -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();
net/src/main/java/com/zfoo/net/task/PacketReceiverTask.java
+8 -9
View File
@@ -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;
}
}
net/src/main/java/com/zfoo/net/task/TaskBus.java
+12 -20
View File
@@ -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);
}
}