[Reactive] Fluent conversion via compose() & to()

[akarnokd]

Since Java has no extension methods, it is generally inconvenient and ugly to apply custom operators to chains.

The closest thing is to add conversion methods that take a function and return the object to continue chaining onto.

compose

Take the current sequence and apply a function to it to produce the new sequence:

Function<Multi<T>, Multi<T>> addLogging = upstream -> 
     upstream
     .peek(Logger:log)
     .onError(Logger::log);

Function<Multi<T>, Multi<T>> addAsynchrony = upstream ->
      Global.isDebugging() 
          ? upstream.observeOn(Runnable::run) 
          : upstream.observeOn(backgroundExecutor);

source.map(mapping)
         .compose(addLogging)
         .compose(addAsynchrony);

to

Often though, the transformation should result in a type other than the base type. For this, use the to() method which can return any type:

Multi.range(1, 5)
    .map(Object::toString)
    .to(upstream -> {
        TestSubscriber<String> ts = new TestSubscriber<>();
        upstream.subscribe(ts);
        return ts;
    })
    .assertResult("1", "2", "3", "4", "5");