15 RxJava

The excellent RxJava can be used in Ratpack applications to elegantly compose asynchronous operations.

The ratpack-rx2 module provides the RxRatpack class that provides static methods for adapting Ratpack promises to RxJava’s Observable.

The ratpack-rx2 module as of 2.0.0-rc-1 is built against (and depends on) RxJava 2.2.21.

1.15 Initialization

The RxRatpack.initialize() must be called to fully enable the integration. This method only needs to be called once for the JVM’s lifetime.

2.15 Observing Ratpack

The integration is based on the RxRatpack.single() and RxRatpack.observe() static methods. These methods adapt Ratpack’s promise type into an observable, which can then be used with all of the observable operators that RxJava offers.

For example, blocking operations can be easily observed.

import ratpack.exec.Promise;
import ratpack.exec.Blocking;
import ratpack.test.handling.HandlingResult;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static ratpack.rx2.RxRatpack.single;
import static ratpack.test.handling.RequestFixture.requestFixture;

public class Example {
  public static void main(String... args) throws Exception {
    HandlingResult result = requestFixture().handle(context -> {
      Promise<String> promise = Blocking.get(() -> "hello world");
      single(promise).map(String::toUpperCase).subscribe(context::render);
    });

    assertEquals("HELLO WORLD", result.rendered(String.class));
  }
}

3.15 Implicit error handling

A key feature of the RxJava integration is the implicit error handling. All observable sequences have an implicit default error handling strategy of forwarding the exception to the execution context error handler. In practice, this means that error handlers rarely need to be defined for observable sequences.

import ratpack.core.error.ServerErrorHandler;
import ratpack.rx2.RxRatpack;
import ratpack.test.handling.RequestFixture;
import ratpack.test.handling.HandlingResult;
import io.reactivex.Observable;

import static org.junit.jupiter.api.Assertions.assertEquals;

public class Example {
  public static void main(String... args) throws Exception {
    RxRatpack.initialize(); // must be called once per JVM

    HandlingResult result = RequestFixture.requestFixture().handleChain(chain -> {
      chain.register(registry ->
          registry.add(ServerErrorHandler.class, (context, throwable) ->
              context.render("caught by error handler: " + throwable.getMessage())
          )
      );

      chain.get(ctx -> Observable.<String>error(new Exception("!")).subscribe((s) -> {}));
    });

    assertEquals("caught by error handler: !", result.rendered(String.class));
  }
}

In this case, the throwable thrown during the blocking operation will be forwarded to the current ServerErrorHandler, which will probably render an error page to the response. If the subscriber does implement an error handling strategy, it will be used instead of the implicit error handler.

The implicit error handling applies to all observables that are created on Ratpack managed threads. It is not restricted to observables that are backed by Ratpack promises.