Comment puis-je combiner DeferredResult avec la liste ListenableFuture?

Question

J'ai une liste de Guava ListenableFuture instances et le printemps DeferredResult. Je veux définir le résultat pour le premier succès de la liste ou si le délai d'expiration n'a pas encore expiré pour obtenir un résultat positif de tous les contrats à terme. Voici mon essai:

DeferredResult<String> foo() { 
    DeferredResult<String> result = new DeferredResult<>(3000L); 

    List<String> resps = newArrayList(); 
    List<ListenableFuture<String>> fList = ... 

    fList.forEach(f -> Futures.addCallback(f, new FutureCallback<String>() { 
     @Override 
     public void onSuccess(String resp) { 
      resps.add(resp); 
     } 

     @Override 
     public void onFailure(Throwable t) { 
      // NOP 
     } 
    })); 

    ListenableFuture<List<String>> f0 = Futures.successfulAsList(fList); 

    Futures.addCallback(f0, new FutureCallback<List<String>>() { 
     @Override 
     public void onSuccess(List<String> r) { 
      if (!result.hasResult()) { 
       result. 
        if (r != null) { 
         result.setResult(...); 
        } 
       } 
      } 

      @Override 
      public void onFailure(Throwable t) { 
       // NOP 
      } 
     }); 

    return result; 
} 

Mon code ne fonctionne pas parce qu'il attend un résultat de tous les contrats à terme, mais je dois le résultat le plus rapide par rapport à délai d'attente DeferredResult. Comment puis-je le réparer?

Answer 1

DeferredResult<String> foo() { 
    DeferredResult<String> result = new DeferredResult<>(3000L); 

    List<ListenableFuture<String>> fList = ... 

    ExecutorService serializingExecutor = Executors.newSingleThreadExecutor(); 
    fList.forEach(
     f -> 
      Futures.addCallback(
       f, 
       new FutureCallback<String>() { 
       @Override 
       public void onSuccess(String value) { 
        if (!result.hasResult()) { 
        result.setResult(value); 
        } 
       } 

       @Override 
       public void onFailure(Throwable throwable) {} 
       }, 
       // To avoid race in the callback: 
       serializingExecutor)); 

    return result; 
} 

Answer 2

Cela ne retourne un objet de wrapper avec la liste des résultats des contrats à terme qui ont réussi à terminer avant délai et indiquer si tous fini (allFinished propriété).

Dans votre code de manipulation, vous pouvez ensuite vérifier si tout est terminé, et renvoyer la liste entière ou juste un premier article (ou aucun, si aucun Future terminé).

DeferredResult<ResultWrapper> foo() { 
     DeferredResult<ResultWrapper> result = new DeferredResult<>(3000L); 

     ListeningExecutorService executor = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(3)); 
     List<ListenableFuture<String>> fList = IntStream.range(1, 1000) 
       .mapToObj(i -> executor.submit(() -> String.valueOf(i))).collect(Collectors.toList()); 

     List<String> resps = Lists.newArrayListWithCapacity(fList.size()); 
     Object lock = new Object(); 

     fList.forEach(f -> Futures.addCallback(f, new FutureCallback<String>() { 
      @Override 
      public void onSuccess(String resp) { 
       synchronized (lock) { 
        resps.add(resp); 
        result.setResult(new ResultWrapper(
          resps.size() == fList.size(), 
          ImmutableList.copyOf(resps))); 
       } 

      } 

      @Override 
      public void onFailure(Throwable t) { 
       // NOP 
      } 
     })); 

     return result; 
    } 

    private static class ResultWrapper { 

     private boolean allFinished; 
     private List<String> list; 

     public ResultWrapper(boolean allFinished, List<String> list) { 
      this.allFinished = allFinished; 
      this.list = list; 
     } 

     public boolean isAllFinished() { 
      return allFinished; 
     } 

     public List<String> getList() { 
      return list; 
     } 
    }