[[Reactive Spring]]

Reactive Spring

by Josh Long (@starbuxman), Spring Developer Advocate, Pivotal

The Missing Metaphor

Reactive programming is an approach to writing software that embraces asynchronous IO. Asynchronous I/O is a small idea that portends big changes for software. The idea is simple: alleviate inefficient resource utilization by reclaiming resources that would otherwise be idle as they waited for I/O activity. Asynchronous IO inverts the normal design IO processing: the clients are notified of new data instead of asking for it; this frees the client to do other things while waiting for new notifications. There is always the risk that too many notifications will overwhelm a client. A client must be able to push back, rejecting work it can’t handle. This is a fundamental aspect of flow control in distributed systems. In reactive programming, the ability of the client to signal how much work it can manage is called backpressure. Many projects - like Vert.x, Akka Streams, and RxJava - support reactive programming. The Spring team has a project called Reactor. There’s common ground across these different approaches extracted into a de-facto standard, the Reactive Streams initiative. The Reactive Streams initiative defines four types:

The Publisher<T> is a producer of values that may eventually arrive. A Publisher<T> produces values of type T.

Example 1. the Reactive Streams Publisher<T>.

package org.reactivestreams;

public interface Publisher<T> {

    void subscribe(Subscriber<? super T> s);
}

The Subscriber subscribes to a Publisher<T>, receiving notifications on any new values of type T.

Example 2. the Reactive Streams Subscriber<T>.

package org.reactivestreams;

public interface Subscriber<T> {

    public void onSubscribe(Subscription s);

    public void onNext(T t);

    public void onError(Throwable t);

    public void onComplete();
}

When a Subscriber<T> subscribes to a Publisher<T>, it results in a Subscription<T>.

Example 3. The Reactive Streams Subscription<T>.

package org.reactivestreams;

public interface Subscription {

    public void request(long n);

    public void cancel();
}

A Publisher<T> that is also a Subscriber<T> is called a Processor<T>.

Example 4. The Reactive Streams Processor<T>.

package org.reactivestreams;

public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {
}

The specification is not meant to be a prescription for the implementations, instead defining types for interoperability. The Reactive Streams types eventually found their way into Java 9 as one to one semantically equivalent interfaces in the java.util.concurrent.Flow class.

Reactor

The Reactive Streams types are not enough; you’ll need higher order implementations to support operators like filtering and transformation. Pivotal’s Reactor project is a good choice here; it builds on top of the Reactive Streams specification. It provides two specializations of the Publisher<T>. The first, Flux<T>, is a Publisher that produces zero or more values. It’s unbounded. The second, Mono<T>, is a Publisher<T> that produces zero or one value. They’re both publishers and you can treat them that way, but they go much further than the Reactive Streams specification. They both provide operators, ways to process a stream of values. Reactor types compose nicely - the output of one thing can be the input to another.

Reactive Spring

As useful as project Reactor is, it’s only a foundation. Our applications need to talk to data sources. They need to produce and consume HTTP, SSE and WebSocket endpoints. They support authentication and authorization. Spring provides these things. If Reactor gives us the missing metaphor, Spring helps us all speak the same language.

Spring Framework 5.0 was released in September 2017. It builds on Reactor and the Reactive Streams specification. It includes a new reactive runtime and component model called Spring WebFlux. Spring WebFlux does not depend on or require the Servlet APIs to work. It ships with adapters that allow it to work on top of a Servlet-engine, if need be, but it’s not required. It also provides a Netty-based web server. Spring Framework 5, which works with a baseline of Java 8 and Java EE 7 - is the foundation for changes in much of the Spring ecosystem.

A Bootiful Application

Let’s look at an example. We’ll build a simple Spring Boot 2.0 application that, oh, I don’t know…​ How about we build a service to manage books? We could call the project Library or something like that. Go to the Spring Initializr. Make sure that some version of Spring Boot 2.0 (or later) is selected in the version drop down menu. We’re writing a service to manage access to books in the library, so give this project the artifact ID library-service. Select Reactive Web, Actuator, Reactive MongoDB, Reactive Security, and Lombok. I chose to use the Kotlin language, even if most of the project we’ll build is in Java. You can keep Java artifacts collocated in a Kotlin project. Click Generate and it’ll download an archive; unzip it and open it in your favorite IDE that supports Java 8 (or later), Kotlin (optionally) and Maven. While we could’ve chosen Gradle at the Spring Initializr, I’ve chosen Maven for the purposes of this article.

Our stock standard Spring Boot application has an entry class that looks like this.

Example 5. the empty husk of a new Spring Boot project.

link:./library-service/src/main/kotlin/com/example/libraryservice/LibraryServiceApplication.java[role=include]

Data Access with Reactive Spring Data modules

Spring Data Kay is the largest update to Spring Data since its inception. This release debuts support, where supported in the underlying data stores (MongoDB, Cassandra, Redis and Couchbase), for reactive data access. It introduces new reactive repository and template implementations. We’ve got the reactive MongoDB driver and Spring Data module on the classpath so let’s use them to manage some data. Create a new entity called Book.

Example 6. a MongoDB @Document entity, Book

link:./library-service/src/main/kotlin/com/example/libraryservice/Book.java[role=include]

Create a Spring Data repository to support the data management lifecycle of the entity. This should look very familiar for anybody who’s ever used Spring Data, except that the repository supports reactive interactions: methods return Publisher types, and input can be given as Publisher<T> instances.

Example 7. A reacive Spring Data MongoDB repository

link:./library-service/src/main/kotlin/com/example/libraryservice/BookRepository.java[role=include]

Install Some Sample Data with an ApplicationRunner

With that we have enough to install some sample data (just for our demo). Spring Boot invokes the #run(ApplicationArguments) method when the application has started, passing in wrappers for the arguments (String [] args) into the application. Let’s create an ApplicationRunner that deletes all the data in the data source, then emits a few book titles, then maps them to Book entities, and then persists those books; then queries all the records in the data source and then prints out everything.

Example 8. an ApplicationRunner to write data

link:./library-service/src/main/kotlin/com/example/libraryservice/SampleBookInitializer.java[role=include]

The example looks at the titles of various books, and one of (the possibly numerous) book’s authors, and writes them to the database. First the strings are split by the | delimiter. Then the title and book author are used to create a Book. Then the records are saved to the data source, MongoDB. The result of the save operation is a Mono<Book>. Something needs to subscribe to each of those resulting Publisher<T>`s, so we use the `flatMap operator. Then, we switch tracks, turning focusing on the results of finding all records and then logging them for inspection.

This code defines a pipeline; each operator defines a stage in a pipeline. The pipeline is not hot, or eager. We need to activate the pipeline by subscribing to it. The Publisher<T> only defines one type of subscription, but Reactor defines a few overloads that provide the same lifecycle hooks as overriding individual methods in a fully formed Subscriber<T> would. There are hooks to process each emitted value, as well any exceptions thrown, among other things.

Reactive and in a Rush? Use a Scheduler!

Were you to put a breakpoint in any of the lambdas in the previous code listing and then inspect Thread.currentThread().getName(), you’d see that the the thread on which processing is running is different than the main thread (which is named main). Reactor defers to a Scheduler implementation for its processing. You can specify the default global Scheduler you’d like to use by calling Schedulers.setFactory(Factory). You can specify on which thread a particular Publisher<T> should run when it subscribes by specifying Mono<T>#subscribeOn(Scheduler) or Flux<T>#subscribeOn(Scheduler).

Spring WebFlux

Now that we’ve got data in the data source, let’s stand up a REST API. We’ll use Spring WebFlux, a brand new reactive web runtime and component model. Spring WebFlux does not depend on the Servlet specification. It can work independently, with a Netty-based web server. It is designed, from the bottom up, to work with Publisher<T> instances.

REST with Spring WebFlux

We can use Spring MVC style controllers, like this:

Example 9. a Spring MVC style REST API.

link:./library-service/src/main/kotlin/com/example/libraryservice/BookRestController.java[role=include]

Note

Spring has the concept of profiles. Profiles are labels, or tags, essentially, for Spring beans. Beans in a given profile don’t exist unless that profile is activated. The easiest way to activate a profile is to use a command line argument when running the java command. If you wanted to activate all the beans under the profile1 and profile2 profiles, you’d use an incantation like this: java -Dspring.profiles.active=profile1,profile2 -jar ... The benefit of the profile, in this case, is that we can have the same HTTP endpoints implemented three different ways in the same codebase and only activate one at a time.

This controller should look familiar to anyone who’s ever used Spring MVC. It may be familiar, but it is not Spring MVC. We’re using a new reactive runtime called Spring WebFlux. The annotations are the same, but the rules are sometimes different.

Functional Reactive Endpoints

The last code listing demonstrates a controller. Spring Web Flux controllers define endpoint handlers and endpoint mappings through declarative annotations. The annotations describe how the routing for a given endpoints is to be handled. They’re sophisticated, but ultimately limited to whatever the framework itself can do with those annotations. If you want more flexible request matching capabilities, you can use Spring WebFlux functional reactive endpoints.

Example 10. The same endpoints reworked as funtional reactive endpoints in Java. Run this using the frp-java profile.

link:./library-service/src/main/kotlin/com/example/libraryservice/BookRestConfigurationJava.java[role=include]

The functional reactive style lets you express HTTP endpoints as request predicates mapped to a handler class. The handler class implementation is easily expressed as concise Java lambdas. You can use the default request predicates or provide your to gain full control over how requests are matched and dispatched.

We produce a result and pass it to the body(Publisher<T>) method, along with a class literal. We need the class literal to help the engine figure out what type of message framing it should do. Remember, that Publisher<T> might produce a TRILLION records - it may never stop! The producer can’t afford to wait until all records have been produced and then marshal the record from an object to JSON. So, it marhsals each record as soon as it’s got it. We need to tell it what kind of message to look out for. In the Spring MVC style controllers, the return value (a Publisher<T>) in the handler methods encodes its generic parameter, T, and the engine can retrieve that generic parameter using reflection. The engine can not do the same thing for the instance variable passed into the body method as a parameter since there’s no easy way to retrieve the generic signature of instance variables. We call this limitation type erasure. The type literal gets us past this restriction.

If you’re using the Kotlin language, things are even more concise thanks to a Kotlin-language DSL that also ships as part of Spring Framework 5. The Kotlin DSL requires less code and also supports retrieving the generic parameter thanks to runtime reification of inline methods. Here are the same endpoints reimplemented using the Kotlin-language DSL:

Example 11. Here are the same endpoints using the Kotlin-language DSL

link:./library-service/src/main/kotlin/com/example/libraryservice/BookRestConfigurationKotlin.kt[role=include]

Spring Security

I know what you’re thinking, but its still not quite ready for production. We need to address security. Spring Security supports a rich set of integrations with all manner of identity providers. It supports authentication by propagating a security context so that application level code - method invocations, HTTP requests, etc. - have easy access to the context. The context has historically been implemented with a ThreadLocal. Thread local state doesn’t make a lot of sense in a reactive world. Reactor provides a Context object, which acts as a sort of dictionary. Spring Security 5.0’s reactive support propagates its security context using this mechanism. Parallel, reactive type hierarchies have been introduced to support non-blocking authentication and authorization. You don’t have to worry about much of this nuance. I just think it’s dope. All we need to know is that in the reactive world, authentication is handled by an object of type ReactiveAuthenticationManager which has a simple job: given an Authentication attempt, return a Mono<Authentication> (indicating whether the authentication attempt succeeded) or throw an Exception.

One implementation of the ReactiveAuthenticationManager supports delegating to a user-provided object of type MapReactiveUserDetailsService. The MapReactiveUserDetailsService connects your custom username and password store to Spring Security’s authentication. You might have a database table called USERS or just a hardcoded Map<K,V> of users. By default, Spring Security locks down the whole application and installs HTTP BASIC authentication. Any attempt at calling any endpoint will fail unless we provide credentials. By default all authenticated principals can access all endpoints. Let’s establish introduce a few users of various roles. All users will have the USER role, but only a privileged few will have the ADMIN role. In this newly secured world, let’s say that all users will be able to view the books they’ve written, but only those with the ADMIN role will be able to see all the books. (Let’s ignore for now whether this domain makes any sense!)

Example 12. The Spring Security configuration

link:./library-service/src/main/kotlin/com/example/libraryservice/SecurityConfiguration.java[role=include]

Let’s try making an HTTP BASIC authenticated call to the service. I’ll make the first request as jlong, a regular USER.

Example 13. curl the endpoint as jlong

curl -ujlong:pw http://localhost:8080/books/jlong

It won’t work if I try to access http://localhost:8080/books/rwinch, though. Only ADMIN role users can access other endpoints.

Example 14. curl the endpoint as rwinch

curl -urwinch:pw http://localhost:8080/books

Deployment

Our application is secure and observable. Now we can deploy it. This is a natural thing to run in a cloud provider like Cloud Foundry, an open-source Apache 2 licensed cloud platform that’s optomized for the continuous management of applications. It sits at a level (or two) above cloud infrastructure. It is infrastructure agnostic, running on local cloud providers like OpenStack and vSphere or on public cloud providers like Amazon Web Services, Google Cloud, Microsoft Azure and, yes, [Oracle Cloud](https://blogs.oracle.com/developers/cloud-foundry-arrives-on-oracle-cloud)! No matter where Cloud Foundry is installed, it’s use is basically the same. You authenticate and then tell the platform about your application workload using the cf CLI and the cf push command.

Example 15. Using cf CLI to push the application.

cf login -a $CF_API_ENDPOINT -u $CF_USER -s $CF_SPACE -o $CF_ORG

cf push -p library-service-0.0.1-SNAPSHOT.jar java-magazine-library-service

Once the application is up and running you can access its public HTTP endpoints. You can provision backing services - message queues, databases, caches, etc. - using cf create-service. You can scale the application up to multiple load-balanced instances using cf scale. You can interrogate the application’s metrics, its Spring Boot Actuator endpoints, its health and so much more, all from the Pivotal Apps Manager dashboard. The application is up and running and our clients can talk to it in a secure fashion.

"Wait, what client?," I hear you saying…​

A (Reactive) Client

We’ve stood up a REST API. We need to connect a client to the service. While we could use the Spring Framework RestTemplate, the general workhorse HTTP client that has served us well for the better part of a decade, it’s not particularly suited to potentially unlimited streams of data. It expects to be able to convert all response payloads into something by waiting for the end of the response. This isn’t going to work if the client is using server sent events, or even just a really large JSON response. Instead, let’s use the new Spring WebFlux WebClient.

Example 16. Configuring and using an authenticated WebClient

link:./library-client/src/main/java/com/example/libraryclient/LibraryClientApplication.java[role=include]

We configure the WebClient and pre-configure a baseUrl as well as an ExchangeFilterFunction that authenticates the client with the service. The WebClient gives us a Publisher<T> for the response, which we then print the console. In this case, it doesn’t really matter; we’ve only got four records in our endpoint! The thing that’s so valuable is that this code would work even if we were consuming a trillion records. It would work if we were consuming server-sent events that never end.

Subscriber#onNext

What’s next? In this article we’ve looked briefly at building a web service with Spring Boot. We looked at Reactor, Spring Data Kay, Spring Framework 5 and Spring WebFlux, Spring Security 5, and Spring Boot 2.0. Spring Boot 2 makes it easy to assemble the various reactive Spring projects into an application. We didn’t look at the Spring Boot Actuator, but it surfaces operational data like metrics, application health and more. It has also been updated to work seamlessly in a reactive world.

Spring Boot 2 sets the stage for Spring Cloud Finchley. Spring Cloud Finchley builds on Spring Boot 2.0 and updates a number of different APIs, where appropriate, to support reactive programming. Service registration and discovery works in Spring WebFlux based applications. Spring Cloud Commons supports client-side load-balancing across services registered in a service registry (like Apache Zookeepere, Hashicorp Consul and Netflix Eureka) for the Spring Framework WebClient. Spring Cloud Netlix Hystrix circuit breakers have always worked naturally with RxJava, which in turn can interop with Publisher<T> instances. This is even easier. Spring Cloud Stream supports working with Publisher<T> instances to describe how messages arrive and are sent to messaging subsraits like RabbitMQ, Apache Kafka or Redis. Spring Cloud Gateway is a brand new reactive API gateway project that supports HTTP and websocket request proxying, rewriting, load-balancing, circuit breaking, rate limiting and so much more. Spring Cloud Sleuth has been updated to support distributed tracing across reactive boundaries. The list goes on and on.

The Future<Spring> is a reactive Publisher<Spring>. Begin your journey building production-worthy, agile and reactive applications and services with Spring Boot at the Spring Initializr. If you have questions, find me on Twitter @starbuxman or email (josh@joshlong.com).