package ai.verta.modeldb.common.futures;

import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;

public class InternalFuture<T> {
  private CompletionStage<T> stage;

  private InternalFuture() {}

  public static <T> InternalFuture<List<T>> sequence(final List<InternalFuture<T>> futures) {
    var completableFuture =
            CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]))
                    .thenApply(
                            unit ->
                                    futures.stream()
                                            .map(f -> f.stage.toCompletableFuture().join())
                                            .collect(Collectors.toList()));
    return InternalFuture.from(completableFuture);
  }

  public static <R> InternalFuture<R> from(CompletionStage<R> other) {
    var ret = new InternalFuture<R>();
    ret.stage = other;
    return ret;
  }

  public static <R> InternalFuture<R> completedInternalFuture(R value) {
    var ret = new InternalFuture<R>();
    ret.stage = CompletableFuture.completedFuture(value);
    return ret;
  }

  public <U> InternalFuture<U> thenCompose(
      Function<? super T, InternalFuture<U>> fn, Executor executor) {
    return from(
        stage.thenComposeAsync(fn.andThen(internalFuture -> internalFuture.stage), executor));
  }

  public <U> InternalFuture<U> thenApply(Function<? super T, ? extends U> fn, Executor executor) {
    return from(stage.thenApplyAsync(fn, executor));
  }

  public InternalFuture<Void> thenAccept(Consumer<? super T> action, Executor executor) {
    return from(stage.thenAcceptAsync(action, executor));
  }

  public <U> InternalFuture<Void> thenRun(Runnable runnable, Executor executor) {
    return from(stage.thenRunAsync(runnable, executor));
  }

  public static <U> InternalFuture<Void> runAsync(Runnable runnable, Executor executor) {
    return from(CompletableFuture.runAsync(runnable, executor));
  }

  public static <U> InternalFuture<U> supplyAsync(Supplier<U> supplier, Executor executor) {
    return from(CompletableFuture.supplyAsync(supplier, executor));
  }

  public static <U> InternalFuture<U> failedStage(Throwable ex) {
    return from(CompletableFuture.failedFuture(ex));
  }

  public InternalFuture<T> whenComplete(
      BiConsumer<? super T, ? super Throwable> action, Executor executor) {
    return from(stage.whenCompleteAsync(action, executor));
  }

  public T get() throws ExecutionException, InterruptedException {
    return stage.toCompletableFuture().get();
  }

  public CompletionStage<T> toCompletionStage() {
    return stage;
  }
}

@@ -35,16 +35,44 @@
 tures;%0A%0A
+import java.util.ArrayList;%0A
 import j
@@ -82,24 +82,24 @@
 .util.List;%0A
-
 import java.
@@ -548,16 +548,69 @@
 e() %7B%7D%0A%0A
+  // Convert a list of futures to a future of a list%0A
   public
@@ -691,24 +691,43 @@
  futures
+, Executor executor
 ) %7B%0A    
 var comp
@@ -722,13 +722,15 @@
     
-var c
+final C
 ompl
@@ -745,85 +745,103 @@
 ture
- =%0A            CompletableFuture.allOf(futures.toArray(new CompletableFuture%5B
+%3CList%3CT%3E%3E promise = new CompletableFuture%3C%3E();%0A    final ArrayList%3CT%3E values = new ArrayList%3C%3E(
 futu
@@ -854,42 +854,31 @@
 ze()
-%5D)
 )
+;
 %0A    
-                .thenApply
+executor.execute
 (%0A  
@@ -879,33 +879,40 @@
 te(%0A            
- 
+() -%3E %7B%0A
                u
@@ -913,16 +913,13 @@
     
- unit -%3E
+try %7B
 %0A   
@@ -923,35 +923,67 @@
                 
-   
+for (final var future : futures) %7B%0A
                 
@@ -987,24 +987,63 @@
     
+ values.add(
 future
-s
 .st
-ream()
+age.toCompletableFuture().join());
 %0A   
@@ -1047,32 +1047,34 @@
                 
+%7D%0A
                 
@@ -1077,84 +1077,122 @@
     
-            .map(f -%3E f.stage.toCompletableFuture().join())%0A                
+promise.complete(values);%0A              %7D catch (Throwable t) %7B%0A                promise.completeExceptionally(t);%0A
     
@@ -1193,34 +1193,34 @@
 ;%0A              
-  
+%7D%0A
             .col
@@ -1219,37 +1219,9 @@
     
-.collect(Collectors.toList())
+%7D
 );%0A 
@@ -1254,24 +1254,14 @@
 rom(
-completableFutur
+promis
 e);%0A

Fix sequencer runtime error (#2114)