Reactor Netty TcpServer with Pipeline

Question

The way I would have put together a Netty Tcp Server before reactor netty was to create the server bootsrap and add my custom pipeline class. With Reactor-Netty there is the TcpServer.create(), but seems that I have to create a new functional interface that takes NettyInbound and NettyOutbound and returns a Mono. However if I want to add a ChannelInitializer that builds my pipeline, I have to block to get the NettyContext. The incoming message is received by the functional interface and I can send a response, but nothing go through my pipeline.

Is there a way to make us of Reactor Netty and have the message flow through a customized pipeline?

Returning the Mono.just("Hi") with neverComplete() successfully sends 'Hi' to the client when a connection is made and when a message is received, but I need to rather offload this to the pipeline and then have the result feed back to the client.

public void startServer() throws InterruptedException{
    EventLoopGroup group = new NioEventLoopGroup(1);
    try {
        final TcpServer server = TcpServer.create(opts -> opts
            .eventLoopGroup(group)
            .listen(tcpSocketAddress));
        server
            .newHandler((in, out) -> {
                in.receive()
                    .take(1)
                    .log(ApolloApplicationTests.class.getName())
                    .subscribe(data -> {
                        log.info("Server Received: {}", data.toString(CharsetUtil.UTF_8));
                       latch.countDown();
                    });
                    return out.sendString(Mono.just("Hi")).neverComplete();
            })
            .block().addHandler(clientEndPoint)
            .channel()
            .closeFuture().sync();

    } finally {
        group.shutdownGracefully().sync();
    }
}



import java.util.List;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Configurable;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.MessageToMessageDecoder;
import reactor.util.Logger;
import reactor.util.Loggers;

@Configurable
@Component
public class ClientEndPoint extends ChannelInitializer<Channel> {

final Logger log = Loggers.getLogger(ApolloApplication.class);

private ChannelPipeline pipeline;

@Autowired
private ChannelHandlerAdapter messageInterchange;

@Autowired
private LengthFieldBasedFrameDecoder lowOrderVliDecoder;

@Autowired
private MessageToMessageDecoder<ByteBuf> messageDecoder;

@Autowired
private LengthFieldPrepender vliEncoder;

@Autowired
@Qualifier("inBound")
List<ChannelHandler> inBoundHandlers;

@Autowired
@Qualifier("outBound")
List<ChannelHandler> outBoundHandlers;

    @Override
    protected void initChannel(Channel sc) throws Exception {
        this.pipeline = sc.pipeline();
        this.pipeline.addLast("lowOrderVliDecoder", this.lowOrderVliDecoder);

        this.pipeline.addLast("messageDecoder", this.messageDecoder);

        this.pipeline.addLast("vliEncoder", this.vliEncoder);

        for (ChannelHandler handler : this.inBoundHandlers) {
            this.pipeline.addLast(handler);
        }

        this.pipeline.addLast("messageInterchange", this.messageInterchange);

        for (ChannelHandler handler : this.outBoundHandlers) {
            this.pipeline.addLast(handler);
        }
    }


    public void accept(Channel sc) {
        this.pipeline = sc.pipeline();
        this.pipeline.addLast("lowOrderVliDecoder", this.lowOrderVliDecoder);

        this.pipeline.addLast("messageDecoder", this.messageDecoder);

        this.pipeline.addLast("vliEncoder", this.vliEncoder);

        for (ChannelHandler handler : this.inBoundHandlers) {
            this.pipeline.addLast(handler);
        }

        this.pipeline.addLast("messageInterchange", this.messageInterchange);

        for (ChannelHandler handler : this.outBoundHandlers) {
            this.pipeline.addLast(handler);
        }
    }
}

Answer 1

So this I figured out

public Mono<? extends NettyContext> initializeServer() throws InterruptedException {
    this.log.debug("Server Initializing");
    BiFunction<? super NettyInbound, ? super NettyOutbound, ? extends Publisher<Void>> serverHandler = (in,
            out) -> {
        in.receive().asString().subscribe(data -> {
            this.log.debug("Received " + data + " on " + in);
        });
        return Flux.never();
    };

    TcpServer server = TcpServer.create(opts -> opts.afterChannelInit(pipeline).listen(tcpSocketAddress));
    return server.newHandler(serverHandler);
}

where pipeline is the class that implements Consumer and builds the pipeline in the accept method as a typical netty pipeline.

Then I start the server

private void startServer(Mono<? extends NettyContext> connected) {
        ChannelFuture f = connected.block(Duration.ofSeconds(5)).channel()
                .closeFuture();
        final CountDownLatch channelLatch = new CountDownLatch(1);
        f.addListener(new ChannelFutureListener() {

            @Override
            public void operationComplete(ChannelFuture cf) throws Exception {
                log.debug("Channel Disconnected");

            }

        });
        f.awaitUninterruptibly();
        // Now we are sure the future is completed.
        assert f.isDone();

        if (f.isCancelled()) {
            this.log.warn("Connection Cancelled");
        } else if (!f.isSuccess()) {
            if (f.cause() != null) {
                f.cause().printStackTrace();
            } else {
                this.log.warn("Connection not successful");
            }
        } else {
            channelLatch.countDown();
            this.log.info("Server Start Successful");
        }
        try {
            channelLatch.await();
        } catch (InterruptedException ex) {
            throw new CancellationException("Interrupted while waiting for streaming " + "connection to arrive.");
        }
}