UFuture.java

package example.playground.ufuture;

import com.google.common.util.concurrent.FluentFuture;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.SettableFuture;
import lombok.EqualsAndHashCode;
import lombok.RequiredArgsConstructor;
import lombok.Value;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.function.Function;

interface UFuture<T> {

  // Base methods

  UFutureData<T> toData();

  <U> UFuture<U> map(Function<? super T, ? extends U> mapper);

  <U> UFuture<U> flatMap(Function<?super T, UFuture<U>> next);

  // Factories

  static <T> UFuture<T> completed(T value) {
    return new UFutureData.Pure<>(value);
  }

  static <T> UFuture<T> failed(Throwable cause) {
    return new UFutureData.Failed<>(cause);
  }

  static <T> UFuture<T> from(CompletionStage<T> stage) {
    return new UFutureData.FromCompletionStage<>(stage);
  }
}

abstract class UFutureData<T> implements UFuture<T> {

  private UFutureData() {}

  // Base methods

  public UFutureData<T> toData() {
    return this;
  }

  public <U> UFuture<U> map(Function<? super T, ? extends U> mapper) {
    return new Map<>(this, mapper);
  }

  public <U> UFuture<U> flatMap(Function<? super T, UFuture<U>> next) {
    return new FlatMap<>(this, next);
  }

  // Reified Base methods

  @Value
  @EqualsAndHashCode(callSuper = false)
  static class Pure<T> extends UFutureData<T> {
    T value;

    @Override
    public <R> R accept(Visitor<T, R> visitor) {
      return visitor.visit(this);
    }
  }
  @Value
  @EqualsAndHashCode(callSuper = false)
  static class Failed<T> extends UFutureData<T> {
    Throwable cause;

    @Override
    public <R> R accept(Visitor<T, R> visitor) {
      return visitor.visit(this);
    }
  }
  @Value
  @EqualsAndHashCode(callSuper = false)
  static class Map<T, U> extends UFutureData<U> {
    UFuture<T> target;
    Function<? super T, ? extends U> mapper;

    @Override
    public <R> R accept(Visitor<U, R> visitor) {
      return visitor.visit(this);
    }
  }
  @Value
  @EqualsAndHashCode(callSuper = false)
  static class FlatMap<T, U> extends UFutureData<U> {
    UFuture<T> target;
    Function<? super T, UFuture<U>> next;

    @Override
    public <R> R accept(Visitor<U, R> visitor) {
      return visitor.visit(this);
    }
  }
  @Value
  @EqualsAndHashCode(callSuper = false)
  static class FromCompletionStage<T> extends UFutureData<T> {
    CompletionStage<T> stage;

    @Override
    public <R> R accept(Visitor<T, R> visitor) {
      return visitor.visit(this);
    }
  }

  // Visitor

  public abstract <R> R accept(Visitor<T, R> visitor);

  interface Visitor<T, R> {
    R visit(Pure<T> instance);
    R visit(Failed<T> instance);
    <X> R visit(Map<X, T> instance);
    <X> R visit(FlatMap<X, T> instance);
    R visit(FromCompletionStage<T> instance);
  }
}

@RequiredArgsConstructor
class CompletionStageBridge {

  private final Executor executor;

  <T> CompletionStage<T> interpret(UFuture<T> future) {
    return future.toData().accept(new UFutureData.Visitor<>() {
      @Override
      public CompletionStage<T> visit(UFutureData.Pure<T> instance) {
        return CompletableFuture.completedFuture(instance.getValue());
      }

      @Override
      public CompletionStage<T> visit(UFutureData.Failed<T> instance) {
        return CompletableFuture.failedFuture(instance.getCause());
      }

      @Override
      public <X> CompletionStage<T> visit(UFutureData.Map<X, T> instance) {
        return interpret(instance.getTarget()).thenApplyAsync(instance.getMapper(), executor);
      }

      @Override
      public <X> CompletionStage<T> visit(UFutureData.FlatMap<X, T> instance) {
        return interpret(instance.getTarget())
                .thenComposeAsync(x -> interpret(instance.getNext().apply(x)), executor);
      }

      @Override
      public CompletionStage<T> visit(UFutureData.FromCompletionStage<T> instance) {
        return instance.getStage();
      }
    });
  }
}

@RequiredArgsConstructor
class FluentFutureBridge {

  private final Executor executor;

  <T> FluentFuture<T> interpret(UFuture<T> future) {
    return future.toData().accept(new UFutureData.Visitor<>() {
      @Override
      public FluentFuture<T> visit(UFutureData.Pure<T> instance) {
        return FluentFuture.from(Futures.immediateFuture(instance.getValue()));
      }

      @Override
      public FluentFuture<T> visit(UFutureData.Failed<T> instance) {
        return FluentFuture.from(Futures.immediateFailedFuture(instance.getCause()));
      }

      @Override
      public <X> FluentFuture<T> visit(UFutureData.Map<X, T> instance) {
        return interpret(instance.getTarget())
                .transform(instance.getMapper()::apply, executor);
      }

      @Override
      public <X> FluentFuture<T> visit(UFutureData.FlatMap<X, T> instance) {
        return interpret(instance.getTarget())
                .transformAsync(
                        x -> interpret(instance.getNext().apply(x)), executor);
      }

      @Override
      public FluentFuture<T> visit(UFutureData.FromCompletionStage<T> instance) {
        SettableFuture<T> dest = SettableFuture.create();
        instance.getStage().whenComplete((t, ex) -> {
          if (t != null) {
            dest.set(t);
          } else if (ex != null) {
            dest.setException(ex);
          } else {
            throw new IllegalStateException("what");
          }
        });
        return FluentFuture.from(dest);
      }
    });
  }
}

public class Program {
  public static void main(String[] args) throws Exception {
    var fut = UFuture.completed(2)
            .map(x -> 10 * x)
            .flatMap(x -> UFuture.completed(5 + x));

    var executor = Executors.newWorkStealingPool();

    var stageBridge = new CompletionStageBridge(executor);
    var fluentBridge = new FluentFutureBridge(executor);

    System.out.println(stageBridge.interpret(fut).toCompletableFuture().get());

    System.out.println(fluentBridge.interpret(fut).get());
  }
}