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

Question

我有一个想在RxJava 2.0中使用的org.reactivestreams.Processor。然而，虽然有一些转换可以将org.reactivestreams.Publisher与RxJava集成在一起，比如io.reactivex.Flowable#fromPublisher，但我并不清楚如何最好地集成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);