package Reactive;

import java.util.concurrent.Callable;

public class TimeConsumingService implements Callable<String> {

    private String service_name;
    private int wait_ms;

    public TimeConsumingService(String name, Integer waiting, String[] depandencies) {
        this.service_name = name;
        this.wait_ms = waiting;
    }

    @Override
    public String call() throws Exception {
                Thread.sleep(wait_ms);
                return String.format("service %s exec time is: %d ms", service_name,wait_ms);
    }
}

package Reactive;

import rx.Observable;
import rx.schedulers.Schedulers;


/**
 * Created by 80374563 on 2018/11/7.
 */
public class ReactiveDemo {

    /**
     * 案例研究：异步任务的依赖

     假设我们的程序需要五个 micro-service 协作完成计算任务，这些 micro-services 之间存在数据依赖关系：

     client <-  fc <- fa
     client <- fd <- fb
     client <- fe <- fb

     为了确保这些函数能并发执行，要点就是要构造足够线程，让没有依赖关系的服务在不同线程中执行。这里我们采用join 设计方法

     画出数据流图；
     选择流程图上的流程归并节点；
     为每条归并点的一条执行路径设计一个调度者（线程）；
     在归并点 merge 这些路径的流。
     *
     */

    public static void testAsyncCompositeJoin() {
        System.out.println("Prepare for execution：Async Composite Join");
        long startTime = System.currentTimeMillis(); //获取开始时间

        // Tasks oa -> oc,  both in the same thread 1.
        Observable<String> oa = Observable.just("oa").observeOn(Schedulers.io()).flatMap(
                soa -> Observable.fromCallable(new TimeConsumingService("fa", 1000, new String[]{}))
        );
        Observable<String> oc = oa.flatMap(
                (String res) -> {
                    System.out.println(res);
                    System.out.println("Executed At： " + (System.currentTimeMillis() - startTime) + "ms");
                    return Observable.fromCallable(
                            new TimeConsumingService("fc", 2000, new String[]{res}));
                });

        // tasks ob -> (od,oe),  ob, od, oe have special thread 2,3,4.
        Observable<String> ob = Observable.just("ob").observeOn(Schedulers.io()).flatMap(
                sob -> Observable.fromCallable(new TimeConsumingService("fb", 2000, new String[]{}))
        );
        Observable<String> od_oe = ob.flatMap(
                (String res) -> {
                    System.out.println(res);
                    System.out.println("Executed At： " + (System.currentTimeMillis() - startTime) + "ms");
                    Observable<String> od = Observable.just("od").observeOn(Schedulers.io()).flatMap(
                            sod -> Observable.fromCallable(new TimeConsumingService("fd", 1000, new String[]{res}))
                    );
                    Observable<String> oe = Observable.just("oe").observeOn(Schedulers.io()).flatMap(
                            sod -> Observable.fromCallable(new TimeConsumingService("fe", 1000, new String[]{res}))
                    );
                    return Observable.merge(od, oe);
                });

        System.out.println("Observable build： " + (System.currentTimeMillis() - startTime) + "ms");

        // tasks join oc,(od_oe) and subscribe
        Observable.merge(oc, od_oe).toBlocking().subscribe(
                (res) -> {
                    System.out.println(res);
                    System.out.println("Executed At： " + (System.currentTimeMillis() - startTime) + "ms");
                });

        System.out.println("End executed: " + (System.currentTimeMillis() - startTime) + "ms");
    }

    public static  void  main(String[] args){
        long st = System.currentTimeMillis();
        testAsyncCompositeJoin();
        System.out.println("done : " + (System.currentTimeMillis() - st) + " msecs");

    }

}