Repository files navigation

This provides a reactive streams implementation for Netty. Essentially it comes in the form of two channel handlers, one that publishes inbound messages received on a channel to aPublisher, and another that writes messages received by aSubscriber outbound.

Features include:

• Full backpressure support, as long as theAUTO_READ channel option is disabled.
• Publishers/subscribers can be dynamically added and removed from the pipeline.
• Multiple publishers/subscribers can be inserted into the pipeline.
• Customisable cancel/complete/failure handling.

This project is released and published through Sonatype. You must have Sonatype credentials installed, preferably in$HOME/.m2/settings.xml and a GPG key that is available to the standard GPG keyservers. If your key is not on the server, you should add it to the keyserver as follows:

gpg --send-key <MY_KEYID> --keyserver pgp.mit.edu

After that, you can perform a release through the following commands:

mvn release:prepare -Darguments=-Dgpg.passphrase=thephrasemvn release:perform -Darguments=-Dgpg.passphrase=thephrase

This will push a release to the staging repository and automatically close and publish the staging repository to production.

Here's an example of creating a client channel that publishes/subscribes toByteBuf's:

EventLoopGroupworkerGroup =newNioEventLoopGroup();Bootstrapbootstrap =newBootstrap()  .group(workerGroup)  .channel(NioSocketChannel.class)  .option(ChannelOption.SO_KEEPALIVE,true);  .handler(newChannelInitializer<SocketChannel>() {@OverridepublicvoidinitChannel(SocketChannelch)throwsException {HandlerPublisher<ByteBuf>publisher =newHandlerPublisher<>(ch.executor(),ByteBuf.class);HandlerSubscriber<ByteBuf>subscriber =newHandlerSubscriber<>(ch.executor());// Here you can subscriber to the publisher, and pass the subscriber to a publisher.ch.pipeline().addLast(publisher,subscriber);    }});

Notice that the channels executor is explicitly passed to the publisher and subscriber constructors.HandlerPublisher andHandlerSubscriber use Netty's happens before guarantees made by itsEventExecutor interface in order to handle asynchronous events from reactive streams, hence they need to have an executor to do this with. This executor must also be the same executor that the handler is registered to use with the channel, so that all channel events are fired from the same context. The publisher/subscriber will throw an exception if the handler is registered with a different executor.

Reactive streams allows implementations to drop messages when a stream is cancelled or failed - Netty reactive streams does this itself. This introduces a problem when those messages hold resources and must be cleaned up, for example, if they areByteBuf's that need to be released. While Netty reactive streams will callReferenceCountUtil.release() on every message that it drops, other implementations that it's interacting with likely won't do this.

For this reason, you should make sure you do one of the following:

• Insert a handler in the pipeline that converts incomingByteBuf's to some non reference counted structure, for examplebyte[], or some high level immutable message structure, and have theHandlerPublisher publish those. Similarly, theHandlerSubscriber should subscribe to some non reference counted structure, and a handler should be placed in the pipeline to convert these structures toByteBuf's.
• Write a wrappingPublisher andSubscriber that synchronously convertsByteBuf's to/from a non reference counted structure.

In addition to raw reactive streams support, thenetty-reactive-streams-http library provides some HTTP model abstractions and a handle for implementing HTTP servers/clients that use reactive streams to stream HTTP message bodies. TheHttpStreamsServerHandler andHttpStreamsClientHandler ensure that everyHttpRequest andHttpResponse in the pipeline is either aFullHttpRequest orFullHttpResponse, for when the body can be worked with in memory, orStreamedHttpRequest orStreamedHttpResponse, which are messages that double asPublisher<HttpContent> for handling the request/response body.