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

Question

我有一个org.reactivestreams.Processor，我希望在RxJava 2.0中使用它。然而，尽管有转换来将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);