TCP server with Reactor Netty and Spring Boot

I'm very new to reactive programing. I have watch mutiple tutorials on reactive programing, reactive spring and project reactor. I understand the concepts and 4 interfaces of the reactive stream specification. However searching around the internet there are not enough examples that can help me wrap my head around applying all of the theory and concepts I learnt into practices.

I'm trying to create a backend TCP server for a game with spring boot and reactor netty. My server application will potentially have multiple TCP server that listening on multiple port. I should probably use reactor kafka and have my servers just talk to each other thru kafka. However, at this stage I don't want to deal with all that yet because it is will just make my learning curve steeper.

This is what I'm doing to create and start the TCP server. It is a class that have a function that I would be able to pass in the port to spin up different TCP servers listening on different port:

public void accept(String host, int port) {
        TcpServer server =
                TcpServer.create()
                        .port(port)
                        .option(ChannelOption.SO_BACKLOG, 1024)
                        .handle(acceptorObserver::onChannelMessage)
                        .doOnChannelInit(acceptorObserver::onChannelInit)
                        .doOnConnection(acceptorObserver::onClientConnect)
                        .doOnBound(acceptorObserver::onServerBound)
                        .doOnUnbound(acceptorObserver::onServerUnbound);
        server.bindUntilJavaShutdown(Duration.ofSeconds(30), null);
    }

my acceptorObserver is a class that just have all the function to handle the lifecycle of the TCP server created. The only method I'm interested in is the handle method where all inbound packets would reach here after all Netty pipelines.

public Publisher<Void> onChannelMessage(NettyInbound nettyInbound, NettyOutbound nettyOutbound) {
        Mono<IPacketWriter> handshake =
                Mono.create(sink ->
                    nettyOutbound.withConnection(conn -> {
                        Channel channel = conn.channel();
                        if (!channel.hasAttr(NettyAttributes.SOCKET)) {
                            logger.info("Client connected from {}", channel.remoteAddress().toString());
                            conn.onDispose().subscribe(null, null, () -> onClientDisconnected(conn));
                            NettySocket newSocket = new NettySocket(conn, (int) (Math.random() * Integer.MAX_VALUE), (int) (Math.random() * Integer.MAX_VALUE), version, patch, locale);

                            IPacketWriter handshakePacket = new PacketWriter().writeShort(version)
                                                                        .writeString(patch)
                                                                        .writeInt(newSocket.getIntSeqRecv())
                                                                        .writeInt(newSocket.getIntSeqSend())
                                                                        .writeByte(locale);

                            sink.success(handshakePacket);

                            channel.attr(NettyAttributes.SOCKET).set(newSocket);
                            sockets.put(newSocket.getId(), newSocket);
                        }
                        else {
                            logger.info("Received message from client {}", channel.remoteAddress().toString());
                            sink.success();
                        }
                    }));

        return nettyOutbound.sendObject(handshake)
                .then(nettyInbound.receiveObject().cast(PacketReader.class).mapNotNull(in -> {
                    logger.info("nettyInbound {}", BitOperator.readableByteArray(in.getBuffer()));
                    return in;
                }).then());
    }

Thanks Violeta she helped me with the above function. So everytime I received an inbound message I will check if the SocketChannel of Netty has my attributes if it doesnt that means it is a newly connected client and I would send my handshake package other wise the handshake Mono will be empty.

My goal is to aggregate all of the NettyInbound received objects into a stream on a different class something like a CentralInboundPacketDispatcher class and I would have different handlers subscribe to this stream and filter out the packets that they care about and handle that. Whenever a new client connected there would be a new NettyInbound stream. My problem is I have no idea how to achieved the above goal. Through my googling It seems that I would need to have an "infinite stream" aka hot stream? There are really not alot of example around or I'm very bad at googling. I would appreciated it if someone could point me to the right direction.

It would be very cool if I could do something like HTTP REST server with Spring Boot where I can use anotation like @GetMapping to tell it if the packet have this header use this function.