RxJavaJdk9Interop

RxJava 3 interop library for supporting Java 9 features such as Flow.*

Release

compile 'com.github.akarnokd:rxjava3-jdk9-interop:3.0.0'

Examples

Converting from RxJava 3 to Java 9 Flow

import hu.akarnokd.rxjava3.jdk9interop.*;

Flow.Publisher<Integer> pub = Flowable.range(1, 5)
    .to(FlowInterop.toFlow());

// --------

Flow.Processor<Integer, Integer> proc = FlowInterop
    .toFlowProcessor(PublishProcessor.create()); 

Converting from Java 9 Flow to RxJava 3

SubmissionPublisher<Integer> sp = new SubmissionPublisher<>();

Flowable<Integer> f = FlowInterop
    .fromFlowPublisher(sp);

// --------

Flow.Processor<Integer, Integer> fp = ...

FlowableProcessor<Integer> fproc = FlowInterop
    .fromFlowProcessor(fp);

Note that RxJava 3 FlowableProcessors don't support different input and output types therefore the Flow.Processor should have the same type arguments.

For convenience, there is a FlowTestSubscriber that extends TestSubscriber and allows asserting on a Flow.Publisher the same way as with Reactive-Streams Publisher types.

FlowTestSubscriber<Integer> ts = new FlowTestSubscriber<>();

SubmissionPublisher<Integer> sp = new SubmissionPublisher<>();

sp.subscribe(ts);

sp.onNext(1);
sp.onNext(2);
sp.onNext(3);
sp.onNext(4);
sp.onNext(5);
sp.close();

ts.awaitDone(5, TimeUnit.SECONDS) // SubmissionPublisher is async by default
  .assertResult(1, 2, 3, 4, 5);