package rx;

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.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

public class RxJavaUtils {


    public static void main(String[] args) throws InterruptedException {
//        intervalRange();
//        intervalRange2();
        intervalRange3();
    }

    private static void intervalRange3() {
    }


    /**
     * 不输出
     *
     * @throws InterruptedException
     */
    private static void intervalRange() throws InterruptedException {
        Observable.intervalRange(0, 10, 0, 1, TimeUnit.MILLISECONDS)
                .subscribeOn(Schedulers.io()) // 指定在IO线程执行
                .observeOn(Schedulers.single()) // 指定在单一线程观察
                .subscribe(new Observer<Long>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        // Called when the Observer subscribes to the Observable
                        System.out.println("Subscribed");
                    }

                    @Override
                    public void onNext(Long value) {
                        // Called when a new value is emitted
                        System.out.println("intervalRange1: " + value);
                    }

                    @Override
                    public void onError(Throwable e) {
                        // Called when an error occurs
                        System.out.println("Error: " + e.getMessage());
                    }

                    @Override
                    public void onComplete() {
                        // Called when the Observable completes emitting all values
                        System.out.println("Completed");
                    }
                });
//        Thread.sleep(1000);
    }

    public static void intervalRange2() throws InterruptedException {
        System.out.println("intervalRange2:");
        // 使用CountDownLatch等待Observable完成
        CountDownLatch latch = new CountDownLatch(1);
        Observable.intervalRange(0, 5, 0, 100, TimeUnit.MILLISECONDS)
                .subscribeOn(Schedulers.io()) // 指定在IO线程执行
                .observeOn(Schedulers.single()) // 指定在单一线程观察
                .subscribe(new Observer<Long>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        // Called when the Observer subscribes to the Observable
                        System.out.println("Subscribed");
                    }

                    @Override
                    public void onNext(Long value) {
                        // Called when a new value is emitted
                        System.out.println("intervalRange2: " + value);
                    }

                    @Override
                    public void onError(Throwable e) {
                        // Called when an error occurs
                        System.out.println("Error: " + e.getMessage());
                        latch.countDown();
                    }

                    @Override
                    public void onComplete() {
                        // Called when the Observable completes emitting all values
                        System.out.println("Completed");
                        latch.countDown();
                    }
                });
        // 等待Observable完成
//        latch.await(1, TimeUnit.SECONDS);
    }

}