¿Hay alguna manera fácil de convertir Future <Future<T>> en Future <T>?

Question

Tengo un código que envía una solicitud a otro hilo que puede o no enviar esa solicitud a otro hilo. Eso produce un tipo de devolución de Future<Future<T>>. ¿Hay alguna forma no atroz de convertir esto inmediatamente en Future<T> que espera la finalización de toda la cadena futura?

Ya estoy usando la biblioteca de Guava para manejar otras cosas divertidas de concurrencia y como reemplazo de Google Collections y su funcionamiento, pero parece que no puedo encontrar algo para este caso.

Answer 1

Guava 13.0 agrega Futures.dereference para hacer esto. Requiere un ListenableFuture<ListenableFuture>, en lugar de un simple Future<Future>. (Operar en un Future llano requeriría una llamada makeListenable, cada uno de los cuales requiere un hilo dedicado para la vida de la tarea (como se aclara con el nuevo nombre del método, JdkFutureAdapters.listenInPoolThread).

Answer 2

Se puede crear una clase como:

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 
     } 
    } 
    ... 
} 

Vas a tener que lidiar con las excepciones que se interponen() puede aumentar, pero otros métodos no pueden.

Answer 3

Esta fue mi primera puñalada, pero estoy seguro de que hay muchas fallas. Estaría más que feliz de simplemente reemplazarlo con algo como 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

Creo que esto es lo mejor que se puede hacer para implementar el contrato de Future. Tomé la táctica de ser lo más imprudente posible para asegurarme de que cumpla con el contrato. No especialmente la implementación de get with 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

Otra posible implementación que utiliza las bibliotecas guava es mucho más simple.

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); 
     } 
    }); 
    } 
}