在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);