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);
 }

}