Используйте процессор Reactive-Streams с RxJava 2.0

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