RxEventSources { @SafeVarargs public static <E> EventSource<E> fromObservables(@NonNull final ObservableSource<E>... sources) { final Observable<E> eventSource = Observable.mergeArray(sources); return new EventSource<E>() { @Nonnull @Override public com.spotify.mobius.disposables.Disposable subscribe( com.spotify.mobius.functions.Consumer<E> eventConsumer) { final Disposable disposable = eventSource.subscribe( new Consumer<E>() { @Override public void accept(E value) throws Throwable { eventConsumer.accept(value); } }, new Consumer<Throwable>() { @Override public void accept(Throwable error) throws Throwable { RxJavaPlugins.onError(error); } }); return new com.spotify.mobius.disposables.Disposable() { @Override public void dispose() { disposable.dispose(); } }; } }; } private  RxEventSources(); @SafeVarargs static EventSource<E> fromObservables(@NonNull final ObservableSource<E>... sources); @NonNull 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); }