ref[route]: refactor the routing rules for message packets

This commit is contained in:
godotg committed 2023-04-26 12:15:28 +08:00
1 parent e0064ec720
commit 0041a7ad31
5 files changed
+11 -33

No files matched your search

@@ -30,6 +30,7 @@ import com.zfoo.protocol.exception.RunException;
import com.zfoo.protocol.util.AssertionUtils;
import com.zfoo.protocol.util.ReflectionUtils;
import com.zfoo.protocol.util.StringUtils;
import io.netty.util.collection.ShortObjectHashMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -46,16 +47,18 @@ public abstract class PacketBus {
private static final Logger logger = LoggerFactory.getLogger(PacketBus.class);
private static final ShortObjectHashMap<IPacketReceiver> receiverMap = new ShortObjectHashMap<>();
/**
* The routing of the message
*/
public static void route(Session session, IPacket packet, IAttachment attachment) {
var packetReceiver = (IPacketReceiver) ProtocolManager.getProtocol(packet.protocolId()).receiver();
if (packetReceiver == null) {
var receiver = receiverMap.get(packet.protocolId());
if (receiver == null) {
var name = packet.getClass().getSimpleName();
throw new RuntimeException(StringUtils.format("no any packetReceiver:[at{}] found for this packet:[{}] or no GatewayAttachment sent back if this server is gateway", name, name));
}
packetReceiver.invoke(session, packet, attachment);
receiver.invoke(session, packet, attachment);
}
@@ -112,15 +115,11 @@ public abstract class PacketBus {
}
try {
var protocolRegistration = ProtocolManager.getProtocol(protocolId);
AssertionUtils.isNull(protocolRegistration.receiver(), "duplicate protocol registration, @PacketReceiver [class:{}] is repeatedly received [at{}]", packetClazz.getSimpleName(), packetClazz.getSimpleName());
AssertionUtils.isNull(receiverMap.get(protocolId), "duplicate protocol registration, @PacketReceiver [class:{}] is repeatedly received [at{}]", packetClazz.getSimpleName(), packetClazz.getSimpleName());
var receiverDefinition = new PacketReceiverDefinition(bean, method, packetClazz, attachmentClazz);
var enhanceReceiverDefinition = EnhanceUtils.createPacketReceiver(receiverDefinition);
var receiverField = ReflectionUtils.getFieldByNameInPOJOClass(protocolRegistration.getClass(), "receiver");
ReflectionUtils.makeAccessible(receiverField);
ReflectionUtils.setField(receiverField, protocolRegistration, enhanceReceiverDefinition);
receiverMap.put(protocolId, enhanceReceiverDefinition);
} catch (Throwable t) {
throw new RunException("Registration protocol [class:{}] unknown exception", packetClazz.getSimpleName(), t);
}
@@ -30,6 +30,7 @@ public abstract class SessionUtils {
private static final String CHANNEL_SIMPLE_INFO_TEMPLATE = "[sid:{}][uid:{}]";
private static final String CHANNEL_TEMPLATE = "[channel:{}]";
public static boolean isActive(Session session) {
return session != null && session.getChannel().isActive();
@@ -47,7 +48,7 @@ public abstract class SessionUtils {
public static String sessionInfo(ChannelHandlerContext ctx) {
var session = SessionUtils.getSession(ctx);
if (session == null) {
return StringUtils.format(CHANNEL_INFO_TEMPLATE, ctx.channel());
return StringUtils.format(CHANNEL_TEMPLATE, ctx.channel());
}
return sessionInfo(session);
}
@@ -70,7 +71,7 @@ public abstract class SessionUtils {
public static String sessionSimpleInfo(ChannelHandlerContext ctx) {
var session = SessionUtils.getSession(ctx);
if (session == null) {
return StringUtils.format(CHANNEL_SIMPLE_INFO_TEMPLATE, ctx.channel());
return StringUtils.format(CHANNEL_TEMPLATE, ctx.channel());
}
return sessionSimpleInfo(session);
}
@@ -124,9 +124,6 @@ public abstract class EnhanceUtils {
constructorFiled.setModifiers(Modifier.PRIVATE);
enhanceClazz.addField(constructorFiled);
CtField receiverFiled = new CtField(classPool.get(Object.class.getCanonicalName()), "receiver", enhanceClazz);
receiverFiled.setModifiers(Modifier.PRIVATE);
enhanceClazz.addField(receiverFiled);
// 定义类所包含的所有子协议成员
var allSubProtocolIds = ProtocolAnalysis.getAllSubProtocolIds(protocolId)
@@ -157,11 +154,6 @@ public abstract class EnhanceUtils {
protocolConstructorMethod.setBody("{return this.constructor;}");
enhanceClazz.addMethod(protocolConstructorMethod);
CtMethod receiverMethod = new CtMethod(classPool.get(Object.class.getCanonicalName()), "receiver", null, enhanceClazz);
receiverMethod.setModifiers(Modifier.PUBLIC + Modifier.FINAL);
receiverMethod.setBody("{return this.receiver;}");
enhanceClazz.addMethod(receiverMethod);
CtMethod moduleMethod = new CtMethod(classPool.get(byte.class.getCanonicalName()), "module", null, enhanceClazz);
moduleMethod.setModifiers(Modifier.PUBLIC + Modifier.FINAL);
moduleMethod.setBody("{return " + registration.module() + ";}");
@@ -30,11 +30,6 @@ public interface IProtocolRegistration {
Constructor<?> protocolConstructor();
/**
* 协议接收器,回调方法,主要是存放一些额外的参数
*/
Object receiver();
/**
* 序列化
*/
@@ -37,12 +37,8 @@ public class ProtocolRegistration implements IProtocolRegistration {
private short id;
private byte module;
private Constructor<?> constructor;
private Object receiver;
private Field[] fields;
/**
* 所有的协议里的发送顺序都是按字段名称排序
*/
@@ -67,11 +63,6 @@ public class ProtocolRegistration implements IProtocolRegistration {
return constructor;
}
@Override
public Object receiver() {
return receiver;
}
@Override
public void write(ByteBuf buffer, IPacket packet) {