RxJava Used by Cortex

The Helix programming model for Cortex uses a subset of RxJava classes as expected return types for all prototype operations. This document covers the specific parts of RxJava used by Cortex. More in-depth documentation can be found at ReactiveX.

All prototype operation interfaces return one of the following RxJava types: Observable<T>, Single<T>, Maybe<T> or Completable.

Observable

Where a prototype class expects zero or more values to be returned, the generated operation interfaces return io.reactivex.Observable.

To return an Observable from a collection:

Observable.just(someList);

In the event of zero values being returned, an empty Observable can be returned:

Observable.empty();

Observable can also return an error:

Observable.error(new MyException());

Single

In cases where a prototype class expects a single return value, the generated operation interfaces return io.reactivex.Single. Single can be used to express that either a value has been emitted or that the operation has failed.

To express a value:

Single.just("foo");

To express a failure:

Single.error(new MyException());

Note that in contrast to Observable<T>, there is no concept of an empty return value. Single must either return a value or an error.

Maybe

In cases where a prototype class expects an optional single return value, the generated operation interfaces return io.reactivex.Maybe. Maybe can be used to express that a value has been emitted, the value is empty, or that the operation has failed.

To express a value:

Maybe.just("foo");

In the event of an empty value, an empty Maybe can be returned:

Maybe.empty();

To express a failure:

Maybe.error(new MyException());

Completable

In cases where a prototype class is not interested in a return value of generated operation interfaces, io.reactivex.Completable is returned. Completable indicates whether an operation was successful or it failed.

To express a successful computation:

Completable.complete();

To express a failure:

Completable.error(new MyException());

Operation Failures

The only runtime exceptions that prototype classes understand are ResourceOperationFailure exceptions. These should be used.

For example, to return a NOT_FOUND failure:

return Observable.error(ResourceOperationFailure.notFound());

Rx Repositories

In some Helix Pattern Library examples, you’ll see repositories that return RxJava Observable, Single, Maybe or Completable types. While this is not necessary, returning these allows prototype implementations to make use of the various Rx combinators.

For example:

@Override
public Single<ThingEntity> onRead() {
    return repository
        .retrieveThing(thingId.getValue())
        .map(thing -> ThingEntity.builder()
        .withName(thing.getName())
        .withDescription(thing.getDescription())
        .build());
}

In the preceding example, the repository returns a Single<Thing> which allows use of the map operator to convert the returned value into a ThingEntity.

The same example, but with a "regular" return type:

@Override
public Single<ThingEntity> onRead() {
    Thing thing = repository.retrieveThing(thingId.getValue());
    ThingEntity thingEntity =  ThingEntity.builder()
      .withName(thing.getName())
      .withDescription(thing.getDescription())
      .build();
    return Single.just(thingEntity);
}

In addition to being able to use Rx combinators, maintaining the "Observable chain" makes handling and propagating errors fairly simple, without need for try/catch blocks.

RxJava Unit Testing

Unit testing of RxJava Observable types is straightforward. The RxJava test() method returns a TestObserver<T> object, which supports a wide range of xunit type test assertions.

Here is an example of testing the Observable<String> getWishlistIds(...) repository method:

repository.getWishlistIds("customer-id", STORE_CODE)
    .test()
    .assertNoErrors()
    .assertValueCount(1)
    .assertValue(GOOD_WISH_LIST_GUID);

note

The TestObserver can be used with all observable types: Observable, Single, Maybe and Completable.

Debugging via Stack Traces

To debug RxJava code via stack traces, use the RxJava2Extensions plugin, available from RxJava’s github.

RxJava code is executed upon subscription to an Observable type. Because of this, when a failure occurs, an exception stack trace may not tell which Rx operation failed. The RxJavaAssemblyTracking class captures the stack trace when Observable, Single, Maybe and Completable operators are instantiated at assembly-time. Whenever an error is signaled via onError(), this assembly-time stack trace is attached as the last cause of that exception.

Example: Enabling Assembly Tracking for the Entire Application

caution

Enabling assembly tracking has severe performance impact, and it should never be enabled in production. It should only be used for debugging purposes in development.

Add the dependency for RxJava2Extensions:

<dependency>
    <groupId>com.github.akarnokd</groupId>
    <artifactId>rxjava2-extensions</artifactId>
    <version>0.18.1</version>
</dependency>

Enable assembly tracking:

@Component
public class RxAssemplyHookInitializer {
  @Activate
  public void initialize() {
    RxJavaAssemblyTracking.enable();
  }
}

Once enabled, use the following snippet:

Observable.empty().single()
  .subscribe(System.out::println, Throwable::printStackTrace);

Which will result in a stacktrace similar to the one shown below:

java.lang.NoSuchElementException
at rx.internal.operators.OnSubscribeSingle(OnSubscribeSingle.java:57)
...
RxJavaAssemblyException: assembled
at com.example.TrackingExample(TrackingExample:10)