Developing reactive microservice using Springboot 2

Reactive Spring is based on the project reactor for building non-blocking applications using the spring platform and spring cloud frameworks. Three important interfaces available in Reactor are Publisher, Subscriber, and Processor.

• Publisher – source of the data
• Subscriber – One which receives data asynchronously
• Processor – nothing but a publisher which is also a subscriber (most of the time we won’t need this)

Reactor introduces reactive types which implement the Publisher interface namely Flux and Mono.

• Flux – represents multiple sequences of a result 0…N (many items) which suggests Flux is a standard publisher
• Mono – As the name suggests, represents single sequence of a result either an empty result or one result 0…1 which suggests Mono will emit at most one item

Starting from Spring 5 and Spring Boot 2 reactive core is completely supported and provides an excellent long-awaited alternative to Akka Streams. If you are coming from a spring background and uses it extensively there is no reason why you shouldn’t try this. However, it is great for people who begin with Spring platform as well.

This post is hugely inspired by Josh Long’s webinar on Reactive streams which you can find here. We are going to use Kotlin for our demo and sample project.

To begin with visit start.spring.io or use the spring initializer in STS or Intellij. Select the Springboot version 2.0.0 snapshot and above and the language as Kotlin. Once you have done that, select the reactive dependencies “Reactive-Web” and “Reactive Mongo” and import the same to your favorite IDE. Please refer the below the screenshot if you have any doubts.

We are using gradle in our example and the dependencies in my build.gradle file looks like this.

[cc]dependencies {
compile('org.springframework.boot:spring-boot-starter-data-mongodb-reactive')
compile('org.springframework.boot:spring-boot-starter-webflux')
compile("org.jetbrains.kotlin:kotlin-stdlib-jre8:${kotlinVersion}")
compile("org.jetbrains.kotlin:kotlin-reflect:${kotlinVersion}")
testCompile('org.springframework.boot:spring-boot-starter-test')
testCompile('io.projectreactor:reactor-test')
}[/cc]

I am going create a Model called Book which has an id and title. Secondly, a BookEvent which contains the Book Title and a date object to see when the event is generated.

[cc lang="java"]@Document
data class Book(@Id var id: String ? = null, var title: String ? = null, var author: String?=null)

data class BookEvent(var id: String? = null, var date : Date?= null)[/cc]

In Kotlin “data” class denotes that the class is created to hold data which will also derive the equals(), hashCode(), toString() and copy() methods by default for you.

If you’re previous experience in using Springboot, you might be aware that the next thing that we need to is to create a Repository.

[cc lang="java"]interface BookRepository : ReactiveMongoRepository<Book, String> {
}[/cc]

If you have noticed we are using the ReactiveMongoRepository instead of the MongoRepository. Reactive Mongo enables you to stream data both in and out of the MongoDB in a non-blocking way.

Next, I am going to create a Service and a Controller to interact with the Mongo Repository and expose them as RESTful APIs.

[cc lang="java"]// Find all the books in the document
fun findAll() = bookRepository.findAll();

// Find a book by its ID
fun findById(id: String) = bookRepository.findById(id);

//Generate events for the book
fun events (id: String) = Flux
.generate({ sink: SynchronousSink -> sink.next(BookEvent(id, Date())) })
.delayElements(Duration.ofSeconds(1L));[/cc]

In the above code, the first two functions are self-explanatory and the third one will generate events. Flux.genreate publishes the data created the subscriber in which we will be using a Synchronous Sink which will emit data one-by-one and its callback method next() can be called at most only once per callback invocation. delayElements() will delay the signal for the specified time un its.

And our controller looks like below.
[cc lang="java"]
@GetMapping("/books")
fun findAllBooks() = bookService.findAll();

@GetMapping("/books/{id}")
fun findBookById(@PathVariable id: String) = bookService.findById(id);

@GetMapping("/books/{id}/events", produces = arrayOf(MediaType.TEXT_EVENT_STREAM_VALUE))
fun events(@PathVariable id: String ) = bookService.events(id);[/cc]

Function events is going to produce an array of Text Event Stream which is an implementation of SSE (Server Sent Events) specification. SSE events are unidirectional unlike websockets and sent over the traditional HTTP. If you do not have MongoDB installed you can use a docker version and run it using

[cc]docker run -p 27017:27017 -d mongo[/cc]

Now we can go ahead and run the application and do curl to test it

[cc]$ curl http://localhost:8080/books
$
[{"id":"59edc7041258c7a930010fb2","title":"Cloud Native Java","author":"Josh Long and Kenny"},{"id":"59edc7041258c7a930010fb4","title":"Patterns of Enterprise Architecture","author":"Martin Fowler"},{"id":"59edc7041258c7a930010fb3","title":"Building Microservices","author":"Sam Newman"}]
$
$ curl http://localhost:8080/books/59edc7041258c7a930010fb4
{"id":"59edc7041258c7a930010fb4","title":"Patterns of Enterprise Architecture","author":"Martin Fowler"}
$
$ curl http://localhost:8080/books/59edc7041258c7a930010fb4/events
data:{"booksId":"59edc7041258c7a930010fb4","date":1508755381442}

data:{"booksId":"59edc7041258c7a930010fb4","date":1508755382452}

data:{"booksId":"59edc7041258c7a930010fb4","date":1508755383458}

data:{"booksId":"59edc7041258c7a930010fb4","date":1508755384459}

data:{"booksId":"59edc7041258c7a930010fb4","date":1508755385465}

data:{"booksId":"59edc7041258c7a930010fb4","date":1508755386471}

data:{"booksId":"59edc7041258c7a930010fb4","date":1508755387473}

data:{"booksId":"59edc7041258c7a930010fb4","date":1508755388477}

data:{"booksId":"59edc7041258c7a930010fb4","date":1508755389479}

data:{"booksId":"59edc7041258c7a930010fb4","date":1508755390482}

data:{"booksId":"59edc7041258c7a930010fb4","date":1508755391487}

data:{"booksId":"59edc7041258c7a930010fb4","date":1508755392492}

data:{"booksId":"59edc7041258c7a930010fb4","date":1508755393496}[/cc]

As the reactive Springboot service ready and now need a consumer for it. Let’s go ahead and create a Web client which will listen to these events. Follow the same steps above from the start.spring.io and create a client service.

[cc lang="java"]@SpringBootApplication
class ReactivewebclientApplication {
@Bean
fun webClient(): WebClient = WebClient.create("http://localhost:8080/books").mutate().build()

@Bean
fun runner (webClient:WebClient)= ApplicationRunner {
webClient.get()
.uri("")
.retrieve()
.bodyToFlux(Book::class.java)
.filter({book -> book.title.equals("Cloud Native Java")})
.flatMap {
webClient.get().uri("/{id}/events", it.id)
.retrieve()
.bodyToFlux(BookEvent::class.java)
}
.subscribe { println(it) }
}

}

fun main(args: Array) {
SpringApplication.run(ReactivewebclientApplication::class.java, *args)
}

data class Book (val id: String? =null, val title: String? =null, val author: String?=null)

data class BookEvent (val id: String? =null, val date: Date? =null)[/cc]

We can use the Spring Rest template and invoke our service but instead of doing that we are goin to use Webclient, which provides a non-blocking way of performing HTTP requests originally part of the Spring Web Flux project but included in the Spring 5 and uses Netty.
Using the Application runner, we will be invoking the service and subscribe to it.

The entire code repository is available here on GitHub.

Tags: Reactive SpringRxjavaSpring5Springboot2