Reactor Netty: HTTP server example
Tags: java, reactiveprogramming, projectreactor, reactornetty
Published: 2024-04-07 15:02:59

This example is to demonstrate

- build server which serve HTTP request
- read/write HTTP header information
- share ByteBuf instance between HTTP connection

Following library is used

- netty 4.1.107.Final
- Project Reactor 3.6.3
- Reactor Netty 1.1.16

# Code

```java
package example;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.HttpHeaderValues;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;
import reactor.netty.DisposableServer;
import reactor.netty.http.server.HttpServer;

import java.nio.charset.Charset;
import java.time.Duration;
import java.util.Scanner;

public class HttpServer {

    private static final Logger LOGGER = LoggerFactory.getLogger(HttpServer.class);

    public static void main(String[] args) {
        Scanner scanner = new Scanner(System.in);

        // (1)
        ByteBuf buf1 = Unpooled.buffer(8);
        buf1.writeCharSequence("Hello ", Charset.defaultCharset());
        ByteBuf buf2 = Unpooled.buffer(8);
        buf2.writeCharSequence("World!", Charset.defaultCharset());

        DisposableServer server =
                HttpServer.create()
                        .host("localhost").port(8080)
                        .wiretap(true)
                        .doOnChannelInit((observer, channel, remoteAddress) -> {
                            // (2)
                            LOGGER.info("pipeline={}", channel.pipeline());
                        })
                        .route(routes ->
                                routes.get("/", (request, response) -> {    // (3)
                                    // (4)
                                    LOGGER.info("Header: host={}", request.requestHeaders().get("Host"));

                                    // (5)
                                    response.addHeader(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.TEXT_PLAIN);
                                    // (6)
                                    return response.send(
                                            Flux.defer( // (7)
                                                    () -> Flux.just(buf1.duplicate(), buf2.duplicate()) // (8)
                                                            .delaySequence(Duration.ofSeconds(1))   // (9)
                                                            .doOnNext(ByteBuf::retain)  // (10)
                                            )
                                    );
                                })
                        )
                        .bindNow();

        // (11)
        server.onDispose().subscribe(
                n -> {
                },
                e -> {
                },
                () -> { // (12)
                    buf1.release();
                    buf2.release();
                    LOGGER.info("Shutdown successfully");
                }
        );

        // (13)
        LOGGER.info("*** Double enter to shutdown ***");
        scanner.nextLine();
        server.dispose();
    }

}
```

# Explanation

(1): create *io.netty.buffer.ByteBuf* instances and write contents into it.

Data should be written into *ByteBuf* before reading. **And can read once only**.

There are reference counting facilities in *ByteBuf* (with *io.netty.util.ReferenceCounted* interface). When an instance is created, refCnt() is 1. retain(int) is for setting times this instance is referenced. After using this, calling release() to decrease the refCnt. When refCnt becomes 0, the system knows that this instance is not used anymore and it will be released.

(2): doOnChannelInit() accepts reactor.netty.ChannelPipelineConfigurer, for configuring the channel pipeline while initializing the channel.

The pipeline contains

```
pipeline=DefaultChannelPipeline{(reactor.left.loggingHandler = reactor.netty.transport.logging.ReactorNettyLoggingHandler), (TransportConfig$TransportChannelInitializer#0 = reactor.netty.transport.TransportConfig$TransportChannelInitializer), (reactor.left.httpCodec = io.netty.handler.codec.http.HttpServerCodec), (reactor.left.httpTrafficHandler = reactor.netty.http.server.HttpTrafficHandler), (reactor.right.reactiveBridge = reactor.netty.channel.ChannelOperationsHandler)}
```

*reactor.netty.transport.TransportConfig$TransportChannelInitializer* is the child class of *io.netty.channel.ChannelInitializer*. It is for setting up pipelines in the "registered" phase of the *Channel* and will be removed by itself after setup.

*io.netty.handler.codec.http.HttpServerCodec* and *reactor.netty.http.server.HttpTrafficHandler* instances is added and as usual, *reactor.netty.channel.ChannelOperationsHandler* must at the end of the pipeline.

(3): *reactor.netty.http.server.HttpServerRoutes#get()* set up a handler when HTTP GET requests on a defined path. request is *reactor.netty.http.server.HttpServerRequest* and response is *reactor.netty.http.server.HttpServerResponse*. Both of them are *reactor.netty.http.server.HttpServerOperations* instances.

The return instance is *Publisher<Void>>*, and only the termination signal of it means to close *Channel* connection.

The actual function processing HTTP GET requests is *reactor.netty.http.server.HttpServer.HttpServerHandle#onStateChange*.

```java
@Override
@SuppressWarnings("FutureReturnValueIgnored")
public void onStateChange(Connection connection, State newState) {
	if (newState == HttpServerState.REQUEST_RECEIVED) {
		try {
			if (log.isDebugEnabled()) {
				log.debug(format(connection.channel(), "Handler is being applied: {}"), handler);
			}
			HttpServerOperations ops = (HttpServerOperations) connection;
			Publisher<Void> publisher = handler.apply(ops, ops);	// <- HERE handler is called
			Mono<Void> mono = Mono.deferContextual(ctx -> {
				ops.currentContext = Context.of(ctx);
				return Mono.fromDirect(publisher);
			});
			if (ops.mapHandle != null) {
				mono = ops.mapHandle.apply(mono, connection);
			}
			mono.subscribe(ops.disposeSubscriber());	// <- subscribe the termination signal
		}
		catch (Throwable t) {
			log.error(format(connection.channel(), ""), t);
			//"FutureReturnValueIgnored" this is deliberate
			connection.channel()
			          .close();
		}
	}
}
```

(4): reading HTTP request header.

(5): setting HTTP response header.

(6): define response's content publisher.

(7): to instruct the following Flux.just() should instantiate objects in subscription phase.

(8): to send two *ByteBuf* out. Please note that duplicate() is called on each *ByteBuf*. A wrapper class *io.netty.buffer.DuplicatedByteBuf* and it copies readerIndex and writerIndex so that the readerIndex of original *ByteBuf* will not be changed when read operation is on this "copied" *ByteBuf*.

If not include this statement, no content in HTTP response after serving several requests.

(9): add 1 second delay between sending two *ByteBuf*

(10): to increase the refCnt of *ByteBuf* by 1. This will be decreased when content of *ByteBuf* is sent out in the lower layer. If not include this statement, the following exception is thrown after serving several requests.

```
15:48:01.665 [reactor-http-nio-6] ERROR r.n.http.server.HttpServerOperations -- [335257bc-1, L:/127.0.0.1:8080 - R:/127.0.0.1:61531] Error finishing response. Closing connection
io.netty.util.IllegalReferenceCountException: refCnt: 0
```

(11): to define handler after server shutdown.

(12): to release two *ByteBuf*.

(13): main thread will be blocked on scanner.nextLine(), and return after pressing 2 times of Enter on console.