простая реализация EventBus

protocol/src/main/java/mc/protocol/NettyServer.java

@@ -5,12 +5,15 @@ import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import mc.protocol.di.DaggerProtocolComponent;
 import mc.protocol.di.ProtocolComponent;
+import mc.protocol.packets.ClientSidePacket;
+import mc.protocol.utils.EventBus;
 
 @Slf4j
 @RequiredArgsConstructor
 public class NettyServer {
 
 	private final ServerBootstrap serverBootstrap;
+	private final EventBus eventBus;
 
 	public void bind(String host, int port) {
 		log.info("Network starting: {}:{}", host, port);
@@ -24,6 +27,10 @@ public class NettyServer {
 		}
 	}
 
+	public <P extends ClientSidePacket> void listenPacket(State state, Class<P> packetClass, EventBus.EventHandler<P> eventHandler) {
+		this.eventBus.subscribe(state, packetClass, eventHandler);
+	}
+
 	public static NettyServer createServer() {
 		ProtocolComponent component = DaggerProtocolComponent.create();
 		return component.getNettyServer();

protocol/src/main/java/mc/protocol/PacketInboundHandler.java

@@ -4,19 +4,16 @@ import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.SimpleChannelInboundHandler;
 import lombok.RequiredArgsConstructor;
 import mc.protocol.packets.ClientSidePacket;
-import reactor.core.publisher.Sinks;
+import mc.protocol.utils.EventBus;
 
 @RequiredArgsConstructor
 public class PacketInboundHandler extends SimpleChannelInboundHandler<ClientSidePacket> {
 
-	@SuppressWarnings("rawtypes")
+	private final EventBus eventBus;
+
 	@Override
 	protected void channelRead0(ChannelHandlerContext ctx, ClientSidePacket packet) {
-		Sinks.Many<ChannelContext> packetSinks = ctx.channel().attr(NetworkAttributes.STATE)
+		State state = ctx.channel().attr(NetworkAttributes.STATE).get();
-				.get().getPacketSinks(packet.getClass());
+		eventBus.emit(state, new ChannelContext<>(ctx, packet));
-
-		if (packetSinks != null) {
-			packetSinks.tryEmitNext(new ChannelContext<>(ctx, packet));
-		}
 	}
 }

protocol/src/main/java/mc/protocol/di/ProtocolModule.java

@@ -16,6 +16,8 @@ import mc.protocol.PacketInboundHandler;
 import mc.protocol.io.codec.ProtocolDecoder;
 import mc.protocol.io.codec.ProtocolEncoder;
 import mc.protocol.io.codec.ProtocolSplitter;
+import mc.protocol.utils.EventBus;
+import mc.protocol.utils.SimpleEventBus;
 
 import javax.annotation.Nonnull;
 import javax.inject.Provider;
@@ -26,8 +28,8 @@ import java.util.Map;
 public class ProtocolModule {
 
 	@Provides
-	NettyServer provideServer(ServerBootstrap serverBootstrap) {
+	NettyServer provideServer(ServerBootstrap serverBootstrap, EventBus eventBus) {
-		return new NettyServer(serverBootstrap);
+		return new NettyServer(serverBootstrap, eventBus);
 	}
 
 	@Provides
@@ -55,15 +57,21 @@ public class ProtocolModule {
 	}
 
 	@Provides
-	Map<String, ChannelHandler> provideChannelHandlerMap() {
+	Map<String, ChannelHandler> provideChannelHandlerMap(EventBus eventBus) {
 		Map<String, ChannelHandler> map = new LinkedHashMap<>();
 
 		map.put("packet_splitter", new ProtocolSplitter());
 		map.put("logger", new LoggingHandler(LogLevel.DEBUG));
 		map.put("packet_decoder", new ProtocolDecoder(true));
 		map.put("packet_encoder", new ProtocolEncoder());
-		map.put("packet_handler", new PacketInboundHandler());
+		map.put("packet_handler", new PacketInboundHandler(eventBus));
 
 		return map;
 	}
+
+	@Provides
+	@ServerScope
+	EventBus provideEventBus() {
+		return new SimpleEventBus();
+	}
 }

protocol/src/main/java/mc/protocol/utils/SimpleEventBus.java

@@ -0,0 +1,25 @@
+package mc.protocol.utils;
+
+import mc.protocol.ChannelContext;
+import mc.protocol.State;
+import mc.protocol.packets.ClientSidePacket;
+
+@SuppressWarnings({ "rawtypes", "unchecked" })
+public class SimpleEventBus implements EventBus {
+
+	private final Table<State, Class<? extends ClientSidePacket>, EventHandler> table = new Table<>();
+
+	@Override
+	public <P extends ClientSidePacket> void subscribe(State state, Class<P> packetClass, EventHandler<P> eventHandler) {
+		table.put(state, packetClass, eventHandler);
+	}
+
+	@Override
+	public <P extends ClientSidePacket> void emit(State state, ChannelContext<P> channelContext) {
+		EventHandler eventHandler = table.getColumnAndRow(state, channelContext.getPacket().getClass());
+
+		if (eventHandler != null) {
+			eventHandler.handle(channelContext);
+		}
+	}
+}

protocol/src/main/java/mc/protocol/utils/Table.java

@@ -0,0 +1,23 @@
+package mc.protocol.utils;
+
+import javax.annotation.Nullable;
+import java.util.HashMap;
+import java.util.Map;
+
+public class Table<C, R, V> {
+
+	private final Map<C, Map<R, V>> map = new HashMap<>();
+
+	@Nullable
+	public V getColumnAndRow(C column, R row) {
+		if (!map.containsKey(column)) {
+			return null;
+		}
+
+		return map.get(column).get(row);
+	}
+
+	public void put(C column, R row, V value) {
+		map.computeIfAbsent(column, c -> new HashMap<>()).put(row, value);
+	}
+}

server/src/main/java/mc/server/Main.java

@@ -49,11 +49,11 @@ public class Main {
 		NettyServer server = NettyServer.createServer();
 		PacketHandler packetHandler = serverComponent.getPacketHandler();
 
-		State.HANDSHAKING.packetFlux(HandshakePacket.class).subscribe(packetHandler::onHandshake);
+		server.listenPacket(State.HANDSHAKING, HandshakePacket.class, packetHandler::onHandshake);
-		State.STATUS.packetFlux(PingPacket.class).subscribe(packetHandler::onKeepAlive);
+		server.listenPacket(State.STATUS, PingPacket.class, packetHandler::onKeepAlive);
-		State.STATUS.packetFlux(StatusServerRequestPacket.class).subscribe(packetHandler::onServerStatus);
+		server.listenPacket(State.STATUS, StatusServerRequestPacket.class, packetHandler::onServerStatus);
-		State.LOGIN.packetFlux(LoginStartPacket.class).subscribe(packetHandler::onLoginStart);
+		server.listenPacket(State.LOGIN, LoginStartPacket.class, packetHandler::onLoginStart);
-		State.PLAY.packetFlux(PingPacket.class).subscribe(packetHandler::onKeepAlivePlay);
+		server.listenPacket(State.PLAY, PingPacket.class, packetHandler::onKeepAlivePlay);
 
 		server.bind(config.server().host(), config.server().port());
 	}