RxJavaFiberInterop

Library for interoperation between RxJava 3 and Project Loom's Fibers.

dependencies {
    implementation "com.github.akarnokd:rxjava3-fiber-interop:0.0.12"
}

Always requires the latest Loom build from https://jdk.java.net/loom/

Note that Loom is in early access and the API, naming and usage keeps changing, a lot.

Components

FiberInterop

create

Creates a Flowable from a generator callback, that can emit via FiberEmitter, run in a Fiber backed by the ForkJoinPool.commonPool() or any Executor/Scheduler provided as argument and suspended automatically on downstream backpressure.

FiberInterop.create(emitter -> {
    for (int i = 1; i <= 5; i++) {
         emitter.emit(1);
    }
})
.test()
.awaitDone(5, TimeUnit.SECONDS)
.assertResult(1, 2, 3, 4, 5);

transform

Transforms each upstream value via a callback that can emit zero or more values for each of those upstream values, run in a Fiber backed by the ForkJoinPool.commonPool() or any Executor/Scheduler provided as argument and suspended automatically on downstream backpressure.

Flowable.range(1, 5)
.compose(FiberInterop.transform((value, emitter) -> {
    emitter.emit(value);
    emitter.emit(value + 1);
}))
.test()
.awaitDone(5, TimeUnit.SECONDS)
.assertResult(1, 2, 2, 3, 3, 4, 4, 5, 5, 6);