package rx;

import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.*;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.functions.Consumer;
import io.reactivex.rxjava3.functions.Function;
import io.reactivex.rxjava3.functions.Predicate;
import log.Log;

import java.util.concurrent.atomic.AtomicInteger;

public class Rx3Java {
 public static void main(String[] args) throws InterruptedException {
// extracted();
// extracted1();
 sysout();

 }

 private static void sysout() {
 Flowable.just("Hello world").subscribe(System.out::println);
 //

 Flowable<Integer> flow = Flowable.range(1, 5)
 .map(v -> v * v)
 .filter(v -> v % 2 == 0);
 flow.subscribe(System.out::println);

 AtomicInteger count = new AtomicInteger();

 Observable.range(1, 100)
 .doOnNext(ignored -> count.incrementAndGet())
 .ignoreElements()
 .andThen(Single.just(count.get()))
 .subscribe(System.out::println);
 }

 private static void extracted1() {
 //创建被观察者Observable(电视剧)
 //使用create操作符从头开始创建一个Observable
 Observable<String> TVplay = Observable.create(new ObservableOnSubscribe<String>() {//ObservableOnSubscribe<T>其实就像一个队列一样
 @Override
 public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Throwable {//ObservableEmitter可以理解为事件发射器
 //重写subscribe()定义需要发送的事件
 //ObservableEmitter有三种发射方法,onNext()、onComplete()、onError()
 //onNext()发送事件,可以无限调用,所有的发射信息观察者都可以收得到
 emitter.onNext("电视剧第一集");
 emitter.onNext("电视剧第二集");
 emitter.onNext("电视剧第三集");
 //onComplete()、onError()是互斥的,观察者只会收到一个
 //onComplete()完成事件发送,可以重复调用,但是观察者只会接收一次
 emitter.onComplete();
 //onError()异常发送,不可以重复调用
 //emitter.onError(new IllegalArgumentException("TVplay Exception!"));
 }
 });
 //RxJava还提供了其他方法用于创建被观察者对象Observable
 //just(T...):直接将传入的参数依次发送出来(将一个或多个对象转换成发射这个或这些对象的一个Observable)
 // Observable<String> observable = Observable.just("电视剧第一集", "电视剧第二集", "电视剧第三集");//这个跟上面的作用是一样的
 //还有很多,比如from等...create、just、from其实就是RxJava的操作符,可以期待本人的RxJava操作符文章

 //创建观察者Observer(观影者)
 Observer<String> viewer = new Observer<String>() {
 //定义Disposable类变量
 //private Disposable mDisposable;

 //事件队列开始
 @Override
 public void onSubscribe(@NonNull Disposable d) {//Disposable可以当作订阅关系
 //当订阅时被调用
 Log.e("RxJava_TAG", "onSubscribe");
 //取消/解除订阅
 //d.dispose();
 //是否已经解除订阅
 //d.isDisposed();
 //可以这样操作
 //mDisposable = d;
 }

 //普通事件
 @Override
 public void onNext(@NonNull String s) {
 //发送数据时被多次调用
 Log.e("RxJava_TAG", "onNext" + s);
 //取消/解除订阅实例
 //if ("电视剧第二集".equals(s)) {
 // //设置在接收到"电视剧第三集"事件后切断观察者和被观察者的连接
 // mDisposable.dispose();
 // Log.d("RxJava_TAG", "是否已经切断了连接:" + mDisposable.isDisposed());
 //}
 }

 //事件队列异常事件
 @Override
 public void onError(@NonNull Throwable e) {
 //在事件处理过程中出现异常时,onError()方法会被触发
 Log.e("RxJava_TAG", "onError");
 }

 //事件队列完结事件
 @Override
 public void onComplete() {
 //RxJava把所有事件都当作队列处理
 //当被观察者不再发送onNext()普通事件的时候
 Log.e("RxJava_TAG", "onComplete");
 }
 };

 //订阅Subscribe(关系绑定)
 //当被观察者Observable被订阅时,即事件序列就会依照设定依次被触发
 //即观察者Observer会依次调用对应事件的复写方法从而响应事件
 //从而实现被观察者Observable调用了观察者Observer的回调方法(或者说被观察者向观察者的事件传递)
 TVplay.subscribe(viewer);
 }

 private static void extracted() {
 Observable.just("Hello", "World")
 .flatMap(new Function<String, ObservableSource<String>>() {
 @Override
 public ObservableSource<String> apply(String s) throws Throwable {
 Log.e(s, s);
 return Observable.just(s + s);
 }
 })
 .map(new Function<String, String>() {
 @Override
 public String apply(String s) throws Throwable {
 return s + " - " + s;
 }
 }).filter(new Predicate<String>() {
 @Override
 public boolean test(String s) throws Throwable {
 return s.contains("He");
 }
 })
 .subscribe(new Consumer<String>() {
 @Override
 public void accept(String s) throws Exception {
 System.err.println(s + " " + Thread.currentThread().getName());
 }
 });
 }
}