En java, comment puis-je traiter CompletableFutures et obtenir le premier résultat désirable qui se termine?

Question

Normalement, avec un CompletableFuture, j'appelle thenApply ou une autre méthode pour faire quelque chose dès que le résultat est disponible. Cependant, j'ai maintenant une situation où je veux traiter des résultats jusqu'à ce que je reçoive un résultat positif, puis ignorer tous les autres résultats.

Si je voulais juste prendre le premier résultat disponible, je pourrais utiliser CompletableFuture.anyOf (bien que je déteste avoir à convertir une liste en tableau pour appeler anyOf). Mais ce n'est pas ce que je veux. Je veux prendre le premier résultat et s'il n'a pas un résultat désirable, alors je veux traiter le deuxième résultat disponible et ainsi de suite jusqu'à ce que j'obtienne un résultat souhaitable.

Voici un exemple simple qui passe par tous les résultats et retourne la première valeur qu'il trouve qui est supérieure à 9. (Notez que ce n'est pas ma tâche réelle. Ceci est juste un exemple simple.)

public Integer findFirstGt9(List<CompletableFuture<Integer>> results) { 
    for(CompletableFuture<Integer> result : results) { 
     Integer v = result.get(); 
     if(v > 9) 
      return v; 
    } 
    return null; 
} 

Bien sûr, cet exemple passe par les résultats du début, pas en regardant les résultats à mesure qu'ils se terminent. Donc, voici celui qui accomplit ce que je veux, mais avec un code beaucoup plus compliqué.

public Integer findFirstGt9(List<CompletableFuture<Integer>> results) { 
    AtomicInteger finalResult = new AtomicInteger(); 
    CountDownLatch latch = new CountDownLatch(results.size()); 
    for(CompletableFuture<Integer> result : results) { 
     result.whenComplete((v,e) -> { 
      if(e!=null) { 
       Logger.getLogger(getClass()).error("",e); 
      } else if(v > 9) { 
       finalResult.set(v); 
       while(latch.getCount() > 0) 
        latch.countDown(); 
       return; 
      } 
      latch.countDown(); 
     }); 
    } 
    latch.await(); 

    if(finalResult.get() > 9) 
     return finalResult.get(); 
    return null; 
}  

Y a-t-il une API où je peux simplement faire cela?

public Integer findFirstGt9(List<CompletableFuture<Integer>> results) { 
    Iterator<Integer> resultIt = getResultsAsAvailable(results); 
    for(; resultIt.hasNext();) { 
     Integer v = resultIt.next(); 
     if(v > 9) 
      return v; 
    } 
    return null; 
} 

Ou encore mieux:

public Integer findFirstGt9(List<CompletableFuture<Integer>> results) { 
    return getFirstMatch(results, r -> {return r > 9;}); 
} 

Answer 1

Vous pouvez utiliser la solution suivante:

public static <T> CompletableFuture<T> anyMatch(
    List<? extends CompletionStage<? extends T>> l, Predicate<? super T> criteria) { 

    CompletableFuture<T> result=new CompletableFuture<>(); 
    Consumer<T> whenMatching=v -> { if(criteria.test(v)) result.complete(v); }; 
    CompletableFuture.allOf(l.stream() 
     .map(f -> f.thenAccept(whenMatching)).toArray(CompletableFuture<?>[]::new)) 
    .whenComplete((ignored, t) -> 
     result.completeExceptionally(t!=null? t: new NoSuchElementException())); 
    return result; 
} 

Le principe de base est le même que dans Pillar’s answer, cependant, il y a quelques différences:

• La signature générique est plus souple.
• La création de la matrice nécessaire pour CompletableFuture.allOf est combinée avec l'enregistrement de l'action de suivi sur les contrats à terme source. En tant qu'effet secondaire, le gestionnaire de l'action allOf dépend de l'achèvement de toutes les tentatives pour compléter le résultat, plutôt que les contrats à terme d'origine uniquement. Cela rend la dépendance réellement souhaitée explicite. De cette façon, cela fonctionnerait même quand nous remplacerons tous thenAccept avec thenAcceptAsync s.
• Cette solution se termine par NoSuchElementException plutôt que de renvoyer null dans le cas où aucun résultat ne répond aux critères. Si au moins un futur est achevé exceptionnellement et qu'il n'y a pas de complétion réussie avec un résultat correspondant, l'une des exceptions survenues est relayée.

Vous pouvez essayer avec

List<CompletableFuture<Integer>> list=Arrays.asList(
    CompletableFuture.supplyAsync(()->5), 
    CompletableFuture.supplyAsync(()->{throw new RuntimeException(); }), 
    CompletableFuture.supplyAsync(()->42), 
    CompletableFuture.completedFuture(0) 
); 
anyMatch(list, i -> i>9) 
    .thenAccept(i->System.out.println("got "+i)) 
    // optionally chain with: 
    .whenComplete((x,t)->{ if(t!=null) t.printStackTrace(); }); 

Answer 2

Je ne sais pas d'une telle API dans le JDK ou ailleurs. Vous pouvez rouler le vôtre.

Vous pouvez profiter du fait que CompletableFuture#complete (et completeExceptionally) ne fait rien si le futur est déjà terminé.

Si pas déjà terminée, fixe la valeur retournée par get() et méthodes liés à la valeur donnée.

Créer une nouvelle résultat finalCompletableFuture. Ajouter une continuation à chacun de vos contrats à terme qui tente de complete ce résultat final si votre condition s'applique. Cet avenir se terminera avec le premier succès. Cependant, si aucun ne réussit, vous avez apparemment besoin de null. Vous pouvez créer un CompletableFuture avec allOf pour tenter également de complete le résultat final avec null.

Quelque chose comme

public static <T> CompletableFuture<T> firstOrNull(List<CompletableFuture<T>> futures, Predicate<T> condition) { 
    CompletableFuture<T> finalResult = new CompletableFuture<>(); 
    // attempt to complete on success 
    futures.stream().forEach(future -> future.thenAccept(successResult -> { 
     if (condition.test(successResult)) 
      finalResult.complete(successResult); 
    })); 
    CompletableFuture<?> all = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])); 
    all.thenRun(() -> { 
     finalResult.complete(null); 
    }); 
    return finalResult; 
} 

Vous payez les frais généraux d'invocations no-op.

Vous pouvez remplacer null par une valeur par défaut appropriée ou gérer les exceptions différemment (completeExceptionally dès qu'une erreur se produit). Vous devrez utiliser whenComplete ou handle au lieu du thenAccept ci-dessus pour accéder au Exception.