rx-java - 在 RxJava 2.0 中使用 Reactive-Streams 处理器

Question

我有一个org.reactivestreams.Processor我想与 RxJava 2.0 一起使用的。然而，虽然有将 aorg.reactivestreams.Publisher与 RxJava 集成的转换，比如io.reactivex.Flowable#fromPublisher，但我不清楚如何最好地集成 a org.reactivestreams.Processor（或org.reactivestreams.Subscriber）。任何人都可以对此有所了解吗？

Answer 1

您将一侧包裹起来Publisher并保持Subscriber原样：

Processor proc = ...

Subscriber sub = proc;
Flowable flow = Flowable.fromPublisher(proc);

flow.map(v -> v.toString()).subscribe(System.out::println);

sub.onNext(1);