package com.gh.gamecenter.common.retrofit;

import java.util.concurrent.CountDownLatch;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;

/**
 * Created by LGT on 2016/11/9.
 * Observable 工具类
 */
public class ObservableUtil {

    // 用于计算任务，如事件循环或和回调处理
    public static <T> void computation(ObservableOnSubscribe<T> onSubscribe, Consumer<T> consumer) {
        if (consumer == null) {
            consumer = t -> {

            };
        }
        Observable.create(onSubscribe)
                .subscribeOn(Schedulers.computation())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(consumer);
    }

    // 用于IO密集型任务，如异步阻塞IO操作
    public static <T> void io(ObservableOnSubscribe<T> onSubscribe, Consumer<T> consumer) {
        if (consumer == null) {
            consumer = t -> {

            };
        }
        Observable.create(onSubscribe)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(consumer);
    }

    public static <T> CountDownLatch latch(int size, Consumer<T> consumer, T t) {
        final CountDownLatch latch = new CountDownLatch(size);
        Observable.create(new ObservableOnSubscribe<T>() {
                    @Override
                    public void subscribe(ObservableEmitter<T> emitter) {
                        try {
                            latch.await();
                            emitter.onNext(t);
                        } catch (Throwable e) {
                            e.printStackTrace();
                            if (e instanceof InterruptedException) {
                                Thread.currentThread().interrupt();
                            }
                        }
                        emitter.onComplete();
                    }
                }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(consumer);
        return latch;
    }


}
