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 java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

public class RxJavaSequentialTasks {
 static Disposable disposable;

 public static void main(String[] args) {
 Observable.fromArray(1, 2, 3, 4, 5)
 .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 {
// Thread.sleep(1000);
 int i = get().get(1000, TimeUnit.MILLISECONDS);
 return i;
 }
 });
 }
 })
 .subscribe(new Observer<Integer>() {
 @Override
 public void onSubscribe(@NonNull Disposable d) {
 disposable = d;

 }

 @Override
 public void onNext(@NonNull Integer s) {
 System.out.println("onNext:" + s);
 if (s == 2) {
 disposable.dispose();
 }
 }

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

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


 }

 private static CompletableFuture<Integer> get() {
 CompletableFuture<Integer> completableFuture = new CompletableFuture<>();
 final int[] i = {0};
 new Thread(new Runnable() {
 @Override
 public void run() {
 try {
 Thread.sleep(500);
 } catch (InterruptedException e) {
 throw new RuntimeException(e);
 }
 i[0] = i[0] + 2;
 completableFuture.complete(i[0]);
 }
 }).start();
 return completableFuture;
 }

 private static String longRunningTask(int number) {
 // 模拟耗时操作,比如网络请求或计算
 try {
 Thread.sleep(100); // 假设每个任务耗时1秒
 } catch (InterruptedException e) {
 Thread.currentThread().interrupt();
 }
 return "Task " + number + " completed";
 }
}