public interface DominoConverter<T, R, S extends Domino<T, ?>> extends Function1<Domino<T, R>, S> {
} public class RetrofitDominoConverter<T, R> implements
        DominoConverter<T, Triple<Boolean, Response<R>, Throwable>, RetrofitDomino<T, R>> {
    @Override
    public RetrofitDomino<T, R> call(Domino<T, Triple<Boolean, Response<R>, Throwable>> domino) {
        return new RetrofitDomino<T, R>(domino.getLabel(), domino.getPlayer());
    }
} public class RetrofitDominoConverter2<T, R, S> implements
        DominoConverter<T, Triple<Boolean, Pair<R, Response<S>>, Throwable>, RetrofitDomino2<T, R, S>> {
    @Override
    public RetrofitDomino2<T, R, S> call(Domino<T, Triple<Boolean, Pair<R, Response<S>>, Throwable>> domino) {
        return new RetrofitDomino2<T, R, S>(domino.getLabel(), domino.getPlayer());
    }
} public interface Action0 extends Action {
    void call();
} public interface Action1<T> extends Action {
    void call(T input);
} public interface Action2<T, R> extends Action {
    void call(T input1, R input2);
} public interface Function1<T, R> extends Function {
    R call(T input);
} public interface Function2<T, R, S> extends Function {
    S call(T input1, R input2);
} public interface TargetAction0<T> extends Action {
    void call(T t);
} public interface TargetAction1<T, R> extends Action {
    void call(T t, R input);
} public class EmptyOperator<T, R> implements CopyOnWriteArrayListFunction1<T, R> {

    public EmptyOperator() {}

    @Override
    public CopyOnWriteArrayList<R> call(CopyOnWriteArrayList<T> input) {
        return new CopyOnWriteArrayList<R>();
    }
} public class FilterOperator<T> implements CopyOnWriteArrayListFunction1<T, T> {

    private Function1<? super T, Boolean> filter;

    public FilterOperator(Function1<? super T, Boolean> filter) {
        this.filter = filter;
    }

    @Override
    public CopyOnWriteArrayList<T> call(CopyOnWriteArrayList<T> input) {
        CopyOnWriteArrayList<T> result = new CopyOnWriteArrayList<T>();
        for (T singleInput : input) {
            if (filter.call(singleInput)) {
                result.add(singleInput);
            }
        }
        return result;
    }
} public class FlatMapOperator<T, R> implements CopyOnWriteArrayListFunction1<T, R> {

    private Function1<? super T, List<R>> map;

    public FlatMapOperator(Function1<? super T, List<R>> map) {
        this.map = map;
    }

    @Override
    public CopyOnWriteArrayList<R> call(CopyOnWriteArrayList<T> input) {
        CopyOnWriteArrayList<R> result = new CopyOnWriteArrayList<R>();
        for (T singleInput : input) {
            result.addAll(map.call(singleInput));
        }
        return result;
    }
} public class ListIdentityOperator<T> extends IdentityOperator<List<T>> {
} public class MapOperator<T, R> implements CopyOnWriteArrayListFunction1<T, R> {

    private Function1<? super T, ? extends R> map;

    public MapOperator(Function1<? super T, ? extends R> map) {
        this.map = map;
    }

    @Override
    public CopyOnWriteArrayList<R> call(CopyOnWriteArrayList<T> input) {
        CopyOnWriteArrayList<R> result = new CopyOnWriteArrayList<R>();
        for (T singleInput : input) {
            result.add(map.call(singleInput));
        }
        return result;
    }
} public class MapOperator2<T, R, U> implements CopyOnWriteArrayListFunction1<T, R> {

    private static final TargetCenter TARGET_CENTER = TargetCenter.getInstance();

    private Class<U> clazz;

    private Function2<? super U, ? super T, ? extends R> map;

    public MapOperator2(Class<U> clazz, Function2<? super U, ? super T, ? extends R> map) {
        this.clazz = clazz;
        this.map = map;
    }

    @Override
    public CopyOnWriteArrayList<R> call(CopyOnWriteArrayList<T> input) {
        CopyOnWriteArrayList<R> result = new CopyOnWriteArrayList<R>();
        CopyOnWriteArrayList<Object> objects = TARGET_CENTER.getTargets(clazz);
        for (Object o : objects) {
            for (T singleInput : input) {
                result.add(map.call((U) o, singleInput));
            }
        }
        return result;
    }
} public class ReducerOperator<T, R> implements CopyOnWriteArrayListFunction1<T, R> {

    private Function1<List<T>, ? extends R> reducer;

    public ReducerOperator(Function1<List<T>, ? extends R> reducer) {
        this.reducer = reducer;
    }

    @Override
    public CopyOnWriteArrayList<R> call(CopyOnWriteArrayList<T> input) {
        CopyOnWriteArrayList<R> result = new CopyOnWriteArrayList<R>();
        result.add(reducer.call(input));
        return result;
    }
} public class RightRefinementOperator<T, R> implements Function2<T, R, R> {
    @Override
    public R call(T input1, R input2) {
        return input2;
    }
} public class BackgroundQueueScheduler extends Scheduler {

    private ExecutorService mExecutorService = Executors.newSingleThreadExecutor();

    @Override
    public void call(Runnable runnable) {
        mExecutorService.execute(runnable);
    }

} public class BackgroundScheduler extends Scheduler {

    private static ExecutorService sExecutorService = Executors.newCachedThreadPool();

    @Override
    public void call(Runnable runnable) {
        boolean isMainThread = Looper.getMainLooper() == Looper.myLooper();
        if (isMainThread) {
            sExecutorService.execute(runnable);
        } else {
            runnable.run();
        }
    }

} public class DefaultScheduler extends Scheduler {

    @Override
    public void call(Runnable runnable) {
        runnable.run();
    }

} public class NewThreadScheduler extends Scheduler {
    @Override
    public void call(Runnable runnable) {
        new Thread(runnable).start();
    }
} public abstract class Scheduler implements Action<Runnable> {

    private static final int STATE_RUNNING = 0;

    private static final int STATE_PAUSE = 1;

    private volatile int mState;

    public Scheduler() {
        mState = STATE_RUNNING;
    }

    public final void pause() {
        mState = STATE_PAUSE;
    }

    public final boolean isRunning() {
        return mState == STATE_RUNNING;
    }
} public class Player<T> {

    private final AugmentedListCanary<PlayerInputs> mInputs;

    private final DoubleKeyMap mStash;

    private Scheduler mScheduler;

    public Player(List<T> input) {
        mInputs = new AugmentedListCanary<PlayerInputs>();
        mInputs.add(new PlayerInputs((List<Object>) input, 0));
        mStash = new DoubleKeyMap();
        mScheduler = new DefaultScheduler();
    }

    public void setScheduler(Scheduler scheduler) {
        mScheduler = scheduler;
    }

    public Scheduler getScheduler() {
        return mScheduler;
    }

    public void prepare(Function function) {
        if (function instanceof StashFunction) {
            ((StashFunction) function).setStash(mStash);
        }
    }

    protected Runnable getRunnable(final Tile<T, ?> tile) {
        return new Runnable() {
            private int mIndex = mInputs.size() - 1;
            @Override
            public void run() {
                tile.call((CopyOnWriteArrayList<T>) mInputs.get(mIndex).getInputs());
            }
        };
    }

    //This method is not thread-safe! But we always call this in a single thread.
    public final <R> Player<R> playRunnable(List<? extends Runnable> runnables) {
        synchronized (this) {
            if (mScheduler.isRunning()) {
                int size = mInputs.size();
                for (Runnable runnable : runnables) {
                    mScheduler.call(new ScheduledRunnable<T>(this, runnable, size));
                }
            }
            return (Player<R>) this;
        }
    }

    public final <R> Player<R> playFunction(List<? extends Function1<CopyOnWriteArrayList<T>, CopyOnWriteArrayList<R>>> functions) {
        synchronized (this) {
            if (mScheduler.isRunning()) {
                int index = mInputs.add(new PlayerInputs(functions.size())) - 1;
                for (Function1<CopyOnWriteArrayList<T>, CopyOnWriteArrayList<R>> function : functions) {
                    mScheduler.call(new ScheduledRunnable<T>(this, new BlockingRunnable<T, R>(this, function, index), index));
                }
            }
            return (Player<R>) this;
        }
    }

    //This method is not thread-safe! But we always call this in a single thread.
    public final void play(Tile<T, ?> tile) {
        synchronized (this) {
            if (mScheduler.isRunning()) {
                playRunnable(Collections.singletonList(getRunnable(tile)));
            }
        }
    }

    public void appendAt(int index, final CopyOnWriteArrayList<Object> object) {
        mInputs.action(index, new Action<PlayerInputs>() {
            @Override
            public void call(PlayerInputs o) {
                o.getInputs().addAll(object);
                o.getFinishedNumber().getAndIncrement();
            }
        });
    }

    public final CopyOnWriteArrayList<Object> waitForFinishing() {
        int index = mInputs.size() - 1;
        return mInputs.get(index, new Condition<PlayerInputs>() {
            @Override
            public boolean satisfy(PlayerInputs o) {
                return o.getFinishedNumber().get() == o.getFunctionNumber();
            }
        }).getInputs();
    }

    public CopyOnWriteArrayList<Object> getInput(int index) {
        return mInputs.get(index).getInputs();
    }

    public AugmentedListCanary<PlayerInputs> getInputs() {
        return mInputs;
    }

} public class ThrottleScheduler extends Scheduler {

    private static final ScheduledExecutorService sExecutorService = Executors.newScheduledThreadPool(10);

    private static final ConcurrentHashMap<Object, Boolean> sRunningMap = new ConcurrentHashMap<Object, Boolean>();

    private final Scheduler mScheduler;

    private final Object mLabel;

    private final long mDuration;

    private final TimeUnit mUnit;

    public ThrottleScheduler(Scheduler scheduler, Object label, long duration, TimeUnit unit) {
        mScheduler = scheduler;
        mLabel = label;
        mDuration = duration;
        mUnit = unit;
    }

    @Override
    public void call(Runnable runnable) {
        Boolean running = sRunningMap.get(mLabel);
        if (running == null || running) {
            mScheduler.call(runnable);
            sRunningMap.put(mLabel, false);
            sExecutorService.schedule(new ResumingRunnable(mLabel), mDuration, mUnit);
        } else {
            pause();
        }
    }

    private static class ResumingRunnable implements Runnable {

        private final Object mLabel;

        ResumingRunnable(Object label) {
            mLabel = label;
        }

        @Override
        public void run() {
            sRunningMap.put(mLabel, true);
        }
    }

} public class UiThreadScheduler extends Scheduler {

    private static Handler sHandler = new Handler(Looper.getMainLooper());

    //下面这个改变比较重要！！！
    private ExecutorService mExecutor = Executors.newSingleThreadExecutor();

    private void scheduleInternal(Runnable runnable) {
        boolean isMainThread = Looper.getMainLooper() == Looper.myLooper();
        if (isMainThread) {
            if (Config.DEBUG) {
                System.out.println("post ui0 " + Thread.currentThread().getName());
            }
            runnable.run();
        } else {
            if (Config.DEBUG) {
                System.out.println("post ui1 " + Thread.currentThread().getName());
            }
            sHandler.post(runnable);
        }
    }

    @Override
    public void call(Runnable runnable) {
        if (runnable instanceof ScheduledRunnable) {
            final ScheduledRunnable scheduledRunnable = (ScheduledRunnable) runnable;
            if (!scheduledRunnable.inputSet()) {
                mExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                        if (Config.DEBUG) {
                            System.out.println("before wait in ui " + Thread.currentThread().getName());
                        }
                        scheduledRunnable.waitForInput();
                        if (Config.DEBUG) {
                            System.out.println("after wait in ui " + Thread.currentThread().getName());
                        }
                        sHandler.post(new Runnable() {
                            @Override
                            public void run() {
                                if (Config.DEBUG) {
                                    System.out.println("post ui2 " + Thread.currentThread().getName());
                                }
                                scheduledRunnable.getRunnable().run();
                            }
                        });
                    }
                });
            } else {
                scheduleInternal(scheduledRunnable.getRunnable());
            }
        } else {
            scheduleInternal(runnable);
        }
    }
} public abstract class AbstractRetrofitTask<T, R> extends Task<T, Response<R>, Throwable> {

    protected abstract Call<R> getCall(T t);

    protected Callback<R> getCallback() {
        return new Callback<R>() {
            @Override
            public void onResponse(Call<R> call, Response<R> response) {
                notifySuccess(response);
            }

            @Override
            public void onFailure(Call<R> call, Throwable t) {
                notifyFailure(t);
            }
        };
    }

    protected abstract void call(Call<R> call);

    @Override
    protected void onExecute(T input) {
        call(getCall(input));
    }
} public abstract class Task<T, R, S> {

    private TaskListener<R, S> mListener;

    public Task() {
    }

    public void setListener(TaskListener<R, S> listener) {
        mListener = listener;
    }

    protected abstract void onExecute(T input);

    public final void execute(T input) {
        onExecute(input);
    }

    public final void notifySuccess(R result) {
        if (mListener != null) {
            mListener.onSuccess(result);
        }
    }

    public final void notifyFailure(S error) {
        if (mListener != null) {
            mListener.onFailure(error);
        }
    }

    public interface TaskListener<R, S> {
        void onSuccess(R result);
        void onFailure(S error);
    }

} public class Pair<T1, T2> {
    public final T1 first;
    public final T2 second;

    @Deprecated
    public Pair(T1 first, T2 second) {
        this.first = first;
        this.second = second;
    }

    public static <T1, T2> Pair<T1, T2> create(T1 first, T2 second) {
        return new Pair<T1, T2>(first, second);
    }
} public class Triple<T1, T2, T3> {
    public final T1 first;
    public final T2 second;
    public final T3 third;

    @Deprecated
    public Triple(T1 first, T2 second, T3 third) {
        this.first = first;
        this.second = second;
        this.third = third;
    }

    public static <T1, T2, T3> Triple<T1, T2, T3> create(T1 first, T2 second, T3 third) {
        return new Triple<T1, T2, T3>(first, second, third);
    }
} public class DominoCenter {

    private static final String TAG = "DominoCenter";

    private static volatile DominoCenter sInstance = null;

    private final ConcurrentHashMap<Object, Domino<?, ?>> mDominoes;

    private DominoCenter() {
        mDominoes = new ConcurrentHashMap<Object, Domino<?, ?>>();
    }

    public static DominoCenter getInstance() {
        if (sInstance == null) {
            synchronized (DominoCenter.class) {
                if (sInstance == null) {
                    sInstance = new DominoCenter();
                }
            }
        }
        return sInstance;
    }

    public void commit(Domino<?, ?> domino) {
        Object label = domino.getLabel();
        if (label == null) {
            throw new UnsupportedOperationException("Domino label cannot be null. "
                    + "Please note that Domino created by Shelly.createDomino(Class<T>) can only "
                    + "be used temporarily and cannot be committed.");
        }
        if (mDominoes.put(label, domino) != null) {
            Log.w(TAG, "Domino name duplicate! Check whether you have commit a domino with the same label before.");
        }
    }

    public <T> void play(Object label, CopyOnWriteArrayList<T> input) {
        Domino<T, ?> domino = (Domino<T, ?>) mDominoes.get(label);
        if (domino == null) {
            throw new IllegalStateException("There is no domino labeled '" + label + "'.");
        }
        domino.play(input);
    }

    public <T, R> Domino<T, R> getDomino(Object label) {
        return (Domino<T, R>) mDominoes.get(label);
    }

} public interface Tile<T, R> extends Function1<List<T>, Player<R>> {

} public class TargetCenter {

    private static volatile TargetCenter sInstance = null;

    private final ConcurrentHashMap<Class<?>, CopyOnWriteArrayList<Object>> mTargets;

    private TargetCenter() {
        mTargets = new ConcurrentHashMap<Class<?>, CopyOnWriteArrayList<Object>>();
    }

    public static TargetCenter getInstance() {
        if (sInstance == null) {
            synchronized (TargetCenter.class) {
                if (sInstance == null) {
                    sInstance = new TargetCenter();
                }
            }
        }
        return sInstance;
    }

    public void register(Object target) {
        synchronized (mTargets) {
            for (Class<?> clazz = target.getClass(); clazz != null; clazz = clazz.getSuperclass()) {
                CopyOnWriteArrayList<Object> targets = mTargets.get(clazz);
                if (targets == null) {
                    mTargets.putIfAbsent(clazz, new CopyOnWriteArrayList<Object>());
                    targets = mTargets.get(clazz);
                }
                targets.add(target);
            }
        }
    }

    public void unregister(Object target) {
        synchronized (mTargets) {
            for (Class<?> clazz = target.getClass(); clazz != null; clazz = clazz.getSuperclass()) {
                CopyOnWriteArrayList<Object> targets = mTargets.get(clazz);
                if (targets == null) {
                    return;
                }
                int size = targets.size();
                for (int i = 0; i < size; ++i) {
                    if (targets.get(i) == target) {
                        targets.remove(i);
                        --i;
                        --size;
                    }
                }
                if (targets.isEmpty()) {
                    mTargets.remove(clazz);
                }
            }
        }
    }

    public boolean isRegistered(Object target) {
        Class<?> clazz = target.getClass();
        CopyOnWriteArrayList<Object> targets = mTargets.get(clazz);
        return targets != null && targets.contains(target);
    }

    public CopyOnWriteArrayList<Object> getTargets(Class<?> clazz) {
        CopyOnWriteArrayList<Object> targets = mTargets.get(clazz);
        return targets == null ? new CopyOnWriteArrayList<Object>() : targets;
    }
} public class TaskFunction<T, R1, R2, U1, U2>
        implements Function1<T, Triple<Boolean, R2, U2>>, Task.TaskListener<R1, U1> {

    private Task<T, R1, U1> mTask;

    private Function2<T, R1, R2> mFunc1;

    private Function2<T, U1, U2> mFunc2;

    private T mInput;

    private volatile ObjectCanary<ResultWrapper<R2, U2>> mResultWrapper;

    public TaskFunction(Task<T, R1, U1> task, Function2<T, R1, R2> func1, Function2<T, U1, U2> func2) {
        task.setListener(this);
        mTask = task;
        mFunc1 = func1;
        mFunc2 = func2;
        mResultWrapper = new ObjectCanary<ResultWrapper<R2, U2>>();
    }

    @Override
    public void onFailure(final U1 error) {
        mResultWrapper.action(new Action<ResultWrapper<R2, U2>>() {
            @Override
            public void call(ResultWrapper<R2, U2> o) {
                o.setError(mFunc2.call(mInput, error));
            }
        });
    }

    @Override
    public void onSuccess(final R1 result) {
        mResultWrapper.action(new Action<ResultWrapper<R2, U2>>() {
            @Override
            public void call(ResultWrapper<R2, U2> o) {
                o.setResult(mFunc1.call(mInput, result));
            }
        });
    }

    @Override
    public Triple<Boolean, R2, U2> call(T input) {
        mInput = input;
        mResultWrapper.set(new ResultWrapper<R2, U2>());
        mTask.execute(input);
        mResultWrapper.wait(new Condition<ResultWrapper<R2, U2>>() {
            @Override
            public boolean satisfy(ResultWrapper<R2, U2> o) {
                return o.getFlag() != -1;
            }
        });
        return Triple.create(
                mResultWrapper.calculate(new Function<ResultWrapper<R2, U2>, Boolean>() {
                    @Override
                    public Boolean call(ResultWrapper<R2, U2> o) {
                        return o.getFlag() == 1;
                    }
                }),
                mResultWrapper.calculate(new Function<ResultWrapper<R2, U2>, R2>() {
                    @Override
                    public R2 call(ResultWrapper<R2, U2> o) {
                        return o.getResult();
                    }
                }),
                mResultWrapper.calculate(new Function<ResultWrapper<R2, U2>, U2>() {
                    @Override
                    public U2 call(ResultWrapper<R2, U2> o) {
                        return o.getError();
                    }
                }));
    }

    private static class ResultWrapper<T, R> {
        private volatile int mFlag;
        private volatile T mResult;
        private volatile R mError;

        ResultWrapper() {
            mFlag = -1;
            mResult = null;
            mError = null;
        }

        public void setError(R error) {
            mError = error;
            mFlag = 0;
        }

        public void setResult(T result) {
            mResult = result;
            mFlag = 1;
        }

        public R getError() {
            return mError;
        }

        public int getFlag() {
            return mFlag;
        }

        public T getResult() {
            return mResult;
        }
    }
}
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import xiaofei.library.shelly.domino.converter.DominoConverter;
import xiaofei.library.shelly.domino.converter.RetrofitDominoConverter;
import xiaofei.library.shelly.domino.converter.RetrofitDominoConverter2;
import xiaofei.library.shelly.function.Action0;
import xiaofei.library.shelly.function.Action1;
import xiaofei.library.shelly.function.Action2;
import xiaofei.library.shelly.function.Function1;
import xiaofei.library.shelly.function.Function2;
import xiaofei.library.shelly.function.TargetAction0;
import xiaofei.library.shelly.function.TargetAction1;
import xiaofei.library.shelly.operator.EmptyOperator;
import xiaofei.library.shelly.operator.FilterOperator;
import xiaofei.library.shelly.operator.FlatMapOperator;
import xiaofei.library.shelly.operator.ListIdentityOperator;
import xiaofei.library.shelly.operator.MapOperator;
import xiaofei.library.shelly.operator.MapOperator2;
import xiaofei.library.shelly.operator.ReducerOperator;
import xiaofei.library.shelly.operator.RightRefinementOperator;
import xiaofei.library.shelly.scheduler.BackgroundQueueScheduler;
import xiaofei.library.shelly.scheduler.BackgroundScheduler;
import xiaofei.library.shelly.scheduler.DefaultScheduler;
import xiaofei.library.shelly.scheduler.NewThreadScheduler;
import xiaofei.library.shelly.scheduler.Scheduler;
import xiaofei.library.shelly.util.Player;
import xiaofei.library.shelly.scheduler.ThrottleScheduler;
import xiaofei.library.shelly.scheduler.UiThreadScheduler;
import xiaofei.library.shelly.task.AbstractRetrofitTask;
import xiaofei.library.shelly.task.Task;
import xiaofei.library.shelly.tuple.Pair;
import xiaofei.library.shelly.tuple.Triple;
import xiaofei.library.shelly.util.DominoCenter;
import xiaofei.library.shelly.util.Tile;
import xiaofei.library.shelly.util.TargetCenter;
import xiaofei.library.shelly.util.TaskFunction;
     */
    public <U> Domino<T, U> merge(final Domino<? super R, ? extends U>[] dominoes) {
        return new Domino<T, U>(mLabel, new Tile<T, U>() {
            @Override
            public Player<U> call(List<T> input) {
                final Player<R> player = mTile.call(input);
                List<Function1<CopyOnWriteArrayList<R>, CopyOnWriteArrayList<U>>> functions =
                        new ArrayList<Function1<CopyOnWriteArrayList<R>, CopyOnWriteArrayList<U>>>();
                for (final Domino<? super R, ? extends U> domino : dominoes) {
                    functions.add(new Function1<CopyOnWriteArrayList<R>, CopyOnWriteArrayList<U>>() {
                        @Override
                        public CopyOnWriteArrayList<U> call(CopyOnWriteArrayList<R> input) {
                            Player<U> player = ((Domino<R, U>) domino).mTile.call(input);
                            return (CopyOnWriteArrayList<U>) player.waitForFinishing();
                        }
                    });
                }
                return player.playFunction(functions);
            }
        });
    }

    public <S1, S2, V> Domino<T, V> combine(Domino<? super R, S1> domino1, Domino<? super R, S2> domino2,
                                            final Function2<? super S1, ? super S2, ? extends V> combiner) {
        /**
         * 想实现的效果是domino1和domino2分开运行，结果经过combiner结合，得到一堆新的结果
         * 为了实现方便，做如下变化：
         * 1、将domino1与domino2分开运行，返回结果变为一个list，这个单独作为一个结果：domino->reduce->merge
         * 2、将这个结果进行合并：reduce
         * 3、将这个结果展开：flatMap
         */
        return merge(
                domino1.reduce(new Function1<List<S1>, Pair<Integer, List<Object>>>() {
                    @Override
                    public Pair<Integer, List<Object>> call(List<S1> input) {
                        return Pair.create(1, (List<Object>) input);
                    }
                }),
                domino2.reduce(new Function1<List<S2>, Pair<Integer, List<Object>>>() {
                    @Override
                    public Pair<Integer, List<Object>> call(List<S2> input) {
                        return Pair.create(2, (List<Object>) input);
                    }
                }))
                .reduce(new Function1<List<Pair<Integer, List<Object>>>, List<V>>() {
                    @Override
                    public List<V> call(List<Pair<Integer, List<Object>>> input) {
                        if (input.size() != 2) {
                            throw new IllegalStateException("Unknown error! Please report this to Xiaofei.");
                        }
                        List<V> result = new ArrayList<V>();
                        List<Object> input1, input2;
                        if (input.get(0).first == 1) {
                            input1 = input.get(0).second;
                            input2 = input.get(1).second;
                        } else {
                            input1 = input.get(1).second;
                            input2 = input.get(0).second;
                        }
                        for (Object o1 : input1) {
                            for (Object o2 : input2) {
                                result.add(combiner.call((S1) o1, (S2) o2));
                            }
                        }
                        return result;
                    }
                })
                .flatMap(new ListIdentityOperator<V>());
    }

    public <S1, S2, U1, U2, V> Domino<T, V> combineTask(
            TaskDomino<? super R, S1, U1> taskDomino1,
            TaskDomino<? super R, S2, U2> taskDomino2,
            final Function2<? super S1, ? super S2, ? extends V> combiner) {
        return combine(
                taskDomino1.map(new Function1<Triple<Boolean,S1,U1>, S1>() {
                    @Override
                    public S1 call(Triple<Boolean, S1, U1> input) {
                        return input.first ? input.second : null;
                    }
                }),
                taskDomino2.map(new Function1<Triple<Boolean,S2,U2>, S2>() {
                    @Override
                    public S2 call(Triple<Boolean, S2, U2> input) {
                        return input.first ? input.second : null;
                    }
                }),
                combiner);
    }

    private Domino<T, R> schedule(final Scheduler scheduler) {
        return new Domino<T, R>(mLabel, new Tile<T, R>() {
            @Override
            public Player<R> call(List<T> input) {
                Player<R> player = mTile.call(input);
                player.setScheduler(scheduler);
                return player;
            }
        });
    }

    public Domino<T, R> background() {
        return schedule(new BackgroundScheduler());
    }

    /**
     * For unit test only.
     */
    Domino<T, R> newThread() {
        return schedule(new NewThreadScheduler());
    }

    /**
     * For unit test only.
     */
    Domino<T, R> defaultScheduler() {
        return schedule(new DefaultScheduler());
    }

    public Domino<T, R> uiThread() {
        return schedule(new UiThreadScheduler());