Micronaut Mastery: Running Code On Startup

When our Micronaut application starts we can listen for the ServiceStartedEvent event and write code that needs to run when the event is fired. We can write a bean that implements the ApplicationEventListener interface with the type ServiceStartedEvent. Or we can use the @EventListener annotation on our method with code we want to run on startup. If the execution of the code can take a while we can also add the @Async annotation to the method, so Micronaut can execute the code on a separate thread.

In our example application we have a reactive repository for a Mongo database and we want to save some data in the database when our Micronaut application starts. First we write a bean that implements the ApplicationEventListener:

// File: src/main/java/mrhaki/Dataloader.java
package mrhaki;

import io.micronaut.context.annotation.Requires;
import io.micronaut.context.env.Environment;
import io.micronaut.context.event.ApplicationEventListener;
import io.micronaut.discovery.event.ServiceStartedEvent;
import io.micronaut.scheduling.annotation.Async;
import io.reactivex.Flowable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.inject.Singleton;

@Singleton
@Requires(notEnv = Environment.TEST) // Don't load data in tests.
public class DataLoader implements ApplicationEventListener<ServiceStartedEvent> {

    private static final Logger log = LoggerFactory.getLogger(DataLoader.class);

    /**
     * Reactive repository for Mongo database to store
     * Conference objects with an id and name property.
     */
    private final ConferenceRepository repository;

    public DataLoader(final ConferenceRepository repository) {
        this.repository = repository;
    }

    @Async
    @Override
    public void onApplicationEvent(final ServiceStartedEvent event) {
        log.info("Loading data at startup");

        // Transform names to Conferences object and save them.
        Flowable.just("Gr8Conf", "Greach", "JavaLand", "JFall", "NextBuild")
                .map(name -> new Conference(name))
                .forEach(this::saveConference);
    }

    /**
     * Save conference in repository.
     * 
     * @param conference Conference to be saved.
     */
    private void saveConference(Conference conference) {
        repository
                .save(conference)
                .subscribe(
                        saved -> log.info("Saved conference {}.", saved),
                        throwable -> log.error("Error saving conference.", throwable));
    }
}

Alternatively we could have used the @EventListener annotation on a method with an argument of type ServiceStartedEvent:

// File: src/main/java/mrhaki/Dataloader.java
package mrhaki;

import io.micronaut.context.annotation.Requires;
import io.micronaut.context.env.Environment;
import io.micronaut.context.event.ApplicationEventListener;
import io.micronaut.discovery.event.ServiceStartedEvent;
import io.micronaut.runtime.event.annotation.EventListener;
import io.micronaut.scheduling.annotation.Async;
import io.reactivex.Flowable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.inject.Singleton;

@Singleton
@Requires(notEnv = Environment.TEST) // Don't load data in tests.
public class DataLoader {

    private static final Logger log = LoggerFactory.getLogger(DataLoader.class);

    /**
     * Reactive repository for Mongo database to store
     * Conference objects with an id and name property.
     */
    private final ConferenceRepository repository;

    public DataLoader(final ConferenceRepository repository) {
        this.repository = repository;
    }

    @EventListener
    @Async
    public void loadConferenceData(final ServiceStartedEvent event) {
        log.info("Loading data at startup");

        // Transform names to Conferences object and save them.
        Flowable.just("Gr8Conf", "Greach", "JavaLand", "JFall", "NextBuild")
                .map(name -> new Conference(name))
                .forEach(this::saveConference);
    }

    /**
     * Save conference in repository.
     * 
     * @param conference Conference to be saved.
     */
    private void saveConference(Conference conference) {
        repository
                .save(conference)
                .subscribe(
                        saved -> log.info("Saved conference {}.", saved),
                        throwable -> log.error("Error saving conference.", throwable));
    }
}

When we start our Micronaut application we can see in the log messages that our conference data is created:

22:58:17.343 [pool-1-thread-1] INFO  mrhaki.DataLoader - Loading data at startup
22:58:17.343 [main] INFO  io.micronaut.runtime.Micronaut - Startup completed in 1230ms. Server Running: http://localhost:9000
22:58:17.573 [Thread-11] INFO  mrhaki.DataLoader - Saved conference Conference{id=5bb134f505d4feefa74d19c7, name='JFall'}.
22:58:17.573 [Thread-8] INFO  mrhaki.DataLoader - Saved conference Conference{id=5bb134f505d4feefa74d19c3, name='Gr8Conf'}.
22:58:17.573 [Thread-10] INFO  mrhaki.DataLoader - Saved conference Conference{id=5bb134f505d4feefa74d19c5, name='Greach'}.
22:58:17.573 [Thread-9] INFO  mrhaki.DataLoader - Saved conference Conference{id=5bb134f505d4feefa74d19c8, name='NextBuild'}.
22:58:17.573 [Thread-6] INFO  mrhaki.DataLoader - Saved conference Conference{id=5bb134f505d4feefa74d19c6, name='JavaLand'}.

Written with Micronaut 1.0.0.RC1.

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.