package rx;

import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.ObservableSource;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.functions.Function;
import io.reactivex.rxjava3.schedulers.Schedulers;

public class RxDemo {
 public static void main(String[] args) throws InterruptedException {
 Observable<Integer> observable = Observable.just(1, 2, 3, 4);

 observable.flatMap(new Function<Integer, ObservableSource<Integer>>() {
 @Override
 public ObservableSource<Integer> apply(Integer integer) throws Throwable {
 System.out.println("apply: t=="+Thread.currentThread().getName());
 System.out.println("apply:"+integer);
 return Observable.just(integer);
 }
 })
 .subscribeOn(Schedulers.computation())
 .observeOn(Schedulers.newThread())
 .subscribe(new Observer<Integer>() {

 @Override
 public void onError(Throwable e) {
 System.out.println("e = [" + e + "]");
 }

 @Override
 public void onComplete() {
 System.out.println("Complete");
 }

 @Override
 public void onSubscribe(@NonNull Disposable d) {
 System.out.println("onSubscribe");
 }

 @Override
 public void onNext(Integer s) {
 System.out.println("s = [" + s + "]");
 }
 });
 }

}