Micronaut Mastery: Using Reactor Mono And Flux

Micronaut is reactive by nature and uses RxJava2 as implementation for the Reactive Streams API by default. RxJava2 is on the compile classpath by default, but we can easily use Project Reactor as implementation of the Reactive Streams API. This allows us to use the Reactor types Mono and Flux. These types are also used by Spring's Webflux framework and makes a transition from Webflux to Micronaut very easy.

How we do use Project Reactor in our Micronaut application? We only have to add the dependency the Project Reactory core library to our project. In the following example we add it to our build.gradle file as:

// File: build.gradle
...
dependencies {
    ...
    // The version of Reactor is resolved
    // via the BOM of Micronaut, so we know
    // the version is valid for Micronaut.
    compile 'io.projectreactor:reactor-core'
    ...
}
...

Now we can use Mono and Flux as return types for methods. If we use them in our controller methods Micronaut will make sure the code is handled on the Netty event loop. This means we must handle blocking calls (like accessing a database using JDBC) with care and make sure a blocking call invoked from the controller methods is handled on a different thread.

In the following example we have a simple controller. Some of the methods use a repository implementation with code that access a databases using JDBC. The methods of the repository implementation are not reactive, therefore we must use Mono.fromCallable with Reactor's elastic scheduler to make sure the code is called on separate threads and will not block our Netty event loop.

package mrhaki;

import io.micronaut.http.annotation.Controller;
import io.micronaut.http.annotation.Get;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.util.concurrent.Callable;

@Controller("/languages")
public class LanguagesController {

    // Repository reads data from database
    // using JDBC and uses simple return types.
    private final LanguagesRepository repository;

    public LanguagesController(final LanguagesRepository repository) {
        this.repository = repository;
    }

    @Get("/{name}")
    public Mono<Language> findByName(final String name) {
        return blockingGet(() -> repository.findByName(name));
    }

    @Get("/")
    public Flux<Language> findAll() {
        return blockingGet(() -> repository.findAll()).flatMapMany(Flux::fromIterable);
    }

    // Run callable code on other thread pool than Netty event loop,
    // so blocking call will not block the event loop.
    private <T> Mono<T> blockingGet(final Callable<T> callable) {
        return Mono.fromCallable(callable)
                   .subscribeOn(Schedulers.elastic());
    }
}

The repository interface looks like this:

package mrhaki;

interface LanguagesRepository {
    List<Language> findAll();
    Language findByName(String name);
}

Let's write a Spock specification to test if our controller works correctly:

package mrhaki

import io.micronaut.context.ApplicationContext
import io.micronaut.core.type.Argument
import io.micronaut.http.HttpRequest
import io.micronaut.http.HttpStatus
import io.micronaut.http.client.HttpClient
import io.micronaut.http.client.exceptions.HttpClientResponseException
import io.micronaut.runtime.server.EmbeddedServer
import spock.lang.AutoCleanup
import spock.lang.Shared
import spock.lang.Specification

class LanguagesControllerSpec extends Specification {

    @AutoCleanup
    @Shared
    private static EmbeddedServer server = ApplicationContext.run(EmbeddedServer)

    @AutoCleanup
    @Shared
    private static HttpClient client = server.applicationContext.createBean(HttpClient, server.URL)

    void '/languages should return all languages'() {
        given:
        final request = HttpRequest.GET('/languages')

        when:
        final response = client.toBlocking().exchange(request, Argument.of(List, Language))

        then:
        response.status() == HttpStatus.OK

        and:
        response.body()*.name == ['Java', 'Groovy', 'Kotlin']

        and:
        response.body().every { language -> language.platform == 'JVM' }
    }

    void '/languages/groovy should find language Groovy'() {
        given:
        final request = HttpRequest.GET('/languages/Groovy')

        when:
        final response = client.toBlocking().exchange(request, Language)

        then:
        response.status() == HttpStatus.OK

        and:
        response.body() == new Language('Groovy', 'JVM')
    }

    void '/languages/dotnet should return 404'() {
        given:
        final request = HttpRequest.GET('/languages/dotnet')

        when:
        client.toBlocking().exchange(request)

        then:
        HttpClientResponseException notFoundException = thrown(HttpClientResponseException)
        notFoundException.status == HttpStatus.NOT_FOUND
    }

}

Written with Micronaut 1.0.0.M4.