RxEventSources { @SafeVarargs public static <E> EventSource<E> fromObservables(ObservableSource<E>... sources) { final Observable<E> eventSource = Observable.mergeArray(sources); return new EventSource<E>() { @Nonnull @Override public Disposable subscribe(final Consumer<E> eventConsumer) { final io.reactivex.disposables.Disposable subscription = eventSource.subscribe( new io.reactivex.functions.Consumer<E>() { @Override public void accept(E e) throws Exception { eventConsumer.accept(e); } }, new io.reactivex.functions.Consumer<Throwable>() { @Override public void accept(Throwable e) throws Exception { RxJavaPlugins.onError(e); } }); return new Disposable() { @Override public void dispose() { subscription.dispose(); } }; } }; } private  RxEventSources(); @SafeVarargs static EventSource<E> fromObservables(ObservableSource<E>... sources); static Observable<E> toObservable(final EventSource<E> eventSource);  }
@Test public void eventsAreForwardedInOrder() throws Exception { EventSource<Integer> source = RxEventSources.fromObservables(Observable.just(1, 2, 3)); RecordingConsumer<Integer> consumer = new RecordingConsumer<>(); source.subscribe(consumer); consumer.waitForChange(50); consumer.assertValues(1, 2, 3); }
@Test public void disposePreventsFurtherEvents() throws Exception { PublishSubject<Integer> subject = PublishSubject.create(); EventSource<Integer> source = RxEventSources.fromObservables(subject); RecordingConsumer<Integer> consumer = new RecordingConsumer<>(); Disposable d = source.subscribe(consumer); subject.onNext(1); subject.onNext(2); d.dispose(); subject.onNext(3); consumer.waitForChange(50); consumer.assertValues(1, 2); }