The MicroProfile community have released the Reactive Streams Operators 1.0 API, a specification that defines a set of operators for Reactive Streams. Inspired by the java.util.stream package, first introduced introduced in JDK 8, and based on the Reactive Streams initiative, the Reactive Streams Operators API allows developers to: create Reactive Streams; process the data transiting in the streams; and accumulate results.

The Reactive Streams initiative, based on the Reactive Manifesto, started in 2013 as a collaboration among Lightbend, Netflix and Pivotal. The initial specification, released in 2015, was ultimately adopted with the release of JDK 9 as the java.util.concurrent.Flow API.

Interfaces included in the Reactive Streams Operators API include: PublisherBuilder ; SubscriberBuilder ; and ProcessorBuilder . It's important to note that due to the 40-50 individual requirements and approximately 100 tests in the TCK, the Reactive Streams Operators interfaces were not intended to be directly implemented by developers. As stated on the website:

The semantics defined by Reactive Streams are very strict, and are non trivial, particularly in the areas of thread safety, to implement correctly. Typically, application developers are expected to use third party libraries that provide the tools necessary to manipulate and control streams. Examples include Akka Streams, RxJava and Reactor.

Other implementations of Reactive Streams include Zero Dependency and SmallRye.

The SmallRye implementation of the Reactive Streams Operators API includes a few examples such as this simple example on how to filter a stream of words:

import org.eclipse.microprofile.reactive.streams.operators.ReactiveStreams; public class QuickStart { public static void main(String[] args) { // create a stream of words ReactiveStreams.of("hello", "from", "smallrye", "reactive", "stream", "operators") .map(String::toUpperCase) // transform the words .filter(s -> s.length() > 4) // filter items .forEach(word -> System.out.println(">> " + word)) // terminal operation .run(); // run it (create the streams, subscribe to it, etc.) } }

James Roper, architect and co-creator of the Lagom microservices framework at Lightbend, spoke to InfoQ about this latest release.

InfoQ: What was the inspiration to create a Reactive Streams Operators API to MicroProfile?

James Roper: Reactive Streams Operators fills a gap between Reactive Streams and users. Reactive Streams offers a way for different asynchronous streaming libraries to integrate with each other, without any prior knowledge of each other; for example, it allows a database library to stream data out to a HTTP server library. This offers a lot of potential to MicroProfile, in that it opens up many possibilities for how reactive streaming can be adopted by MicroProfile specifications. However, the gap is that as an integration SPI, Reactive Streams is not designed to be used or implemented directly by end users, in fact, it's very hard to implement. So for example, if my database library returns a Publisher , and my HTTP server library accepts a Publisher , how do I connect these to each other? Before MicroProfile Reactive Streams Operators, I had two options, one was to implement a Publisher that did the conversion myself, which as I just said, is not how Reactive Streams is designed to be used, and is fraught with possibilities for errors; the other is to bring in a third party library that offers a way to convert publishers of one type to another, via a map function. This is also not desirable; users should be able to use a MicroProfile implementation without bringing in third party libraries for core features. So, we decided that the first thing we should do was create MicroProfile Reactive Streams Operators, to address this gap, before embracing Reactive Streams in any other specs in MicroProfile.

InfoQ: Now that the Reactive Streams Operators API is part of MicroProfile, would it be feasible for a MicroProfile CQRS API? Would there be any challenges in implementing such an API?

Roper: Technically, Reactive Streams Operators is not needed to create a MicroProfile CQRS API; a MicroProfile CQRS API could exist completely independently of Reactive Streams. That said, just as the Reactive Streams, and the MicroProfile Reactive Streams Operators API offers opportunities to integrate different existing opportunities that provide streaming, MicroProfile Reactive Streams Operators offers the potential for a MicroProfile CQRS API to be far more useful, in that it will be, off the bat, able to be integrated with any other library that provides a Reactive Streams implementation. Of even more interest to a MicroProfile CQRS API is the upcoming MicroProfile Reactive Messaging specification. This provides the ability to process streams of messages from different sources to different sinks. A CQRS event stream could well be one of those sources of messages, by leveraging MicroProfile Reactive Messaging, consuming a CQRS event stream, or publishing it to a message broker, would be no different than handling messages from a message broker, so there would be less concepts for a user to learn to use the API.

InfoQ: Do you anticipate other vendors to implement the Reactive Streams Operators API?

Roper: Both RedHat and Lightbend have provided implementations of the API, and I believe IBM has taken, or is taking, the RedHat implementation to provide support for it in OpenLiberty.

InfoQ: What's on the horizon for Reactive Streams Operators?

Roper: I expect it will evolve on an as-needed basis. As I said, it fills a gap, but that gap doesn't actually exist yet, because we don't yet have much support for Reactive Streams in other MicroProfile specs. This is why MicroProfile Reactive Operators is not part of the MicroProfile specification yet. By itself, it's not that interesting. But MicroProfile Reactive Messaging will be the first specification to build on it, and from experience with implementing and using that, additional use cases and requirements for MicroProfile Reactive Streams Operators are likely to be uncovered, which will drive its future evolution. We're also looking at whether it would make sense to propose the inclusion of MicroProfile Reactive Streams Operators in the JDK. Having a standard operators API will be useful far beyond MicroProfile, so it will be interesting to see if it ends up getting there.

Resources