package rx;


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

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;

public class RxJavaAllTasksCompleteExample {
 public static void main(String[] args) {
 Observable<Integer> task1 = Observable.just(1);
 Observable<Integer> task2 = Observable.just(2);
 Observable<Integer> task3 = Observable.just(3);
 Observable<String> task4 = Observable.fromCallable(new Callable<String>() {
 @Override
 public String call() throws Exception {
 System.out.println("call t= " + Thread.currentThread().getName());
 Thread.sleep(5000);
 return "1111";
 }
 });
 List<Observable<String>> list = new ArrayList<>();
 list.add(task4);
 Observable.fromArray(list);
 Observable.just(list);
 Observable.merge(task1, task2, task4, task3)
// .flatMap(new Function<Integer, ObservableSource<Integer>>() {
// @Override
// public ObservableSource<Integer> apply(Integer integer) throws Throwable {
// return Observable.fromCallable(new Callable<Integer>() {
// @Override
// public Integer call() throws Exception {
// return integer;
// }
// });
// }
// })

 .subscribeOn(Schedulers.computation())
 .observeOn(Schedulers.computation())
 .subscribe(new Observer<Object>() {
 @Override
 public void onSubscribe(@NonNull Disposable d) {

 }

 @Override
 public void onNext(@NonNull Object s) {
 if (s instanceof String) {
 System.out.println("onNext:instanceof " + s);
 } else {

 System.out.println("onNext:" + s);
 }
 }

 @Override
 public void onError(@NonNull Throwable e) {
 System.out.println(e.getMessage());
 }

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


 }
}