FutureOps.java

package com.dimitarg.demo;

import io.netty.util.concurrent.EventExecutor;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.Promise;

import java.util.function.Function;

/**
 * Created by fmap on 25.11.16.
 */
public class FutureOps
{

    private static final Exception CANCEL = new Exception()
    {
        @Override public synchronized Throwable fillInStackTrace()
        {
            return this;
        }
    };

    private final EventExecutor ex;

    public FutureOps(EventExecutor ex)
    {
        this.ex = ex;
    }

    public <A, B> Future<B> map(Future<A> future, Function<A, B> f)
    {
        Promise<B> result = ex.newPromise();

        failOrEffect(future, result, () -> result.setSuccess(f.apply(future.getNow())));

        return result;
    }

    public <A, B> Future<B> bind(Future<A> future, Function<A, Future<B>> f)
    {
        Promise<B> result = ex.newPromise();

        failOrEffect(future, result, () -> {
            Future<B> next = f.apply(future.getNow());
            failOrEffect(next, result, () -> result.setSuccess(next.getNow()));

        });

        return result;
    }

    public <A> Future<A> unit(A a)
    {
        Promise<A> result = ex.newPromise();
        result.setSuccess(a);
        return result;
    }

    public static void failOrEffect(Future<?> upstream, Promise<?> promise, Effect f)
    {
        upstream.addListener(x ->
        {
            if (upstream.isCancelled())
            {
                promise.setFailure(CANCEL);
            }
            else if (upstream.cause() != null)
            {
                promise.setFailure(x.cause());
            }
            else
            {
                f.run();
            }
        });
    }

    public interface Effect
    {
        void run();
    }

}