in mobius-rx2/src/main/java/com/spotify/mobius/rx2/RxEventSources.java [85:108]
public static <E> Observable<E> toObservable(final EventSource<E> eventSource) {
return Observable.create(
new ObservableOnSubscribe<E>() {
@Override
public void subscribe(final ObservableEmitter<E> emitter) throws Exception {
final Disposable disposable =
eventSource.subscribe(
new Consumer<E>() {
@Override
public void accept(E value) {
emitter.onNext(value);
}
});
emitter.setCancellable(
new Cancellable() {
@Override
public void cancel() throws Exception {
disposable.dispose();
}
});
}
});
}