C'è un modo semplice per trasformare Future <Future<T>> in Future <T>?

Question

Ho un codice che invia una richiesta a un altro thread che può o non può inviare tale richiesta a un altro thread. Ciò produce un tipo di ritorno di Future<Future<T>>. Esiste un modo non odioso per trasformare immediatamente questo in Future<T> che attende il completamento dell'intera catena futura?

Sto già utilizzando la libreria Guava per gestire altre divertenti attività di concorrenza e in sostituzione di Google Collections e funziona bene, ma non riesco a trovare qualcosa per questo caso.

Answer 1

Guava 13,0 aggiunge Futures.dereference per eseguire questa operazione. Richiede un ListenableFuture<ListenableFuture>, anziché un semplice Future<Future>. (Operando su un piano Future richiederebbe una chiamata makeListenable, ognuna delle quali richiede un thread dedicato per la durata dell'attività (come è reso più chiaro dal nuovo nome del metodo, JdkFutureAdapters.listenInPoolThread).

Answer 2

si potrebbe creare una classe come:

public class UnwrapFuture<T> implements Future<T> { 
    Future<Future<T>> wrappedFuture; 

    public UnwrapFuture(Future<Future<T>> wrappedFuture) { 
     this.wrappedFuture = wrappedFuture; 
    } 

    public boolean cancel(boolean mayInterruptIfRunning) { 
     try { 
      return wrappedFuture.get().cancel(mayInterruptIfRunning); 
     } catch (InterruptedException e) { 
      //todo: do something 
     } catch (ExecutionException e) { 
      //todo: do something 
     } 
    } 
    ... 
} 

Dovrete avere a che fare con le eccezioni che ottengono() può aumentare, ma altri metodi non possono.

Answer 3

Questa è stata la mia prima pugnalata, ma sono sicuro che ci sia molto male. Sarei più che felice di sostituirlo con qualcosa come Futures.compress(f).

public class CompressedFuture<T> implements Future<T> { 
    private final Future<Future<T>> delegate; 

    public CompressedFuture(Future<Future<T>> delegate) { 
     this.delegate = delegate; 
    } 

    @Override 
    public boolean cancel(boolean mayInterruptIfRunning) { 
     if (delegate.isDone()) { 
      return delegate.cancel(mayInterruptIfRunning); 
     } 
     try { 
      return delegate.get().cancel(mayInterruptIfRunning); 
     } catch (InterruptedException e) { 
      throw new RuntimeException("Error fetching a finished future", e); 
     } catch (ExecutionException e) { 
      throw new RuntimeException("Error fetching a finished future", e); 
     } 
    } 

    @Override 
    public T get() throws InterruptedException, ExecutionException { 
     return delegate.get().get(); 
    } 

    @Override 
    public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { 
     long endTime = System.currentTimeMillis() + unit.toMillis(timeout); 
     Future<T> next = delegate.get(timeout, unit); 
     return next.get(endTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS); 
    } 

    @Override 
    public boolean isCancelled() { 
     if (!delegate.isDone()) { 
      return delegate.isCancelled(); 
     } 
     try { 
      return delegate.get().isCancelled(); 
     } catch (InterruptedException e) { 
      throw new RuntimeException("Error fetching a finished future", e); 
     } catch (ExecutionException e) { 
      throw new RuntimeException("Error fetching a finished future", e); 
     } 
    } 

    @Override 
    public boolean isDone() { 
     if (!delegate.isDone()) { 
      return false; 
     } 
     try { 
      return delegate.get().isDone(); 
     } catch (InterruptedException e) { 
      throw new RuntimeException("Error fetching a finished future", e); 
     } catch (ExecutionException e) { 
      throw new RuntimeException("Error fetching a finished future", e); 
     } 
    } 
} 

Answer 4

Penso che questo sia il meglio che si possa fare per implementare il contratto del futuro. Ho preso la decisione di essere il più scortese possibile per essere sicuro che rispetti il ​​contratto. Non particolarmente l'implementazione di get con timeout.

import java.util.concurrent.*; 

public class Futures { 
    public <T> Future<T> flatten(Future<Future<T>> future) { 
    return new FlattenedFuture<T>(future); 
    } 

    private static class FlattenedFuture<T> implements Future<T> { 
    private final Future<Future<T>> future; 

    public FlattenedFuture(Future<Future<T>> future) { 
     this.future = future; 
    } 

    public boolean cancel(boolean mayInterruptIfRunning) { 
     if (!future.isDone()) { 
     return future.cancel(mayInterruptIfRunning); 
     } else { 
     while (true) { 
      try { 
      return future.get().cancel(mayInterruptIfRunning); 
      } catch (CancellationException ce) { 
      return true; 
      } catch (ExecutionException ee) { 
      return false; 
      } catch (InterruptedException ie) { 
      // pass 
      } 
     } 
     } 
    } 

    public T get() throws InterruptedException, 
          CancellationException, 
          ExecutionException 
    { 
     return future.get().get(); 
    } 

    public T get(long timeout, TimeUnit unit) throws InterruptedException, 
                CancellationException, 
                ExecutionException, 
                TimeoutException 
    { 
     if (future.isDone()) { 
     return future.get().get(timeout, unit); 
     } else { 
     return future.get(timeout, unit).get(0, TimeUnit.SECONDS); 
     } 
    } 

    public boolean isCancelled() { 
     while (true) { 
     try { 
      return future.isCancelled() || future.get().isCancelled(); 
     } catch (CancellationException ce) { 
      return true; 
     } catch (ExecutionException ee) { 
      return false; 
     } catch (InterruptedException ie) { 
      // pass 
     } 
     } 
    } 

    public boolean isDone() { 
     return future.isDone() && innerIsDone(); 
    } 

    private boolean innerIsDone() { 
     while (true) { 
     try { 
      return future.get().isDone(); 
     } catch (CancellationException ce) { 
      return true; 
     } catch (ExecutionException ee) { 
      return true; 
     } catch (InterruptedException ie) { 
      // pass 
     } 
     } 
    } 
    } 
} 

Answer 5

Un'altra possibile implementazione che utilizza le librerie guava ed è molto più semplice.

import java.util.concurrent.*; 
import com.google.common.util.concurrent.*; 
import com.google.common.base.*; 

public class FFutures { 
    public <T> Future<T> flatten(Future<Future<T>> future) { 
    return Futures.chain(Futures.makeListenable(future), new Function<Future<T>, ListenableFuture<T>>() { 
     public ListenableFuture<T> apply(Future<T> f) { 
     return Futures.makeListenable(f); 
     } 
    }); 
    } 
}