RxJavaExtensions

RxJava 3.x implementation of extra sources, operators and components and ports of many 1.x companion libraries.

Releases

gradle

dependencies {
    implementation "com.github.akarnokd:rxjava3-extensions:3.0.1"
}

Javadoc: https://akarnokd.github.com/RxJavaExtensions/javadoc/index.html

Maven search:

http://search.maven.org

Features

• Extra functional interfaces
• Mathematical operations over numerical sequences
• String operations
• Asynchronous jumpstarting a sequence
• Computational expressions
• Join patterns
• Debug support
  • Function tagging
  • Protocol validation
  • Multi-hook handlers
• Custom Processors and Subjects
  • SoloProcessor, PerhapsProcessor and NonoProcessor
  • MulticastProcessor (Deprecated in 0.19.2!),
  • UnicastWorkSubject,
  • DispatchWorkSubject,
  • DispatchWorkProcessor
• FlowableProcessor utils
  • wrap, refCount
• Custom Schedulers
  • SharedScheduler
  • ParallelScheduler
  • BlockingScheduler
• Custom operators and transformers
  • valve(), orderedMerge(), bufferWhile(),
  • bufferUntil(), bufferSplit(), spanout(),
  • mapFilter(), onBackpressureTimeout(), repeat(),
  • repeatCallable(), every(), intervalBackpressure(),
  • cacheLast(), timeoutLast(), timeoutLastAbsolute(),
  • debounceFirst(), switchFlatMap(), flatMapSync(),
  • flatMapAsync(), switchIfEmpty(),
  • expand(), mapAsync(), filterAsync(),
  • zipLatest(), coalesce(),
  • windowWhile(), windowUntil(), windowSplit(),
  • indexOf(), requestObserveOn(), requestSample()
  • observeOnDrop(), observeOnLatest(), generateAsync(),
  • partialCollect(), flatMapDrop(), flatMapLatest(),
  • errorJump(), flatMap on signal type, switchOnFirst()
• Custom parallel operators and transformers
  • sumX()
  • orderedMerge()
• Special Publisher implementations
• Custom consumers
  • FlowableConsumers
  • subscribeAutoDispose()
  • ObservableConsumers
  • SingleConsumers
  • MaybeConsumers
  • CompletableConsumers

Extra functional interfaces

Support the join-patterns and async-util with functional interfaces of consumers with 3-9 type arguments and have functional interfaces of functions without the throws Exception.

• SimpleCallable<T> - Callable<T> without throws Exception
• Consumer3 - 3 argument Consumer
• Consumer4 - 4 argument Consumer
• Consumer5 - 5 argument Consumer
• Consumer6 - 6 argument Consumer
• Consumer7 - 7 argument Consumer
• Consumer8 - 8 argument Consumer
• Consumer9 - 9 argument Consumer
• PlainFunction - Function without throws Exception
• PlainBiFunction - BiFunction without throws Exception
• PlainFunction3 - Function3 without throws Exception
• PlainFunction4 - Function4 without throws Exception
• PlainFunction5 - Function5 without throws Exception
• PlainFunction6 - Function6 without throws Exception
• PlainFunction7 - Function7 without throws Exception
• PlainFunction8 - Function8 without throws Exception
• PlainFunction9 - Function9 without throws Exception

Utility functions supporting these can be found in FunctionsEx class.

Mathematical operations over numerical sequences

Although most of the operations can be performed with reduce, these operators have lower overhead as they cut out the reboxing of primitive intermediate values.

The following operations are available in MathFlowable for Flowable sequences and MathObservable in Observable sequences:

• averageDouble()
• averageFloat()
• max()
• min()
• sumDouble()
• sumFloat()
• sumInt()
• sumLong()

Example

MathFlowable.averageDouble(Flowable.range(1, 10))
.test()
.assertResult(5.5);

Flowable.just(5, 1, 3, 2, 4)
.to(MathFlowable::min)
.test()
.assertResult(1);

String operations

characters

The StringFlowable and StringObservable support streaming the characters of a CharSequence:

StringFlowable.characters("Hello world")
.map(v -> Characters.toLower((char)v))
.subscribe(System.out::print, Throwable::printStackTrace, System.out::println);

split

Splits an incoming sequence of Strings based on a Regex pattern within and between subsequent elements if necessary.

Flowable.just("abqw", "ercdqw", "eref")
.compose(StringFlowable.split("qwer"))
.test()
.assertResult("ab", "cd", "ef");

Flowable.just("ab", ":cde:" "fg")
.compose(StringFlowable.split(":"))
.test()
.assertResult("ab", "cde", "fg");

Asynchronous jumpstarting a sequence

Wrap functions and consumers into Flowables and Observables or into another layer of Functions. Most of these can now be achieved via fromCallable and some function composition in plain RxJava.

start

Run a function or action once on a background thread and cache its result.

AtomicInteger counter = new AtomicInteger();

Flowable<Integer> source = AsyncFlowable.start(() -> counter.incrementAndGet());

source.test()
    .awaitDone(5, TimeUnit.SECONDS)
    .assertResult(1);

source.test()
    .awaitDone(5, TimeUnit.SECONDS)
    .assertResult(1);

toAsync

Call a function (with parameters) to call a function inside a Flowable/Observable with the same parameter and have the result emitted by that Flowable/Observable from a background thread.

Function<Integer, Flowable<String>> func = AsyncFlowable.toAsync(
    param -> "[" + param + "]"
);

func.apply(1)
    .test()
    .awaitDone(5, TimeUnit.SECONDS)
    .assertResult("[1]")
;

startFuture

Run a Supplier that returns a Future to call blocking get() on to get the solo value or exception.

ExecutorService exec = Executors.newSingleThreadedScheduler();

AsyncFlowable.startFuture(() -> exec.submit(() -> 1))
    .test()
    .awaitDone(5, TimeUnit.SECONDS)
    .assertResult(1);

exec.shutdown();

deferFuture

Run a Supplier that returns a Future to call blocking get() on to get a Publisher to stream back.

ExecutorService exec = Executors.newSingleThreadedScheduler();

AsyncFlowable.startFuture(() -> exec.submit(() -> Flowable.range(1, 5)))
    .test()
    .awaitDone(5, TimeUnit.SECONDS)
    .assertResult(1, 2, 3, 4, 5);

exec.shutdown();

forEachFuture

Consume a Publisher and have Future that completes when the consumption ends with onComplete or onError.

Future<Object> f = AsyncFlowable.forEachFuture(Flowable.range(1, 100), System.out::println);

f.get();

runAsync

Allows emitting multiple values through a Processor mediator from a background thread and allows disposing the sequence externally.

AsyncFlowable.runAsync(Schedulers.single(),
        UnicastProcessor.<Object>create(),
        new BiConsumer<Subscriber<Object>, Disposable>() {
            @Override
            public void accept(Subscriber<? super Object> s, Disposable d) throws Exception {
                s.onNext(1);
                s.onNext(2);
                s.onNext(3);
                Thread.sleep(200);
                s.onNext(4);
                s.onNext(5);
                s.onComplete();
            }
        }
).test()
.awaitDone(5, TimeUnit.SECONDS)
.assertResult(1, 2, 3, 4, 5);

Computational expressions

The operators on StatementFlowable and StatementObservable allow taking different branches at subscription time:

ifThen

Conditionally chose a source to subscribe to. This is similar to the imperative if statement but with reactive flows:

if ((System.currentTimeMillis() & 1) != 0) {
    System.out.println("An odd millisecond");
} else {
    System.out.println("An even millisecond");
}

Flowable<String> source = StatementFlowable.ifThen(
    () -> (System.currentTimeMillis() & 1) != 0,
    Flowable.just("An odd millisecond"),
    Flowable.just("An even millisecond")
);

source
.delay(1, TimeUnit.MILLISECONDS)
.repeat(1000)
.subscribe(System.out::println);

switchCase

Calculate a key and pick a source from a Map. This is similar to the imperative switch statement:

switch ((int)(System.currentTimeMillis() & 7)) {
case 1: System.out.println("one"); break;
case 2: System.out.println("two"); break;
case 3: System.out.println("three"); break;
default: System.out.println("Something else");
}

Map<Integer, Flowable<String>> map = new HashMap<>();

map.put(1, Flowable.just("one"));
map.put(2, Flowable.just("two"));
map.put(3, Flowable.just("three"));

Flowable<String> source = StatementFlowable.switchCase(
    () -> (int)(System.currentTimeMillis() & 7),
    map,
    Flowable.just("Something else")
);

source
.delay(1, TimeUnit.MILLISECONDS)
.repeat(1000)
.subscribe(System.out::println);

doWhile

Resubscribe if a condition is true after the last subscription completed normally. This is similar to the imperative do-while loop (executing the loop body at least once):

long start = System.currentTimeMillis();
do {
    Thread.sleep(1);
    System.out.println("Working...");
while (start + 100 > System.currentTimeMillis());

long start = System.currentTimeMillis();

Flowable<String> source = StatementFlowable.doWhile(
    Flowable.just("Working...").delay(1, TimeUnit.MILLISECONDS),
    () -> start + 100 > System.currentTimeMillis()
);

source.subscribe(System.out::println);

whileDo

Subscribe and resubscribe if a condition is true. This is similar to the imperative while loop (where the loop body may not execute if the condition is false to begin with):

while ((System.currentTimeMillis() & 1) != 0) {
    System.out.println("What an odd millisecond!");
}

Flowable<String> source = StatementFlowable.whileDo(
    Flowable.just("What an odd millisecond!"),
    () -> (System.currentTimeMillis() & 1) != 0
);

source.subscribe(System.out::println);

Join patterns

(Conversion done)

TBD: examples

Debug support

By default, RxJava 3's RxJavaPlugins only offers the ability to hook into the assembly process (i.e., when you apply an operator on a sequence or create one) unlike 1.x where there is an RxJavaHooks.enableAssemblyTracking() method. Since the standard format is of discussion there, 3.x doesn't have such feature built in but only in this extension library.

Usage

You enable tracking via:

RxJavaAssemblyTracking.enable();

and disable via:

RxJavaAssemblyTracking.disable();

Note that this doesn't save or preserve the old hooks (named Assembly) you may have set as of now.

Output

In debug mode, you can walk through the reference graph of Disposables and Subscriptions to find an FlowableOnAssemblyX named nodes (similar in the other base types) where there is