Skip to content

Instantly share code, notes, and snippets.

@anitaa1990
Created August 18, 2018 08:41
Show Gist options
  • Save anitaa1990/565d8896a27827bc0847d7d062169c9c to your computer and use it in GitHub Desktop.
Save anitaa1990/565d8896a27827bc0847d7d062169c9c to your computer and use it in GitHub Desktop.
getOriginalObservable()
.switchMap(new Function<Integer, Observable<Integer>>() {
@Override
public Observable<Integer> apply(final Integer integer) {
return getModifiedObservable(integer);
}
})
.subscribeOn(Schedulers.io())
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Integer integer) {
System.out.println("onNext: " + integer);
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
private Observable<Integer> getOriginalObservable() {
final List<Integer> integers = Arrays.asList(1, 2, 3, 4, 5, 6);
return Observable
.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) {
for(Integer integer : integers) {
if (!emitter.isDisposed()) {
emitter.onNext(integer);
}
}
if(!emitter.isDisposed()) {
emitter.onComplete();
}
}
});
}
private Observable<Integer> getModifiedObservable(final Integer integer) {
return Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws InterruptedException {
emitter.onNext((integer * 2));
emitter.onComplete();
}
}).subscribeOn(Schedulers.io());
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment