001/*
002 * Copyright (C) 2017 The Guava Authors
003 *
004 * Licensed under the Apache License, Version 2.0 (the "License");
005 * you may not use this file except in compliance with the License.
006 * You may obtain a copy of the License at
007 *
008 * http://www.apache.org/licenses/LICENSE-2.0
009 *
010 * Unless required by applicable law or agreed to in writing, software
011 * distributed under the License is distributed on an "AS IS" BASIS,
012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
013 * See the License for the specific language governing permissions and
014 * limitations under the License.
015 */
016
017package com.google.common.util.concurrent;
018
019import static com.google.common.base.Functions.constant;
020import static com.google.common.base.MoreObjects.toStringHelper;
021import static com.google.common.base.Preconditions.checkArgument;
022import static com.google.common.base.Preconditions.checkNotNull;
023import static com.google.common.base.Preconditions.checkState;
024import static com.google.common.collect.Lists.asList;
025import static com.google.common.util.concurrent.ClosingFuture.State.CLOSED;
026import static com.google.common.util.concurrent.ClosingFuture.State.CLOSING;
027import static com.google.common.util.concurrent.ClosingFuture.State.OPEN;
028import static com.google.common.util.concurrent.ClosingFuture.State.SUBSUMED;
029import static com.google.common.util.concurrent.ClosingFuture.State.WILL_CLOSE;
030import static com.google.common.util.concurrent.ClosingFuture.State.WILL_CREATE_VALUE_AND_CLOSER;
031import static com.google.common.util.concurrent.Futures.getDone;
032import static com.google.common.util.concurrent.Futures.immediateFuture;
033import static com.google.common.util.concurrent.Futures.nonCancellationPropagating;
034import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
035import static java.util.logging.Level.FINER;
036import static java.util.logging.Level.SEVERE;
037import static java.util.logging.Level.WARNING;
038
039import com.google.common.annotations.Beta;
040import com.google.common.annotations.VisibleForTesting;
041import com.google.common.base.Function;
042import com.google.common.collect.FluentIterable;
043import com.google.common.collect.ImmutableList;
044import com.google.common.util.concurrent.ClosingFuture.Combiner.AsyncCombiningCallable;
045import com.google.common.util.concurrent.ClosingFuture.Combiner.CombiningCallable;
046import com.google.common.util.concurrent.Futures.FutureCombiner;
047import com.google.errorprone.annotations.CanIgnoreReturnValue;
048import com.google.errorprone.annotations.DoNotMock;
049import com.google.j2objc.annotations.RetainedWith;
050import java.io.Closeable;
051import java.util.IdentityHashMap;
052import java.util.Map;
053import java.util.concurrent.Callable;
054import java.util.concurrent.CancellationException;
055import java.util.concurrent.CountDownLatch;
056import java.util.concurrent.ExecutionException;
057import java.util.concurrent.Executor;
058import java.util.concurrent.Future;
059import java.util.concurrent.RejectedExecutionException;
060import java.util.concurrent.atomic.AtomicReference;
061import java.util.logging.Logger;
062import org.checkerframework.checker.nullness.qual.Nullable;
063
064/**
065 * A step in a pipeline of an asynchronous computation. When the last step in the computation is
066 * complete, some objects captured during the computation are closed.
067 *
068 * <p>A pipeline of {@code ClosingFuture}s is a tree of steps. Each step represents either an
069 * asynchronously-computed intermediate value, or else an exception that indicates the failure or
070 * cancellation of the operation so far. The only way to extract the value or exception from a step
071 * is by declaring that step to be the last step of the pipeline. Nevertheless, we refer to the
072 * "value" of a successful step or the "result" (value or exception) of any step.
073 *
074 * <ol>
075 *   <li>A pipeline starts at its leaf step (or steps), which is created from either a callable
076 *       block or a {@link ListenableFuture}.
077 *   <li>Each other step is derived from one or more input steps. At each step, zero or more objects
078 *       can be captured for later closing.
079 *   <li>There is one last step (the root of the tree), from which you can extract the final result
080 *       of the computation. After that result is available (or the computation fails), all objects
081 *       captured by any of the steps in the pipeline are closed.
082 * </ol>
083 *
084 * <h3>Starting a pipeline</h3>
085 *
086 * Start a {@code ClosingFuture} pipeline {@linkplain #submit(ClosingCallable, Executor) from a
087 * callable block} that may capture objects for later closing. To start a pipeline from a {@link
088 * ListenableFuture} that doesn't create resources that should be closed later, you can use {@link
089 * #from(ListenableFuture)} instead.
090 *
091 * <h3>Derived steps</h3>
092 *
093 * A {@code ClosingFuture} step can be derived from one or more input {@code ClosingFuture} steps in
094 * ways similar to {@link FluentFuture}s:
095 *
096 * <ul>
097 *   <li>by transforming the value from a successful input step,
098 *   <li>by catching the exception from a failed input step, or
099 *   <li>by combining the results of several input steps.
100 * </ul>
101 *
102 * Each derivation can capture the next value or any intermediate objects for later closing.
103 *
104 * <p>A step can be the input to at most one derived step. Once you transform its value, catch its
105 * exception, or combine it with others, you cannot do anything else with it, including declare it
106 * to be the last step of the pipeline.
107 *
108 * <h4>Transforming</h4>
109 *
110 * To derive the next step by asynchronously applying a function to an input step's value, call
111 * {@link #transform(ClosingFunction, Executor)} or {@link #transformAsync(AsyncClosingFunction,
112 * Executor)} on the input step.
113 *
114 * <h4>Catching</h4>
115 *
116 * To derive the next step from a failed input step, call {@link #catching(Class, ClosingFunction,
117 * Executor)} or {@link #catchingAsync(Class, AsyncClosingFunction, Executor)} on the input step.
118 *
119 * <h4>Combining</h4>
120 *
121 * To derive a {@code ClosingFuture} from two or more input steps, pass the input steps to {@link
122 * #whenAllComplete(Iterable)} or {@link #whenAllSucceed(Iterable)} or its overloads.
123 *
124 * <h3>Cancelling</h3>
125 *
126 * Any step in a pipeline can be {@linkplain #cancel(boolean) cancelled}, even after another step
127 * has been derived, with the same semantics as cancelling a {@link Future}. In addition, a
128 * successfully cancelled step will immediately start closing all objects captured for later closing
129 * by it and by its input steps.
130 *
131 * <h3>Ending a pipeline</h3>
132 *
133 * Each {@code ClosingFuture} pipeline must be ended. To end a pipeline, decide whether you want to
134 * close the captured objects automatically or manually.
135 *
136 * <h4>Automatically closing</h4>
137 *
138 * You can extract a {@link Future} that represents the result of the last step in the pipeline by
139 * calling {@link #finishToFuture()}. All objects the pipeline has captured for closing will begin
140 * to be closed asynchronously <b>after</b> the returned {@code Future} is done: the future
141 * completes before closing starts, rather than once it has finished.
142 *
143 * <pre>{@code
144 * FluentFuture<UserName> userName =
145 *     ClosingFuture.submit(
146 *             closer -> closer.eventuallyClose(database.newTransaction(), closingExecutor),
147 *             executor)
148 *         .transformAsync((closer, transaction) -> transaction.queryClosingFuture("..."), executor)
149 *         .transform((closer, result) -> result.get("userName"), directExecutor())
150 *         .catching(DBException.class, e -> "no user", directExecutor())
151 *         .finishToFuture();
152 * }</pre>
153 *
154 * In this example, when the {@code userName} {@link Future} is done, the transaction and the query
155 * result cursor will both be closed, even if the operation is cancelled or fails.
156 *
157 * <h4>Manually closing</h4>
158 *
159 * If you want to close the captured objects manually, after you've used the final result, call
160 * {@link #finishToValueAndCloser(ValueAndCloserConsumer, Executor)} to get an object that holds the
161 * final result. You then call {@link ValueAndCloser#closeAsync()} to close the captured objects.
162 *
163 * <pre>{@code
164 *     ClosingFuture.submit(
165 *             closer -> closer.eventuallyClose(database.newTransaction(), closingExecutor),
166 *             executor)
167 *     .transformAsync((closer, transaction) -> transaction.queryClosingFuture("..."), executor)
168 *     .transform((closer, result) -> result.get("userName"), directExecutor())
169 *     .catching(DBException.class, e -> "no user", directExecutor())
170 *     .finishToValueAndCloser(
171 *         valueAndCloser -> this.userNameValueAndCloser = valueAndCloser, executor);
172 *
173 * // later
174 * try { // get() will throw if the operation failed or was cancelled.
175 *   UserName userName = userNameValueAndCloser.get();
176 *   // do something with userName
177 * } finally {
178 *   userNameValueAndCloser.closeAsync();
179 * }
180 * }</pre>
181 *
182 * In this example, when {@code userNameValueAndCloser.closeAsync()} is called, the transaction and
183 * the query result cursor will both be closed, even if the operation is cancelled or fails.
184 *
185 * <p>Note that if you don't call {@code closeAsync()}, the captured objects will not be closed. The
186 * automatic-closing approach described above is safer.
187 *
188 * @param <V> the type of the value of this step
189 * @since 30.0
190 */
191// TODO(dpb): Consider reusing one CloseableList for the entire pipeline, modulo combinations.
192@Beta // @Beta for one release.
193@DoNotMock("Use ClosingFuture.from(Futures.immediate*Future)")
194// TODO(dpb): GWT compatibility.
195public final class ClosingFuture<V> {
196
197  private static final Logger logger = Logger.getLogger(ClosingFuture.class.getName());
198
199  /**
200   * An object that can capture objects to be closed later, when a {@link ClosingFuture} pipeline is
201   * done.
202   */
203  public static final class DeferredCloser {
204    @RetainedWith private final CloseableList list;
205
206    DeferredCloser(CloseableList list) {
207      this.list = list;
208    }
209
210    /**
211     * Captures an object to be closed when a {@link ClosingFuture} pipeline is done.
212     *
213     * <p>For users of the {@code -jre} flavor of Guava, the object can be any {@code
214     * AutoCloseable}. For users of the {@code -android} flavor, the object must be a {@code
215     * Closeable}. (For more about the flavors, see <a
216     * href="https://github.com/google/guava#adding-guava-to-your-build">Adding Guava to your
217     * build</a>.)
218     *
219     * <p>Be careful when targeting an older SDK than you are building against (most commonly when
220     * building for Android): Ensure that any object you pass implements the interface not just in
221     * your current SDK version but also at the oldest version you support. For example, <a
222     * href="https://developer.android.com/sdk/api_diff/16/">API Level 16</a> is the first version
223     * in which {@code Cursor} is {@code Closeable}. To support older versions, pass a wrapper
224     * {@code Closeable} with a method reference like {@code cursor::close}.
225     *
226     * <p>Note that this method is still binary-compatible between flavors because the erasure of
227     * its parameter type is {@code Object}, not {@code AutoCloseable} or {@code Closeable}.
228     *
229     * @param closeable the object to be closed (see notes above)
230     * @param closingExecutor the object will be closed on this executor
231     * @return the first argument
232     */
233    @CanIgnoreReturnValue
234    public <C extends @Nullable Object & @Nullable AutoCloseable> C eventuallyClose(
235        C closeable, Executor closingExecutor) {
236      checkNotNull(closingExecutor);
237      if (closeable != null) {
238        list.add(closeable, closingExecutor);
239      }
240      return closeable;
241    }
242  }
243
244  /**
245   * An operation that computes a result.
246   *
247   * @param <V> the type of the result
248   */
249  @FunctionalInterface
250  public interface ClosingCallable<V extends @Nullable Object> {
251    /**
252     * Computes a result, or throws an exception if unable to do so.
253     *
254     * <p>Any objects that are passed to {@link DeferredCloser#eventuallyClose(Closeable, Executor)
255     * closer.eventuallyClose()} will be closed when the {@link ClosingFuture} pipeline is done (but
256     * not before this method completes), even if this method throws or the pipeline is cancelled.
257     */
258    V call(DeferredCloser closer) throws Exception;
259  }
260
261  /**
262   * An operation that computes a {@link ClosingFuture} of a result.
263   *
264   * @param <V> the type of the result
265   * @since 30.1
266   */
267  @FunctionalInterface
268  public interface AsyncClosingCallable<V extends @Nullable Object> {
269    /**
270     * Computes a result, or throws an exception if unable to do so.
271     *
272     * <p>Any objects that are passed to {@link DeferredCloser#eventuallyClose(Closeable, Executor)
273     * closer.eventuallyClose()} will be closed when the {@link ClosingFuture} pipeline is done (but
274     * not before this method completes), even if this method throws or the pipeline is cancelled.
275     */
276    ClosingFuture<V> call(DeferredCloser closer) throws Exception;
277  }
278
279  /**
280   * A function from an input to a result.
281   *
282   * @param <T> the type of the input to the function
283   * @param <U> the type of the result of the function
284   */
285  @FunctionalInterface
286  public interface ClosingFunction<T extends @Nullable Object, U extends @Nullable Object> {
287
288    /**
289     * Applies this function to an input, or throws an exception if unable to do so.
290     *
291     * <p>Any objects that are passed to {@link DeferredCloser#eventuallyClose(Closeable, Executor)
292     * closer.eventuallyClose()} will be closed when the {@link ClosingFuture} pipeline is done (but
293     * not before this method completes), even if this method throws or the pipeline is cancelled.
294     */
295    U apply(DeferredCloser closer, T input) throws Exception;
296  }
297
298  /**
299   * A function from an input to a {@link ClosingFuture} of a result.
300   *
301   * @param <T> the type of the input to the function
302   * @param <U> the type of the result of the function
303   */
304  @FunctionalInterface
305  public interface AsyncClosingFunction<T extends @Nullable Object, U extends @Nullable Object> {
306    /**
307     * Applies this function to an input, or throws an exception if unable to do so.
308     *
309     * <p>Any objects that are passed to {@link DeferredCloser#eventuallyClose(Closeable, Executor)
310     * closer.eventuallyClose()} will be closed when the {@link ClosingFuture} pipeline is done (but
311     * not before this method completes), even if this method throws or the pipeline is cancelled.
312     */
313    ClosingFuture<U> apply(DeferredCloser closer, T input) throws Exception;
314  }
315
316  /**
317   * An object that holds the final result of an asynchronous {@link ClosingFuture} operation and
318   * allows the user to close all the closeable objects that were captured during it for later
319   * closing.
320   *
321   * <p>The asynchronous operation will have completed before this object is created.
322   *
323   * @param <V> the type of the value of a successful operation
324   * @see ClosingFuture#finishToValueAndCloser(ValueAndCloserConsumer, Executor)
325   */
326  public static final class ValueAndCloser<V> {
327
328    private final ClosingFuture<? extends V> closingFuture;
329
330    ValueAndCloser(ClosingFuture<? extends V> closingFuture) {
331      this.closingFuture = checkNotNull(closingFuture);
332    }
333
334    /**
335     * Returns the final value of the associated {@link ClosingFuture}, or throws an exception as
336     * {@link Future#get()} would.
337     *
338     * <p>Because the asynchronous operation has already completed, this method is synchronous and
339     * returns immediately.
340     *
341     * @throws CancellationException if the computation was cancelled
342     * @throws ExecutionException if the computation threw an exception
343     */
344    @Nullable
345    public V get() throws ExecutionException {
346      return getDone(closingFuture.future);
347    }
348
349    /**
350     * Starts closing all closeable objects captured during the {@link ClosingFuture}'s asynchronous
351     * operation on the {@link Executor}s specified by calls to {@link
352     * DeferredCloser#eventuallyClose(Closeable, Executor)}.
353     *
354     * <p>If any such calls specified {@link MoreExecutors#directExecutor()}, those objects will be
355     * closed synchronously.
356     *
357     * <p>Idempotent: objects will be closed at most once.
358     */
359    public void closeAsync() {
360      closingFuture.close();
361    }
362  }
363
364  /**
365   * Represents an operation that accepts a {@link ValueAndCloser} for the last step in a {@link
366   * ClosingFuture} pipeline.
367   *
368   * @param <V> the type of the final value of a successful pipeline
369   * @see ClosingFuture#finishToValueAndCloser(ValueAndCloserConsumer, Executor)
370   */
371  @FunctionalInterface
372  public interface ValueAndCloserConsumer<V> {
373
374    /** Accepts a {@link ValueAndCloser} for the last step in a {@link ClosingFuture} pipeline. */
375    void accept(ValueAndCloser<V> valueAndCloser);
376  }
377
378  /**
379   * Starts a {@link ClosingFuture} pipeline by submitting a callable block to an executor.
380   *
381   * @throws java.util.concurrent.RejectedExecutionException if the task cannot be scheduled for
382   *     execution
383   */
384  public static <V> ClosingFuture<V> submit(ClosingCallable<V> callable, Executor executor) {
385    return new ClosingFuture<>(callable, executor);
386  }
387
388  /**
389   * Starts a {@link ClosingFuture} pipeline by submitting a callable block to an executor.
390   *
391   * @throws java.util.concurrent.RejectedExecutionException if the task cannot be scheduled for
392   *     execution
393   * @since 30.1
394   */
395  public static <V> ClosingFuture<V> submitAsync(
396      AsyncClosingCallable<V> callable, Executor executor) {
397    return new ClosingFuture<>(callable, executor);
398  }
399
400  /**
401   * Starts a {@link ClosingFuture} pipeline with a {@link ListenableFuture}.
402   *
403   * <p>{@code future}'s value will not be closed when the pipeline is done even if {@code V}
404   * implements {@link Closeable}. In order to start a pipeline with a value that will be closed
405   * when the pipeline is done, use {@link #submit(ClosingCallable, Executor)} instead.
406   */
407  public static <V> ClosingFuture<V> from(ListenableFuture<V> future) {
408    return new ClosingFuture<V>(future);
409  }
410
411  /**
412   * Starts a {@link ClosingFuture} pipeline with a {@link ListenableFuture}.
413   *
414   * <p>If {@code future} succeeds, its value will be closed (using {@code closingExecutor)} when
415   * the pipeline is done, even if the pipeline is canceled or fails.
416   *
417   * <p>Cancelling the pipeline will not cancel {@code future}, so that the pipeline can access its
418   * value in order to close it.
419   *
420   * @param future the future to create the {@code ClosingFuture} from. For discussion of the
421   *     future's result type {@code C}, see {@link DeferredCloser#eventuallyClose(Closeable,
422   *     Executor)}.
423   * @param closingExecutor the future's result will be closed on this executor
424   * @deprecated Creating {@link Future}s of closeable types is dangerous in general because the
425   *     underlying value may never be closed if the {@link Future} is canceled after its operation
426   *     begins. Consider replacing code that creates {@link ListenableFuture}s of closeable types,
427   *     including those that pass them to this method, with {@link #submit(ClosingCallable,
428   *     Executor)} in order to ensure that resources do not leak. Or, to start a pipeline with a
429   *     {@link ListenableFuture} that doesn't create values that should be closed, use {@link
430   *     ClosingFuture#from}.
431   */
432  @Deprecated
433  public static <C extends @Nullable Object & @Nullable AutoCloseable>
434      ClosingFuture<C> eventuallyClosing(
435          ListenableFuture<C> future, final Executor closingExecutor) {
436    checkNotNull(closingExecutor);
437    final ClosingFuture<C> closingFuture = new ClosingFuture<>(nonCancellationPropagating(future));
438    Futures.addCallback(
439        future,
440        new FutureCallback<@Nullable AutoCloseable>() {
441          @Override
442          public void onSuccess(@Nullable AutoCloseable result) {
443            closingFuture.closeables.closer.eventuallyClose(result, closingExecutor);
444          }
445
446          @Override
447          public void onFailure(Throwable t) {}
448        },
449        directExecutor());
450    return closingFuture;
451  }
452
453  /**
454   * Starts specifying how to combine {@link ClosingFuture}s into a single pipeline.
455   *
456   * @throws IllegalStateException if a {@code ClosingFuture} has already been derived from any of
457   *     the {@code futures}, or if any has already been {@linkplain #finishToFuture() finished}
458   */
459  public static Combiner whenAllComplete(Iterable<? extends ClosingFuture<?>> futures) {
460    return new Combiner(false, futures);
461  }
462
463  /**
464   * Starts specifying how to combine {@link ClosingFuture}s into a single pipeline.
465   *
466   * @throws IllegalStateException if a {@code ClosingFuture} has already been derived from any of
467   *     the arguments, or if any has already been {@linkplain #finishToFuture() finished}
468   */
469  public static Combiner whenAllComplete(
470      ClosingFuture<?> future1, ClosingFuture<?>... moreFutures) {
471    return whenAllComplete(asList(future1, moreFutures));
472  }
473
474  /**
475   * Starts specifying how to combine {@link ClosingFuture}s into a single pipeline, assuming they
476   * all succeed. If any fail, the resulting pipeline will fail.
477   *
478   * @throws IllegalStateException if a {@code ClosingFuture} has already been derived from any of
479   *     the {@code futures}, or if any has already been {@linkplain #finishToFuture() finished}
480   */
481  public static Combiner whenAllSucceed(Iterable<? extends ClosingFuture<?>> futures) {
482    return new Combiner(true, futures);
483  }
484
485  /**
486   * Starts specifying how to combine two {@link ClosingFuture}s into a single pipeline, assuming
487   * they all succeed. If any fail, the resulting pipeline will fail.
488   *
489   * <p>Calling this method allows you to use lambdas or method references typed with the types of
490   * the input {@link ClosingFuture}s.
491   *
492   * @throws IllegalStateException if a {@code ClosingFuture} has already been derived from any of
493   *     the arguments, or if any has already been {@linkplain #finishToFuture() finished}
494   */
495  public static <V1, V2> Combiner2<V1, V2> whenAllSucceed(
496      ClosingFuture<V1> future1, ClosingFuture<V2> future2) {
497    return new Combiner2<>(future1, future2);
498  }
499
500  /**
501   * Starts specifying how to combine three {@link ClosingFuture}s into a single pipeline, assuming
502   * they all succeed. If any fail, the resulting pipeline will fail.
503   *
504   * <p>Calling this method allows you to use lambdas or method references typed with the types of
505   * the input {@link ClosingFuture}s.
506   *
507   * @throws IllegalStateException if a {@code ClosingFuture} has already been derived from any of
508   *     the arguments, or if any has already been {@linkplain #finishToFuture() finished}
509   */
510  public static <V1, V2, V3> Combiner3<V1, V2, V3> whenAllSucceed(
511      ClosingFuture<V1> future1, ClosingFuture<V2> future2, ClosingFuture<V3> future3) {
512    return new Combiner3<>(future1, future2, future3);
513  }
514
515  /**
516   * Starts specifying how to combine four {@link ClosingFuture}s into a single pipeline, assuming
517   * they all succeed. If any fail, the resulting pipeline will fail.
518   *
519   * <p>Calling this method allows you to use lambdas or method references typed with the types of
520   * the input {@link ClosingFuture}s.
521   *
522   * @throws IllegalStateException if a {@code ClosingFuture} has already been derived from any of
523   *     the arguments, or if any has already been {@linkplain #finishToFuture() finished}
524   */
525  public static <V1, V2, V3, V4> Combiner4<V1, V2, V3, V4> whenAllSucceed(
526      ClosingFuture<V1> future1,
527      ClosingFuture<V2> future2,
528      ClosingFuture<V3> future3,
529      ClosingFuture<V4> future4) {
530    return new Combiner4<>(future1, future2, future3, future4);
531  }
532
533  /**
534   * Starts specifying how to combine five {@link ClosingFuture}s into a single pipeline, assuming
535   * they all succeed. If any fail, the resulting pipeline will fail.
536   *
537   * <p>Calling this method allows you to use lambdas or method references typed with the types of
538   * the input {@link ClosingFuture}s.
539   *
540   * @throws IllegalStateException if a {@code ClosingFuture} has already been derived from any of
541   *     the arguments, or if any has already been {@linkplain #finishToFuture() finished}
542   */
543  public static <V1, V2, V3, V4, V5> Combiner5<V1, V2, V3, V4, V5> whenAllSucceed(
544      ClosingFuture<V1> future1,
545      ClosingFuture<V2> future2,
546      ClosingFuture<V3> future3,
547      ClosingFuture<V4> future4,
548      ClosingFuture<V5> future5) {
549    return new Combiner5<>(future1, future2, future3, future4, future5);
550  }
551
552  /**
553   * Starts specifying how to combine {@link ClosingFuture}s into a single pipeline, assuming they
554   * all succeed. If any fail, the resulting pipeline will fail.
555   *
556   * @throws IllegalStateException if a {@code ClosingFuture} has already been derived from any of
557   *     the arguments, or if any has already been {@linkplain #finishToFuture() finished}
558   */
559  public static Combiner whenAllSucceed(
560      ClosingFuture<?> future1,
561      ClosingFuture<?> future2,
562      ClosingFuture<?> future3,
563      ClosingFuture<?> future4,
564      ClosingFuture<?> future5,
565      ClosingFuture<?> future6,
566      ClosingFuture<?>... moreFutures) {
567    return whenAllSucceed(
568        FluentIterable.of(future1, future2, future3, future4, future5, future6)
569            .append(moreFutures));
570  }
571
572  private final AtomicReference<State> state = new AtomicReference<>(OPEN);
573  private final CloseableList closeables = new CloseableList();
574  private final FluentFuture<V> future;
575
576  private ClosingFuture(ListenableFuture<V> future) {
577    this.future = FluentFuture.from(future);
578  }
579
580  private ClosingFuture(final ClosingCallable<V> callable, Executor executor) {
581    checkNotNull(callable);
582    TrustedListenableFutureTask<V> task =
583        TrustedListenableFutureTask.create(
584            new Callable<V>() {
585              @Override
586              public V call() throws Exception {
587                return callable.call(closeables.closer);
588              }
589
590              @Override
591              public String toString() {
592                return callable.toString();
593              }
594            });
595    executor.execute(task);
596    this.future = task;
597  }
598
599  private ClosingFuture(final AsyncClosingCallable<V> callable, Executor executor) {
600    checkNotNull(callable);
601    TrustedListenableFutureTask<V> task =
602        TrustedListenableFutureTask.create(
603            new AsyncCallable<V>() {
604              @Override
605              public ListenableFuture<V> call() throws Exception {
606                CloseableList newCloseables = new CloseableList();
607                try {
608                  ClosingFuture<V> closingFuture = callable.call(newCloseables.closer);
609                  closingFuture.becomeSubsumedInto(closeables);
610                  return closingFuture.future;
611                } finally {
612                  closeables.add(newCloseables, directExecutor());
613                }
614              }
615
616              @Override
617              public String toString() {
618                return callable.toString();
619              }
620            });
621    executor.execute(task);
622    this.future = task;
623  }
624
625  /**
626   * Returns a future that finishes when this step does. Calling {@code get()} on the returned
627   * future returns {@code null} if the step is successful or throws the same exception that would
628   * be thrown by calling {@code finishToFuture().get()} if this were the last step. Calling {@code
629   * cancel()} on the returned future has no effect on the {@code ClosingFuture} pipeline.
630   *
631   * <p>{@code statusFuture} differs from most methods on {@code ClosingFuture}: You can make calls
632   * to {@code statusFuture} <i>in addition to</i> the call you make to {@link #finishToFuture()} or
633   * a derivation method <i>on the same instance</i>. This is important because calling {@code
634   * statusFuture} alone does not provide a way to close the pipeline.
635   */
636  public ListenableFuture<?> statusFuture() {
637    return nonCancellationPropagating(future.transform(constant(null), directExecutor()));
638  }
639
640  /**
641   * Returns a new {@code ClosingFuture} pipeline step derived from this one by applying a function
642   * to its value. The function can use a {@link DeferredCloser} to capture objects to be closed
643   * when the pipeline is done.
644   *
645   * <p>If this {@code ClosingFuture} fails, the function will not be called, and the derived {@code
646   * ClosingFuture} will be equivalent to this one.
647   *
648   * <p>If the function throws an exception, that exception is used as the result of the derived
649   * {@code ClosingFuture}.
650   *
651   * <p>Example usage:
652   *
653   * <pre>{@code
654   * ClosingFuture<List<Row>> rowsFuture =
655   *     queryFuture.transform((closer, result) -> result.getRows(), executor);
656   * }</pre>
657   *
658   * <p>When selecting an executor, note that {@code directExecutor} is dangerous in some cases. See
659   * the discussion in the {@link ListenableFuture#addListener} documentation. All its warnings
660   * about heavyweight listeners are also applicable to heavyweight functions passed to this method.
661   *
662   * <p>After calling this method, you may not call {@link #finishToFuture()}, {@link
663   * #finishToValueAndCloser(ValueAndCloserConsumer, Executor)}, or any other derivation method on
664   * this {@code ClosingFuture}.
665   *
666   * @param function transforms the value of this step to the value of the derived step
667   * @param executor executor to run the function in
668   * @return the derived step
669   * @throws IllegalStateException if a {@code ClosingFuture} has already been derived from this
670   *     one, or if this {@code ClosingFuture} has already been {@linkplain #finishToFuture()
671   *     finished}
672   */
673  public <U> ClosingFuture<U> transform(
674      final ClosingFunction<? super V, U> function, Executor executor) {
675    checkNotNull(function);
676    AsyncFunction<V, U> applyFunction =
677        new AsyncFunction<V, U>() {
678          @Override
679          public ListenableFuture<U> apply(V input) throws Exception {
680            return closeables.applyClosingFunction(function, input);
681          }
682
683          @Override
684          public String toString() {
685            return function.toString();
686          }
687        };
688    // TODO(dpb): Switch to future.transformSync when that exists (passing a throwing function).
689    return derive(future.transformAsync(applyFunction, executor));
690  }
691
692  /**
693   * Returns a new {@code ClosingFuture} pipeline step derived from this one by applying a function
694   * that returns a {@code ClosingFuture} to its value. The function can use a {@link
695   * DeferredCloser} to capture objects to be closed when the pipeline is done (other than those
696   * captured by the returned {@link ClosingFuture}).
697   *
698   * <p>If this {@code ClosingFuture} succeeds, the derived one will be equivalent to the one
699   * returned by the function.
700   *
701   * <p>If this {@code ClosingFuture} fails, the function will not be called, and the derived {@code
702   * ClosingFuture} will be equivalent to this one.
703   *
704   * <p>If the function throws an exception, that exception is used as the result of the derived
705   * {@code ClosingFuture}. But if the exception is thrown after the function creates a {@code
706   * ClosingFuture}, then none of the closeable objects in that {@code ClosingFuture} will be
707   * closed.
708   *
709   * <p>Usage guidelines for this method:
710   *
711   * <ul>
712   *   <li>Use this method only when calling an API that returns a {@link ListenableFuture} or a
713   *       {@code ClosingFuture}. If possible, prefer calling {@link #transform(ClosingFunction,
714   *       Executor)} instead, with a function that returns the next value directly.
715   *   <li>Call {@link DeferredCloser#eventuallyClose(Closeable, Executor) closer.eventuallyClose()}
716   *       for every closeable object this step creates in order to capture it for later closing.
717   *   <li>Return a {@code ClosingFuture}. To turn a {@link ListenableFuture} into a {@code
718   *       ClosingFuture} call {@link #from(ListenableFuture)}.
719   *   <li>In case this step doesn't create new closeables, you can adapt an API that returns a
720   *       {@link ListenableFuture} to return a {@code ClosingFuture} by wrapping it with a call to
721   *       {@link #withoutCloser(AsyncFunction)}
722   * </ul>
723   *
724   * <p>Example usage:
725   *
726   * <pre>{@code
727   * // Result.getRowsClosingFuture() returns a ClosingFuture.
728   * ClosingFuture<List<Row>> rowsFuture =
729   *     queryFuture.transformAsync((closer, result) -> result.getRowsClosingFuture(), executor);
730   *
731   * // Result.writeRowsToOutputStreamFuture() returns a ListenableFuture that resolves to the
732   * // number of written rows. openOutputFile() returns a FileOutputStream (which implements
733   * // Closeable).
734   * ClosingFuture<Integer> rowsFuture2 =
735   *     queryFuture.transformAsync(
736   *         (closer, result) -> {
737   *           FileOutputStream fos = closer.eventuallyClose(openOutputFile(), closingExecutor);
738   *           return ClosingFuture.from(result.writeRowsToOutputStreamFuture(fos));
739   *      },
740   *      executor);
741   *
742   * // Result.getRowsFuture() returns a ListenableFuture (no new closeables are created).
743   * ClosingFuture<List<Row>> rowsFuture3 =
744   *     queryFuture.transformAsync(withoutCloser(Result::getRowsFuture), executor);
745   *
746   * }</pre>
747   *
748   * <p>When selecting an executor, note that {@code directExecutor} is dangerous in some cases. See
749   * the discussion in the {@link ListenableFuture#addListener} documentation. All its warnings
750   * about heavyweight listeners are also applicable to heavyweight functions passed to this method.
751   * (Specifically, {@code directExecutor} functions should avoid heavyweight operations inside
752   * {@code AsyncClosingFunction.apply}. Any heavyweight operations should occur in other threads
753   * responsible for completing the returned {@code ClosingFuture}.)
754   *
755   * <p>After calling this method, you may not call {@link #finishToFuture()}, {@link
756   * #finishToValueAndCloser(ValueAndCloserConsumer, Executor)}, or any other derivation method on
757   * this {@code ClosingFuture}.
758   *
759   * @param function transforms the value of this step to a {@code ClosingFuture} with the value of
760   *     the derived step
761   * @param executor executor to run the function in
762   * @return the derived step
763   * @throws IllegalStateException if a {@code ClosingFuture} has already been derived from this
764   *     one, or if this {@code ClosingFuture} has already been {@linkplain #finishToFuture()
765   *     finished}
766   */
767  public <U> ClosingFuture<U> transformAsync(
768      final AsyncClosingFunction<? super V, U> function, Executor executor) {
769    checkNotNull(function);
770    AsyncFunction<V, U> applyFunction =
771        new AsyncFunction<V, U>() {
772          @Override
773          public ListenableFuture<U> apply(V input) throws Exception {
774            return closeables.applyAsyncClosingFunction(function, input);
775          }
776
777          @Override
778          public String toString() {
779            return function.toString();
780          }
781        };
782    return derive(future.transformAsync(applyFunction, executor));
783  }
784
785  /**
786   * Returns an {@link AsyncClosingFunction} that applies an {@link AsyncFunction} to an input,
787   * ignoring the DeferredCloser and returning a {@code ClosingFuture} derived from the returned
788   * {@link ListenableFuture}.
789   *
790   * <p>Use this method to pass a transformation to {@link #transformAsync(AsyncClosingFunction,
791   * Executor)} or to {@link #catchingAsync(Class, AsyncClosingFunction, Executor)} as long as it
792   * meets these conditions:
793   *
794   * <ul>
795   *   <li>It does not need to capture any {@link Closeable} objects by calling {@link
796   *       DeferredCloser#eventuallyClose(Closeable, Executor)}.
797   *   <li>It returns a {@link ListenableFuture}.
798   * </ul>
799   *
800   * <p>Example usage:
801   *
802   * <pre>{@code
803   * // Result.getRowsFuture() returns a ListenableFuture.
804   * ClosingFuture<List<Row>> rowsFuture =
805   *     queryFuture.transformAsync(withoutCloser(Result::getRowsFuture), executor);
806   * }</pre>
807   *
808   * @param function transforms the value of a {@code ClosingFuture} step to a {@link
809   *     ListenableFuture} with the value of a derived step
810   */
811  public static <V, U> AsyncClosingFunction<V, U> withoutCloser(
812      final AsyncFunction<V, U> function) {
813    checkNotNull(function);
814    return new AsyncClosingFunction<V, U>() {
815      @Override
816      public ClosingFuture<U> apply(DeferredCloser closer, V input) throws Exception {
817        return ClosingFuture.from(function.apply(input));
818      }
819    };
820  }
821
822  /**
823   * Returns a new {@code ClosingFuture} pipeline step derived from this one by applying a function
824   * to its exception if it is an instance of a given exception type. The function can use a {@link
825   * DeferredCloser} to capture objects to be closed when the pipeline is done.
826   *
827   * <p>If this {@code ClosingFuture} succeeds or fails with a different exception type, the
828   * function will not be called, and the derived {@code ClosingFuture} will be equivalent to this
829   * one.
830   *
831   * <p>If the function throws an exception, that exception is used as the result of the derived
832   * {@code ClosingFuture}.
833   *
834   * <p>Example usage:
835   *
836   * <pre>{@code
837   * ClosingFuture<QueryResult> queryFuture =
838   *     queryFuture.catching(
839   *         QueryException.class, (closer, x) -> Query.emptyQueryResult(), executor);
840   * }</pre>
841   *
842   * <p>When selecting an executor, note that {@code directExecutor} is dangerous in some cases. See
843   * the discussion in the {@link ListenableFuture#addListener} documentation. All its warnings
844   * about heavyweight listeners are also applicable to heavyweight functions passed to this method.
845   *
846   * <p>After calling this method, you may not call {@link #finishToFuture()}, {@link
847   * #finishToValueAndCloser(ValueAndCloserConsumer, Executor)}, or any other derivation method on
848   * this {@code ClosingFuture}.
849   *
850   * @param exceptionType the exception type that triggers use of {@code fallback}. The exception
851   *     type is matched against this step's exception. "This step's exception" means the cause of
852   *     the {@link ExecutionException} thrown by {@link Future#get()} on the {@link Future}
853   *     underlying this step or, if {@code get()} throws a different kind of exception, that
854   *     exception itself. To avoid hiding bugs and other unrecoverable errors, callers should
855   *     prefer more specific types, avoiding {@code Throwable.class} in particular.
856   * @param fallback the function to be called if this step fails with the expected exception type.
857   *     The function's argument is this step's exception. "This step's exception" means the cause
858   *     of the {@link ExecutionException} thrown by {@link Future#get()} on the {@link Future}
859   *     underlying this step or, if {@code get()} throws a different kind of exception, that
860   *     exception itself.
861   * @param executor the executor that runs {@code fallback} if the input fails
862   */
863  public <X extends Throwable> ClosingFuture<V> catching(
864      Class<X> exceptionType, ClosingFunction<? super X, ? extends V> fallback, Executor executor) {
865    return catchingMoreGeneric(exceptionType, fallback, executor);
866  }
867
868  // Avoids generic type capture inconsistency problems where |? extends V| is incompatible with V.
869  private <X extends Throwable, W extends V> ClosingFuture<V> catchingMoreGeneric(
870      Class<X> exceptionType, final ClosingFunction<? super X, W> fallback, Executor executor) {
871    checkNotNull(fallback);
872    AsyncFunction<X, W> applyFallback =
873        new AsyncFunction<X, W>() {
874          @Override
875          public ListenableFuture<W> apply(X exception) throws Exception {
876            return closeables.applyClosingFunction(fallback, exception);
877          }
878
879          @Override
880          public String toString() {
881            return fallback.toString();
882          }
883        };
884    // TODO(dpb): Switch to future.catchingSync when that exists (passing a throwing function).
885    return derive(future.catchingAsync(exceptionType, applyFallback, executor));
886  }
887
888  /**
889   * Returns a new {@code ClosingFuture} pipeline step derived from this one by applying a function
890   * that returns a {@code ClosingFuture} to its exception if it is an instance of a given exception
891   * type. The function can use a {@link DeferredCloser} to capture objects to be closed when the
892   * pipeline is done (other than those captured by the returned {@link ClosingFuture}).
893   *
894   * <p>If this {@code ClosingFuture} fails with an exception of the given type, the derived {@code
895   * ClosingFuture} will be equivalent to the one returned by the function.
896   *
897   * <p>If this {@code ClosingFuture} succeeds or fails with a different exception type, the
898   * function will not be called, and the derived {@code ClosingFuture} will be equivalent to this
899   * one.
900   *
901   * <p>If the function throws an exception, that exception is used as the result of the derived
902   * {@code ClosingFuture}. But if the exception is thrown after the function creates a {@code
903   * ClosingFuture}, then none of the closeable objects in that {@code ClosingFuture} will be
904   * closed.
905   *
906   * <p>Usage guidelines for this method:
907   *
908   * <ul>
909   *   <li>Use this method only when calling an API that returns a {@link ListenableFuture} or a
910   *       {@code ClosingFuture}. If possible, prefer calling {@link #catching(Class,
911   *       ClosingFunction, Executor)} instead, with a function that returns the next value
912   *       directly.
913   *   <li>Call {@link DeferredCloser#eventuallyClose(Closeable, Executor) closer.eventuallyClose()}
914   *       for every closeable object this step creates in order to capture it for later closing.
915   *   <li>Return a {@code ClosingFuture}. To turn a {@link ListenableFuture} into a {@code
916   *       ClosingFuture} call {@link #from(ListenableFuture)}.
917   *   <li>In case this step doesn't create new closeables, you can adapt an API that returns a
918   *       {@link ListenableFuture} to return a {@code ClosingFuture} by wrapping it with a call to
919   *       {@link #withoutCloser(AsyncFunction)}
920   * </ul>
921   *
922   * <p>Example usage:
923   *
924   * <pre>{@code
925   * // Fall back to a secondary input stream in case of IOException.
926   * ClosingFuture<InputStream> inputFuture =
927   *     firstInputFuture.catchingAsync(
928   *         IOException.class, (closer, x) -> secondaryInputStreamClosingFuture(), executor);
929   * }
930   * }</pre>
931   *
932   * <p>When selecting an executor, note that {@code directExecutor} is dangerous in some cases. See
933   * the discussion in the {@link ListenableFuture#addListener} documentation. All its warnings
934   * about heavyweight listeners are also applicable to heavyweight functions passed to this method.
935   * (Specifically, {@code directExecutor} functions should avoid heavyweight operations inside
936   * {@code AsyncClosingFunction.apply}. Any heavyweight operations should occur in other threads
937   * responsible for completing the returned {@code ClosingFuture}.)
938   *
939   * <p>After calling this method, you may not call {@link #finishToFuture()}, {@link
940   * #finishToValueAndCloser(ValueAndCloserConsumer, Executor)}, or any other derivation method on
941   * this {@code ClosingFuture}.
942   *
943   * @param exceptionType the exception type that triggers use of {@code fallback}. The exception
944   *     type is matched against this step's exception. "This step's exception" means the cause of
945   *     the {@link ExecutionException} thrown by {@link Future#get()} on the {@link Future}
946   *     underlying this step or, if {@code get()} throws a different kind of exception, that
947   *     exception itself. To avoid hiding bugs and other unrecoverable errors, callers should
948   *     prefer more specific types, avoiding {@code Throwable.class} in particular.
949   * @param fallback the function to be called if this step fails with the expected exception type.
950   *     The function's argument is this step's exception. "This step's exception" means the cause
951   *     of the {@link ExecutionException} thrown by {@link Future#get()} on the {@link Future}
952   *     underlying this step or, if {@code get()} throws a different kind of exception, that
953   *     exception itself.
954   * @param executor the executor that runs {@code fallback} if the input fails
955   */
956  // TODO(dpb): Should this do something special if the function throws CancellationException or
957  // ExecutionException?
958  public <X extends Throwable> ClosingFuture<V> catchingAsync(
959      Class<X> exceptionType,
960      AsyncClosingFunction<? super X, ? extends V> fallback,
961      Executor executor) {
962    return catchingAsyncMoreGeneric(exceptionType, fallback, executor);
963  }
964
965  // Avoids generic type capture inconsistency problems where |? extends V| is incompatible with V.
966  private <X extends Throwable, W extends V> ClosingFuture<V> catchingAsyncMoreGeneric(
967      Class<X> exceptionType,
968      final AsyncClosingFunction<? super X, W> fallback,
969      Executor executor) {
970    checkNotNull(fallback);
971    AsyncFunction<X, W> asyncFunction =
972        new AsyncFunction<X, W>() {
973          @Override
974          public ListenableFuture<W> apply(X exception) throws Exception {
975            return closeables.applyAsyncClosingFunction(fallback, exception);
976          }
977
978          @Override
979          public String toString() {
980            return fallback.toString();
981          }
982        };
983    return derive(future.catchingAsync(exceptionType, asyncFunction, executor));
984  }
985
986  /**
987   * Marks this step as the last step in the {@code ClosingFuture} pipeline.
988   *
989   * <p>The returned {@link Future} is completed when the pipeline's computation completes, or when
990   * the pipeline is cancelled.
991   *
992   * <p>All objects the pipeline has captured for closing will begin to be closed asynchronously
993   * <b>after</b> the returned {@code Future} is done: the future completes before closing starts,
994   * rather than once it has finished.
995   *
996   * <p>After calling this method, you may not call {@link
997   * #finishToValueAndCloser(ValueAndCloserConsumer, Executor)}, this method, or any other
998   * derivation method on this {@code ClosingFuture}.
999   *
1000   * @return a {@link Future} that represents the final value or exception of the pipeline
1001   */
1002  public FluentFuture<V> finishToFuture() {
1003    if (compareAndUpdateState(OPEN, WILL_CLOSE)) {
1004      logger.log(FINER, "will close {0}", this);
1005      future.addListener(
1006          new Runnable() {
1007            @Override
1008            public void run() {
1009              checkAndUpdateState(WILL_CLOSE, CLOSING);
1010              close();
1011              checkAndUpdateState(CLOSING, CLOSED);
1012            }
1013          },
1014          directExecutor());
1015    } else {
1016      switch (state.get()) {
1017        case SUBSUMED:
1018          throw new IllegalStateException(
1019              "Cannot call finishToFuture() after deriving another step");
1020
1021        case WILL_CREATE_VALUE_AND_CLOSER:
1022          throw new IllegalStateException(
1023              "Cannot call finishToFuture() after calling finishToValueAndCloser()");
1024
1025        case WILL_CLOSE:
1026        case CLOSING:
1027        case CLOSED:
1028          throw new IllegalStateException("Cannot call finishToFuture() twice");
1029
1030        case OPEN:
1031          throw new AssertionError();
1032      }
1033    }
1034    return future;
1035  }
1036
1037  /**
1038   * Marks this step as the last step in the {@code ClosingFuture} pipeline. When this step is done,
1039   * {@code receiver} will be called with an object that contains the result of the operation. The
1040   * receiver can store the {@link ValueAndCloser} outside the receiver for later synchronous use.
1041   *
1042   * <p>After calling this method, you may not call {@link #finishToFuture()}, this method again, or
1043   * any other derivation method on this {@code ClosingFuture}.
1044   *
1045   * @param consumer a callback whose method will be called (using {@code executor}) when this
1046   *     operation is done
1047   */
1048  public void finishToValueAndCloser(
1049      final ValueAndCloserConsumer<? super V> consumer, Executor executor) {
1050    checkNotNull(consumer);
1051    if (!compareAndUpdateState(OPEN, WILL_CREATE_VALUE_AND_CLOSER)) {
1052      switch (state.get()) {
1053        case SUBSUMED:
1054          throw new IllegalStateException(
1055              "Cannot call finishToValueAndCloser() after deriving another step");
1056
1057        case WILL_CLOSE:
1058        case CLOSING:
1059        case CLOSED:
1060          throw new IllegalStateException(
1061              "Cannot call finishToValueAndCloser() after calling finishToFuture()");
1062
1063        case WILL_CREATE_VALUE_AND_CLOSER:
1064          throw new IllegalStateException("Cannot call finishToValueAndCloser() twice");
1065
1066        case OPEN:
1067          break;
1068      }
1069      throw new AssertionError(state);
1070    }
1071    future.addListener(
1072        new Runnable() {
1073          @Override
1074          public void run() {
1075            provideValueAndCloser(consumer, ClosingFuture.this);
1076          }
1077        },
1078        executor);
1079  }
1080
1081  private static <C, V extends C> void provideValueAndCloser(
1082      ValueAndCloserConsumer<C> consumer, ClosingFuture<V> closingFuture) {
1083    consumer.accept(new ValueAndCloser<C>(closingFuture));
1084  }
1085
1086  /**
1087   * Attempts to cancel execution of this step. This attempt will fail if the step has already
1088   * completed, has already been cancelled, or could not be cancelled for some other reason. If
1089   * successful, and this step has not started when {@code cancel} is called, this step should never
1090   * run.
1091   *
1092   * <p>If successful, causes the objects captured by this step (if already started) and its input
1093   * step(s) for later closing to be closed on their respective {@link Executor}s. If any such calls
1094   * specified {@link MoreExecutors#directExecutor()}, those objects will be closed synchronously.
1095   *
1096   * @param mayInterruptIfRunning {@code true} if the thread executing this task should be
1097   *     interrupted; otherwise, in-progress tasks are allowed to complete, but the step will be
1098   *     cancelled regardless
1099   * @return {@code false} if the step could not be cancelled, typically because it has already
1100   *     completed normally; {@code true} otherwise
1101   */
1102  @CanIgnoreReturnValue
1103  public boolean cancel(boolean mayInterruptIfRunning) {
1104    logger.log(FINER, "cancelling {0}", this);
1105    boolean cancelled = future.cancel(mayInterruptIfRunning);
1106    if (cancelled) {
1107      close();
1108    }
1109    return cancelled;
1110  }
1111
1112  private void close() {
1113    logger.log(FINER, "closing {0}", this);
1114    closeables.close();
1115  }
1116
1117  private <U> ClosingFuture<U> derive(FluentFuture<U> future) {
1118    ClosingFuture<U> derived = new ClosingFuture<>(future);
1119    becomeSubsumedInto(derived.closeables);
1120    return derived;
1121  }
1122
1123  private void becomeSubsumedInto(CloseableList otherCloseables) {
1124    checkAndUpdateState(OPEN, SUBSUMED);
1125    otherCloseables.add(closeables, directExecutor());
1126  }
1127
1128  /**
1129   * An object that can return the value of the {@link ClosingFuture}s that are passed to {@link
1130   * #whenAllComplete(Iterable)} or {@link #whenAllSucceed(Iterable)}.
1131   *
1132   * <p>Only for use by a {@link CombiningCallable} or {@link AsyncCombiningCallable} object.
1133   */
1134  public static final class Peeker {
1135    private final ImmutableList<ClosingFuture<?>> futures;
1136    private volatile boolean beingCalled;
1137
1138    private Peeker(ImmutableList<ClosingFuture<?>> futures) {
1139      this.futures = checkNotNull(futures);
1140    }
1141
1142    /**
1143     * Returns the value of {@code closingFuture}.
1144     *
1145     * @throws ExecutionException if {@code closingFuture} is a failed step
1146     * @throws CancellationException if the {@code closingFuture}'s future was cancelled
1147     * @throws IllegalArgumentException if {@code closingFuture} is not one of the futures passed to
1148     *     {@link #whenAllComplete(Iterable)} or {@link #whenAllComplete(Iterable)}
1149     * @throws IllegalStateException if called outside of a call to {@link
1150     *     CombiningCallable#call(DeferredCloser, Peeker)} or {@link
1151     *     AsyncCombiningCallable#call(DeferredCloser, Peeker)}
1152     */
1153    public final <D extends @Nullable Object> D getDone(ClosingFuture<D> closingFuture)
1154        throws ExecutionException {
1155      checkState(beingCalled);
1156      checkArgument(futures.contains(closingFuture));
1157      return Futures.getDone(closingFuture.future);
1158    }
1159
1160    private <V extends @Nullable Object> V call(
1161        CombiningCallable<V> combiner, CloseableList closeables) throws Exception {
1162      beingCalled = true;
1163      CloseableList newCloseables = new CloseableList();
1164      try {
1165        return combiner.call(newCloseables.closer, this);
1166      } finally {
1167        closeables.add(newCloseables, directExecutor());
1168        beingCalled = false;
1169      }
1170    }
1171
1172    private <V extends @Nullable Object> FluentFuture<V> callAsync(
1173        AsyncCombiningCallable<V> combiner, CloseableList closeables) throws Exception {
1174      beingCalled = true;
1175      CloseableList newCloseables = new CloseableList();
1176      try {
1177        ClosingFuture<V> closingFuture = combiner.call(newCloseables.closer, this);
1178        closingFuture.becomeSubsumedInto(closeables);
1179        return closingFuture.future;
1180      } finally {
1181        closeables.add(newCloseables, directExecutor());
1182        beingCalled = false;
1183      }
1184    }
1185  }
1186
1187  /**
1188   * A builder of a {@link ClosingFuture} step that is derived from more than one input step.
1189   *
1190   * <p>See {@link #whenAllComplete(Iterable)} and {@link #whenAllSucceed(Iterable)} for how to
1191   * instantiate this class.
1192   *
1193   * <p>Example:
1194   *
1195   * <pre>{@code
1196   * final ClosingFuture<BufferedReader> file1ReaderFuture = ...;
1197   * final ClosingFuture<BufferedReader> file2ReaderFuture = ...;
1198   * ListenableFuture<Integer> numberOfDifferentLines =
1199   *       ClosingFuture.whenAllSucceed(file1ReaderFuture, file2ReaderFuture)
1200   *           .call(
1201   *               (closer, peeker) -> {
1202   *                 BufferedReader file1Reader = peeker.getDone(file1ReaderFuture);
1203   *                 BufferedReader file2Reader = peeker.getDone(file2ReaderFuture);
1204   *                 return countDifferentLines(file1Reader, file2Reader);
1205   *               },
1206   *               executor)
1207   *           .closing(executor);
1208   * }</pre>
1209   */
1210  // TODO(cpovirk): Use simple name instead of fully qualified after we stop building with JDK 8.
1211  @com.google.errorprone.annotations.DoNotMock(
1212      "Use ClosingFuture.whenAllSucceed() or .whenAllComplete() instead.")
1213  public static class Combiner {
1214
1215    private final CloseableList closeables = new CloseableList();
1216
1217    /**
1218     * An operation that returns a result and may throw an exception.
1219     *
1220     * @param <V> the type of the result
1221     */
1222    @FunctionalInterface
1223    public interface CombiningCallable<V extends @Nullable Object> {
1224      /**
1225       * Computes a result, or throws an exception if unable to do so.
1226       *
1227       * <p>Any objects that are passed to {@link DeferredCloser#eventuallyClose(Closeable,
1228       * Executor) closer.eventuallyClose()} will be closed when the {@link ClosingFuture} pipeline
1229       * is done (but not before this method completes), even if this method throws or the pipeline
1230       * is cancelled.
1231       *
1232       * @param peeker used to get the value of any of the input futures
1233       */
1234      V call(DeferredCloser closer, Peeker peeker) throws Exception;
1235    }
1236
1237    /**
1238     * An operation that returns a {@link ClosingFuture} result and may throw an exception.
1239     *
1240     * @param <V> the type of the result
1241     */
1242    @FunctionalInterface
1243    public interface AsyncCombiningCallable<V extends @Nullable Object> {
1244      /**
1245       * Computes a {@link ClosingFuture} result, or throws an exception if unable to do so.
1246       *
1247       * <p>Any objects that are passed to {@link DeferredCloser#eventuallyClose(Closeable,
1248       * Executor) closer.eventuallyClose()} will be closed when the {@link ClosingFuture} pipeline
1249       * is done (but not before this method completes), even if this method throws or the pipeline
1250       * is cancelled.
1251       *
1252       * @param peeker used to get the value of any of the input futures
1253       */
1254      ClosingFuture<V> call(DeferredCloser closer, Peeker peeker) throws Exception;
1255    }
1256
1257    private final boolean allMustSucceed;
1258    protected final ImmutableList<ClosingFuture<?>> inputs;
1259
1260    private Combiner(boolean allMustSucceed, Iterable<? extends ClosingFuture<?>> inputs) {
1261      this.allMustSucceed = allMustSucceed;
1262      this.inputs = ImmutableList.copyOf(inputs);
1263      for (ClosingFuture<?> input : inputs) {
1264        input.becomeSubsumedInto(closeables);
1265      }
1266    }
1267
1268    /**
1269     * Returns a new {@code ClosingFuture} pipeline step derived from the inputs by applying a
1270     * combining function to their values. The function can use a {@link DeferredCloser} to capture
1271     * objects to be closed when the pipeline is done.
1272     *
1273     * <p>If this combiner was returned by a {@link #whenAllSucceed} method and any of the inputs
1274     * fail, so will the returned step.
1275     *
1276     * <p>If the combiningCallable throws a {@code CancellationException}, the pipeline will be
1277     * cancelled.
1278     *
1279     * <p>If the combiningCallable throws an {@code ExecutionException}, the cause of the thrown
1280     * {@code ExecutionException} will be extracted and used as the failure of the derived step.
1281     */
1282    public <V> ClosingFuture<V> call(
1283        final CombiningCallable<V> combiningCallable, Executor executor) {
1284      Callable<V> callable =
1285          new Callable<V>() {
1286            @Override
1287            public V call() throws Exception {
1288              return new Peeker(inputs).call(combiningCallable, closeables);
1289            }
1290
1291            @Override
1292            public String toString() {
1293              return combiningCallable.toString();
1294            }
1295          };
1296      ClosingFuture<V> derived = new ClosingFuture<>(futureCombiner().call(callable, executor));
1297      derived.closeables.add(closeables, directExecutor());
1298      return derived;
1299    }
1300
1301    /**
1302     * Returns a new {@code ClosingFuture} pipeline step derived from the inputs by applying a
1303     * {@code ClosingFuture}-returning function to their values. The function can use a {@link
1304     * DeferredCloser} to capture objects to be closed when the pipeline is done (other than those
1305     * captured by the returned {@link ClosingFuture}).
1306     *
1307     * <p>If this combiner was returned by a {@link #whenAllSucceed} method and any of the inputs
1308     * fail, so will the returned step.
1309     *
1310     * <p>If the combiningCallable throws a {@code CancellationException}, the pipeline will be
1311     * cancelled.
1312     *
1313     * <p>If the combiningCallable throws an {@code ExecutionException}, the cause of the thrown
1314     * {@code ExecutionException} will be extracted and used as the failure of the derived step.
1315     *
1316     * <p>If the combiningCallable throws any other exception, it will be used as the failure of the
1317     * derived step.
1318     *
1319     * <p>If an exception is thrown after the combiningCallable creates a {@code ClosingFuture},
1320     * then none of the closeable objects in that {@code ClosingFuture} will be closed.
1321     *
1322     * <p>Usage guidelines for this method:
1323     *
1324     * <ul>
1325     *   <li>Use this method only when calling an API that returns a {@link ListenableFuture} or a
1326     *       {@code ClosingFuture}. If possible, prefer calling {@link #call(CombiningCallable,
1327     *       Executor)} instead, with a function that returns the next value directly.
1328     *   <li>Call {@link DeferredCloser#eventuallyClose(Closeable, Executor)
1329     *       closer.eventuallyClose()} for every closeable object this step creates in order to
1330     *       capture it for later closing.
1331     *   <li>Return a {@code ClosingFuture}. To turn a {@link ListenableFuture} into a {@code
1332     *       ClosingFuture} call {@link #from(ListenableFuture)}.
1333     * </ul>
1334     *
1335     * <p>The same warnings about doing heavyweight operations within {@link
1336     * ClosingFuture#transformAsync(AsyncClosingFunction, Executor)} apply here.
1337     */
1338    public <V> ClosingFuture<V> callAsync(
1339        final AsyncCombiningCallable<V> combiningCallable, Executor executor) {
1340      AsyncCallable<V> asyncCallable =
1341          new AsyncCallable<V>() {
1342            @Override
1343            public ListenableFuture<V> call() throws Exception {
1344              return new Peeker(inputs).callAsync(combiningCallable, closeables);
1345            }
1346
1347            @Override
1348            public String toString() {
1349              return combiningCallable.toString();
1350            }
1351          };
1352      ClosingFuture<V> derived =
1353          new ClosingFuture<>(futureCombiner().callAsync(asyncCallable, executor));
1354      derived.closeables.add(closeables, directExecutor());
1355      return derived;
1356    }
1357
1358    private FutureCombiner<Object> futureCombiner() {
1359      return allMustSucceed
1360          ? Futures.whenAllSucceed(inputFutures())
1361          : Futures.whenAllComplete(inputFutures());
1362    }
1363
1364    private static final Function<ClosingFuture<?>, FluentFuture<?>> INNER_FUTURE =
1365        new Function<ClosingFuture<?>, FluentFuture<?>>() {
1366          @Override
1367          public FluentFuture<?> apply(ClosingFuture<?> future) {
1368            return future.future;
1369          }
1370        };
1371
1372    private ImmutableList<FluentFuture<?>> inputFutures() {
1373      return FluentIterable.from(inputs).transform(INNER_FUTURE).toList();
1374    }
1375  }
1376
1377  /**
1378   * A generic {@link Combiner} that lets you use a lambda or method reference to combine two {@link
1379   * ClosingFuture}s. Use {@link #whenAllSucceed(ClosingFuture, ClosingFuture)} to start this
1380   * combination.
1381   *
1382   * @param <V1> the type returned by the first future
1383   * @param <V2> the type returned by the second future
1384   */
1385  public static final class Combiner2<V1 extends @Nullable Object, V2 extends @Nullable Object>
1386      extends Combiner {
1387
1388    /**
1389     * A function that returns a value when applied to the values of the two futures passed to
1390     * {@link #whenAllSucceed(ClosingFuture, ClosingFuture)}.
1391     *
1392     * @param <V1> the type returned by the first future
1393     * @param <V2> the type returned by the second future
1394     * @param <U> the type returned by the function
1395     */
1396    @FunctionalInterface
1397    public interface ClosingFunction2<
1398        V1 extends @Nullable Object, V2 extends @Nullable Object, U extends @Nullable Object> {
1399
1400      /**
1401       * Applies this function to two inputs, or throws an exception if unable to do so.
1402       *
1403       * <p>Any objects that are passed to {@link DeferredCloser#eventuallyClose(Closeable,
1404       * Executor) closer.eventuallyClose()} will be closed when the {@link ClosingFuture} pipeline
1405       * is done (but not before this method completes), even if this method throws or the pipeline
1406       * is cancelled.
1407       */
1408      U apply(DeferredCloser closer, V1 value1, V2 value2) throws Exception;
1409    }
1410
1411    /**
1412     * A function that returns a {@link ClosingFuture} when applied to the values of the two futures
1413     * passed to {@link #whenAllSucceed(ClosingFuture, ClosingFuture)}.
1414     *
1415     * @param <V1> the type returned by the first future
1416     * @param <V2> the type returned by the second future
1417     * @param <U> the type returned by the function
1418     */
1419    @FunctionalInterface
1420    public interface AsyncClosingFunction2<
1421        V1 extends @Nullable Object, V2 extends @Nullable Object, U extends @Nullable Object> {
1422
1423      /**
1424       * Applies this function to two inputs, or throws an exception if unable to do so.
1425       *
1426       * <p>Any objects that are passed to {@link DeferredCloser#eventuallyClose(Closeable,
1427       * Executor) closer.eventuallyClose()} will be closed when the {@link ClosingFuture} pipeline
1428       * is done (but not before this method completes), even if this method throws or the pipeline
1429       * is cancelled.
1430       */
1431      ClosingFuture<U> apply(DeferredCloser closer, V1 value1, V2 value2) throws Exception;
1432    }
1433
1434    private final ClosingFuture<V1> future1;
1435    private final ClosingFuture<V2> future2;
1436
1437    private Combiner2(ClosingFuture<V1> future1, ClosingFuture<V2> future2) {
1438      super(true, ImmutableList.of(future1, future2));
1439      this.future1 = future1;
1440      this.future2 = future2;
1441    }
1442
1443    /**
1444     * Returns a new {@code ClosingFuture} pipeline step derived from the inputs by applying a
1445     * combining function to their values. The function can use a {@link DeferredCloser} to capture
1446     * objects to be closed when the pipeline is done.
1447     *
1448     * <p>If this combiner was returned by {@link #whenAllSucceed(ClosingFuture, ClosingFuture)} and
1449     * any of the inputs fail, so will the returned step.
1450     *
1451     * <p>If the function throws a {@code CancellationException}, the pipeline will be cancelled.
1452     *
1453     * <p>If the function throws an {@code ExecutionException}, the cause of the thrown {@code
1454     * ExecutionException} will be extracted and used as the failure of the derived step.
1455     */
1456    public <U extends @Nullable Object> ClosingFuture<U> call(
1457        final ClosingFunction2<V1, V2, U> function, Executor executor) {
1458      return call(
1459          new CombiningCallable<U>() {
1460            @Override
1461            public U call(DeferredCloser closer, Peeker peeker) throws Exception {
1462              return function.apply(closer, peeker.getDone(future1), peeker.getDone(future2));
1463            }
1464
1465            @Override
1466            public String toString() {
1467              return function.toString();
1468            }
1469          },
1470          executor);
1471    }
1472
1473    /**
1474     * Returns a new {@code ClosingFuture} pipeline step derived from the inputs by applying a
1475     * {@code ClosingFuture}-returning function to their values. The function can use a {@link
1476     * DeferredCloser} to capture objects to be closed when the pipeline is done (other than those
1477     * captured by the returned {@link ClosingFuture}).
1478     *
1479     * <p>If this combiner was returned by {@link #whenAllSucceed(ClosingFuture, ClosingFuture)} and
1480     * any of the inputs fail, so will the returned step.
1481     *
1482     * <p>If the function throws a {@code CancellationException}, the pipeline will be cancelled.
1483     *
1484     * <p>If the function throws an {@code ExecutionException}, the cause of the thrown {@code
1485     * ExecutionException} will be extracted and used as the failure of the derived step.
1486     *
1487     * <p>If the function throws any other exception, it will be used as the failure of the derived
1488     * step.
1489     *
1490     * <p>If an exception is thrown after the function creates a {@code ClosingFuture}, then none of
1491     * the closeable objects in that {@code ClosingFuture} will be closed.
1492     *
1493     * <p>Usage guidelines for this method:
1494     *
1495     * <ul>
1496     *   <li>Use this method only when calling an API that returns a {@link ListenableFuture} or a
1497     *       {@code ClosingFuture}. If possible, prefer calling {@link #call(CombiningCallable,
1498     *       Executor)} instead, with a function that returns the next value directly.
1499     *   <li>Call {@link DeferredCloser#eventuallyClose(Closeable, Executor)
1500     *       closer.eventuallyClose()} for every closeable object this step creates in order to
1501     *       capture it for later closing.
1502     *   <li>Return a {@code ClosingFuture}. To turn a {@link ListenableFuture} into a {@code
1503     *       ClosingFuture} call {@link #from(ListenableFuture)}.
1504     * </ul>
1505     *
1506     * <p>The same warnings about doing heavyweight operations within {@link
1507     * ClosingFuture#transformAsync(AsyncClosingFunction, Executor)} apply here.
1508     */
1509    public <U extends @Nullable Object> ClosingFuture<U> callAsync(
1510        final AsyncClosingFunction2<V1, V2, U> function, Executor executor) {
1511      return callAsync(
1512          new AsyncCombiningCallable<U>() {
1513            @Override
1514            public ClosingFuture<U> call(DeferredCloser closer, Peeker peeker) throws Exception {
1515              return function.apply(closer, peeker.getDone(future1), peeker.getDone(future2));
1516            }
1517
1518            @Override
1519            public String toString() {
1520              return function.toString();
1521            }
1522          },
1523          executor);
1524    }
1525  }
1526
1527  /**
1528   * A generic {@link Combiner} that lets you use a lambda or method reference to combine three
1529   * {@link ClosingFuture}s. Use {@link #whenAllSucceed(ClosingFuture, ClosingFuture,
1530   * ClosingFuture)} to start this combination.
1531   *
1532   * @param <V1> the type returned by the first future
1533   * @param <V2> the type returned by the second future
1534   * @param <V3> the type returned by the third future
1535   */
1536  public static final class Combiner3<
1537          V1 extends @Nullable Object, V2 extends @Nullable Object, V3 extends @Nullable Object>
1538      extends Combiner {
1539    /**
1540     * A function that returns a value when applied to the values of the three futures passed to
1541     * {@link #whenAllSucceed(ClosingFuture, ClosingFuture, ClosingFuture)}.
1542     *
1543     * @param <V1> the type returned by the first future
1544     * @param <V2> the type returned by the second future
1545     * @param <V3> the type returned by the third future
1546     * @param <U> the type returned by the function
1547     */
1548    @FunctionalInterface
1549    public interface ClosingFunction3<
1550        V1 extends @Nullable Object,
1551        V2 extends @Nullable Object,
1552        V3 extends @Nullable Object,
1553        U extends @Nullable Object> {
1554      /**
1555       * Applies this function to three inputs, or throws an exception if unable to do so.
1556       *
1557       * <p>Any objects that are passed to {@link DeferredCloser#eventuallyClose(Closeable,
1558       * Executor) closer.eventuallyClose()} will be closed when the {@link ClosingFuture} pipeline
1559       * is done (but not before this method completes), even if this method throws or the pipeline
1560       * is cancelled.
1561       */
1562      U apply(DeferredCloser closer, V1 value1, V2 value2, V3 v3) throws Exception;
1563    }
1564
1565    /**
1566     * A function that returns a {@link ClosingFuture} when applied to the values of the three
1567     * futures passed to {@link #whenAllSucceed(ClosingFuture, ClosingFuture, ClosingFuture)}.
1568     *
1569     * @param <V1> the type returned by the first future
1570     * @param <V2> the type returned by the second future
1571     * @param <V3> the type returned by the third future
1572     * @param <U> the type returned by the function
1573     */
1574    @FunctionalInterface
1575    public interface AsyncClosingFunction3<
1576        V1 extends @Nullable Object,
1577        V2 extends @Nullable Object,
1578        V3 extends @Nullable Object,
1579        U extends @Nullable Object> {
1580      /**
1581       * Applies this function to three inputs, or throws an exception if unable to do so.
1582       *
1583       * <p>Any objects that are passed to {@link DeferredCloser#eventuallyClose(Closeable,
1584       * Executor) closer.eventuallyClose()} will be closed when the {@link ClosingFuture} pipeline
1585       * is done (but not before this method completes), even if this method throws or the pipeline
1586       * is cancelled.
1587       */
1588      ClosingFuture<U> apply(DeferredCloser closer, V1 value1, V2 value2, V3 value3)
1589          throws Exception;
1590    }
1591
1592    private final ClosingFuture<V1> future1;
1593    private final ClosingFuture<V2> future2;
1594    private final ClosingFuture<V3> future3;
1595
1596    private Combiner3(
1597        ClosingFuture<V1> future1, ClosingFuture<V2> future2, ClosingFuture<V3> future3) {
1598      super(true, ImmutableList.of(future1, future2, future3));
1599      this.future1 = future1;
1600      this.future2 = future2;
1601      this.future3 = future3;
1602    }
1603
1604    /**
1605     * Returns a new {@code ClosingFuture} pipeline step derived from the inputs by applying a
1606     * combining function to their values. The function can use a {@link DeferredCloser} to capture
1607     * objects to be closed when the pipeline is done.
1608     *
1609     * <p>If this combiner was returned by {@link #whenAllSucceed(ClosingFuture, ClosingFuture,
1610     * ClosingFuture)} and any of the inputs fail, so will the returned step.
1611     *
1612     * <p>If the function throws a {@code CancellationException}, the pipeline will be cancelled.
1613     *
1614     * <p>If the function throws an {@code ExecutionException}, the cause of the thrown {@code
1615     * ExecutionException} will be extracted and used as the failure of the derived step.
1616     */
1617    public <U extends @Nullable Object> ClosingFuture<U> call(
1618        final ClosingFunction3<V1, V2, V3, U> function, Executor executor) {
1619      return call(
1620          new CombiningCallable<U>() {
1621            @Override
1622            public U call(DeferredCloser closer, Peeker peeker) throws Exception {
1623              return function.apply(
1624                  closer,
1625                  peeker.getDone(future1),
1626                  peeker.getDone(future2),
1627                  peeker.getDone(future3));
1628            }
1629
1630            @Override
1631            public String toString() {
1632              return function.toString();
1633            }
1634          },
1635          executor);
1636    }
1637
1638    /**
1639     * Returns a new {@code ClosingFuture} pipeline step derived from the inputs by applying a
1640     * {@code ClosingFuture}-returning function to their values. The function can use a {@link
1641     * DeferredCloser} to capture objects to be closed when the pipeline is done (other than those
1642     * captured by the returned {@link ClosingFuture}).
1643     *
1644     * <p>If this combiner was returned by {@link #whenAllSucceed(ClosingFuture, ClosingFuture,
1645     * ClosingFuture)} and any of the inputs fail, so will the returned step.
1646     *
1647     * <p>If the function throws a {@code CancellationException}, the pipeline will be cancelled.
1648     *
1649     * <p>If the function throws an {@code ExecutionException}, the cause of the thrown {@code
1650     * ExecutionException} will be extracted and used as the failure of the derived step.
1651     *
1652     * <p>If the function throws any other exception, it will be used as the failure of the derived
1653     * step.
1654     *
1655     * <p>If an exception is thrown after the function creates a {@code ClosingFuture}, then none of
1656     * the closeable objects in that {@code ClosingFuture} will be closed.
1657     *
1658     * <p>Usage guidelines for this method:
1659     *
1660     * <ul>
1661     *   <li>Use this method only when calling an API that returns a {@link ListenableFuture} or a
1662     *       {@code ClosingFuture}. If possible, prefer calling {@link #call(CombiningCallable,
1663     *       Executor)} instead, with a function that returns the next value directly.
1664     *   <li>Call {@link DeferredCloser#eventuallyClose(Closeable, Executor)
1665     *       closer.eventuallyClose()} for every closeable object this step creates in order to
1666     *       capture it for later closing.
1667     *   <li>Return a {@code ClosingFuture}. To turn a {@link ListenableFuture} into a {@code
1668     *       ClosingFuture} call {@link #from(ListenableFuture)}.
1669     * </ul>
1670     *
1671     * <p>The same warnings about doing heavyweight operations within {@link
1672     * ClosingFuture#transformAsync(AsyncClosingFunction, Executor)} apply here.
1673     */
1674    public <U extends @Nullable Object> ClosingFuture<U> callAsync(
1675        final AsyncClosingFunction3<V1, V2, V3, U> function, Executor executor) {
1676      return callAsync(
1677          new AsyncCombiningCallable<U>() {
1678            @Override
1679            public ClosingFuture<U> call(DeferredCloser closer, Peeker peeker) throws Exception {
1680              return function.apply(
1681                  closer,
1682                  peeker.getDone(future1),
1683                  peeker.getDone(future2),
1684                  peeker.getDone(future3));
1685            }
1686
1687            @Override
1688            public String toString() {
1689              return function.toString();
1690            }
1691          },
1692          executor);
1693    }
1694  }
1695
1696  /**
1697   * A generic {@link Combiner} that lets you use a lambda or method reference to combine four
1698   * {@link ClosingFuture}s. Use {@link #whenAllSucceed(ClosingFuture, ClosingFuture, ClosingFuture,
1699   * ClosingFuture)} to start this combination.
1700   *
1701   * @param <V1> the type returned by the first future
1702   * @param <V2> the type returned by the second future
1703   * @param <V3> the type returned by the third future
1704   * @param <V4> the type returned by the fourth future
1705   */
1706  public static final class Combiner4<
1707          V1 extends @Nullable Object,
1708          V2 extends @Nullable Object,
1709          V3 extends @Nullable Object,
1710          V4 extends @Nullable Object>
1711      extends Combiner {
1712    /**
1713     * A function that returns a value when applied to the values of the four futures passed to
1714     * {@link #whenAllSucceed(ClosingFuture, ClosingFuture, ClosingFuture, ClosingFuture)}.
1715     *
1716     * @param <V1> the type returned by the first future
1717     * @param <V2> the type returned by the second future
1718     * @param <V3> the type returned by the third future
1719     * @param <V4> the type returned by the fourth future
1720     * @param <U> the type returned by the function
1721     */
1722    @FunctionalInterface
1723    public interface ClosingFunction4<
1724        V1 extends @Nullable Object,
1725        V2 extends @Nullable Object,
1726        V3 extends @Nullable Object,
1727        V4 extends @Nullable Object,
1728        U extends @Nullable Object> {
1729      /**
1730       * Applies this function to four inputs, or throws an exception if unable to do so.
1731       *
1732       * <p>Any objects that are passed to {@link DeferredCloser#eventuallyClose(Closeable,
1733       * Executor) closer.eventuallyClose()} will be closed when the {@link ClosingFuture} pipeline
1734       * is done (but not before this method completes), even if this method throws or the pipeline
1735       * is cancelled.
1736       */
1737      U apply(DeferredCloser closer, V1 value1, V2 value2, V3 value3, V4 value4) throws Exception;
1738    }
1739
1740    /**
1741     * A function that returns a {@link ClosingFuture} when applied to the values of the four
1742     * futures passed to {@link #whenAllSucceed(ClosingFuture, ClosingFuture, ClosingFuture,
1743     * ClosingFuture)}.
1744     *
1745     * @param <V1> the type returned by the first future
1746     * @param <V2> the type returned by the second future
1747     * @param <V3> the type returned by the third future
1748     * @param <V4> the type returned by the fourth future
1749     * @param <U> the type returned by the function
1750     */
1751    @FunctionalInterface
1752    public interface AsyncClosingFunction4<
1753        V1 extends @Nullable Object,
1754        V2 extends @Nullable Object,
1755        V3 extends @Nullable Object,
1756        V4 extends @Nullable Object,
1757        U extends @Nullable Object> {
1758      /**
1759       * Applies this function to four inputs, or throws an exception if unable to do so.
1760       *
1761       * <p>Any objects that are passed to {@link DeferredCloser#eventuallyClose(Closeable,
1762       * Executor) closer.eventuallyClose()} will be closed when the {@link ClosingFuture} pipeline
1763       * is done (but not before this method completes), even if this method throws or the pipeline
1764       * is cancelled.
1765       */
1766      ClosingFuture<U> apply(DeferredCloser closer, V1 value1, V2 value2, V3 value3, V4 value4)
1767          throws Exception;
1768    }
1769
1770    private final ClosingFuture<V1> future1;
1771    private final ClosingFuture<V2> future2;
1772    private final ClosingFuture<V3> future3;
1773    private final ClosingFuture<V4> future4;
1774
1775    private Combiner4(
1776        ClosingFuture<V1> future1,
1777        ClosingFuture<V2> future2,
1778        ClosingFuture<V3> future3,
1779        ClosingFuture<V4> future4) {
1780      super(true, ImmutableList.of(future1, future2, future3, future4));
1781      this.future1 = future1;
1782      this.future2 = future2;
1783      this.future3 = future3;
1784      this.future4 = future4;
1785    }
1786
1787    /**
1788     * Returns a new {@code ClosingFuture} pipeline step derived from the inputs by applying a
1789     * combining function to their values. The function can use a {@link DeferredCloser} to capture
1790     * objects to be closed when the pipeline is done.
1791     *
1792     * <p>If this combiner was returned by {@link #whenAllSucceed(ClosingFuture, ClosingFuture,
1793     * ClosingFuture, ClosingFuture)} and any of the inputs fail, so will the returned step.
1794     *
1795     * <p>If the function throws a {@code CancellationException}, the pipeline will be cancelled.
1796     *
1797     * <p>If the function throws an {@code ExecutionException}, the cause of the thrown {@code
1798     * ExecutionException} will be extracted and used as the failure of the derived step.
1799     */
1800    public <U extends @Nullable Object> ClosingFuture<U> call(
1801        final ClosingFunction4<V1, V2, V3, V4, U> function, Executor executor) {
1802      return call(
1803          new CombiningCallable<U>() {
1804            @Override
1805            public U call(DeferredCloser closer, Peeker peeker) throws Exception {
1806              return function.apply(
1807                  closer,
1808                  peeker.getDone(future1),
1809                  peeker.getDone(future2),
1810                  peeker.getDone(future3),
1811                  peeker.getDone(future4));
1812            }
1813
1814            @Override
1815            public String toString() {
1816              return function.toString();
1817            }
1818          },
1819          executor);
1820    }
1821
1822    /**
1823     * Returns a new {@code ClosingFuture} pipeline step derived from the inputs by applying a
1824     * {@code ClosingFuture}-returning function to their values. The function can use a {@link
1825     * DeferredCloser} to capture objects to be closed when the pipeline is done (other than those
1826     * captured by the returned {@link ClosingFuture}).
1827     *
1828     * <p>If this combiner was returned by {@link #whenAllSucceed(ClosingFuture, ClosingFuture,
1829     * ClosingFuture, ClosingFuture)} and any of the inputs fail, so will the returned step.
1830     *
1831     * <p>If the function throws a {@code CancellationException}, the pipeline will be cancelled.
1832     *
1833     * <p>If the function throws an {@code ExecutionException}, the cause of the thrown {@code
1834     * ExecutionException} will be extracted and used as the failure of the derived step.
1835     *
1836     * <p>If the function throws any other exception, it will be used as the failure of the derived
1837     * step.
1838     *
1839     * <p>If an exception is thrown after the function creates a {@code ClosingFuture}, then none of
1840     * the closeable objects in that {@code ClosingFuture} will be closed.
1841     *
1842     * <p>Usage guidelines for this method:
1843     *
1844     * <ul>
1845     *   <li>Use this method only when calling an API that returns a {@link ListenableFuture} or a
1846     *       {@code ClosingFuture}. If possible, prefer calling {@link #call(CombiningCallable,
1847     *       Executor)} instead, with a function that returns the next value directly.
1848     *   <li>Call {@link DeferredCloser#eventuallyClose(Closeable, Executor)
1849     *       closer.eventuallyClose()} for every closeable object this step creates in order to
1850     *       capture it for later closing.
1851     *   <li>Return a {@code ClosingFuture}. To turn a {@link ListenableFuture} into a {@code
1852     *       ClosingFuture} call {@link #from(ListenableFuture)}.
1853     * </ul>
1854     *
1855     * <p>The same warnings about doing heavyweight operations within {@link
1856     * ClosingFuture#transformAsync(AsyncClosingFunction, Executor)} apply here.
1857     */
1858    public <U extends @Nullable Object> ClosingFuture<U> callAsync(
1859        final AsyncClosingFunction4<V1, V2, V3, V4, U> function, Executor executor) {
1860      return callAsync(
1861          new AsyncCombiningCallable<U>() {
1862            @Override
1863            public ClosingFuture<U> call(DeferredCloser closer, Peeker peeker) throws Exception {
1864              return function.apply(
1865                  closer,
1866                  peeker.getDone(future1),
1867                  peeker.getDone(future2),
1868                  peeker.getDone(future3),
1869                  peeker.getDone(future4));
1870            }
1871
1872            @Override
1873            public String toString() {
1874              return function.toString();
1875            }
1876          },
1877          executor);
1878    }
1879  }
1880
1881  /**
1882   * A generic {@link Combiner} that lets you use a lambda or method reference to combine five
1883   * {@link ClosingFuture}s. Use {@link #whenAllSucceed(ClosingFuture, ClosingFuture, ClosingFuture,
1884   * ClosingFuture, ClosingFuture)} to start this combination.
1885   *
1886   * @param <V1> the type returned by the first future
1887   * @param <V2> the type returned by the second future
1888   * @param <V3> the type returned by the third future
1889   * @param <V4> the type returned by the fourth future
1890   * @param <V5> the type returned by the fifth future
1891   */
1892  public static final class Combiner5<
1893          V1 extends @Nullable Object,
1894          V2 extends @Nullable Object,
1895          V3 extends @Nullable Object,
1896          V4 extends @Nullable Object,
1897          V5 extends @Nullable Object>
1898      extends Combiner {
1899    /**
1900     * A function that returns a value when applied to the values of the five futures passed to
1901     * {@link #whenAllSucceed(ClosingFuture, ClosingFuture, ClosingFuture, ClosingFuture,
1902     * ClosingFuture)}.
1903     *
1904     * @param <V1> the type returned by the first future
1905     * @param <V2> the type returned by the second future
1906     * @param <V3> the type returned by the third future
1907     * @param <V4> the type returned by the fourth future
1908     * @param <V5> the type returned by the fifth future
1909     * @param <U> the type returned by the function
1910     */
1911    @FunctionalInterface
1912    public interface ClosingFunction5<
1913        V1 extends @Nullable Object,
1914        V2 extends @Nullable Object,
1915        V3 extends @Nullable Object,
1916        V4 extends @Nullable Object,
1917        V5 extends @Nullable Object,
1918        U extends @Nullable Object> {
1919      /**
1920       * Applies this function to five inputs, or throws an exception if unable to do so.
1921       *
1922       * <p>Any objects that are passed to {@link DeferredCloser#eventuallyClose(Closeable,
1923       * Executor) closer.eventuallyClose()} will be closed when the {@link ClosingFuture} pipeline
1924       * is done (but not before this method completes), even if this method throws or the pipeline
1925       * is cancelled.
1926       */
1927      U apply(DeferredCloser closer, V1 value1, V2 value2, V3 value3, V4 value4, V5 value5)
1928          throws Exception;
1929    }
1930
1931    /**
1932     * A function that returns a {@link ClosingFuture} when applied to the values of the five
1933     * futures passed to {@link #whenAllSucceed(ClosingFuture, ClosingFuture, ClosingFuture,
1934     * ClosingFuture, ClosingFuture)}.
1935     *
1936     * @param <V1> the type returned by the first future
1937     * @param <V2> the type returned by the second future
1938     * @param <V3> the type returned by the third future
1939     * @param <V4> the type returned by the fourth future
1940     * @param <V5> the type returned by the fifth future
1941     * @param <U> the type returned by the function
1942     */
1943    @FunctionalInterface
1944    public interface AsyncClosingFunction5<
1945        V1 extends @Nullable Object,
1946        V2 extends @Nullable Object,
1947        V3 extends @Nullable Object,
1948        V4 extends @Nullable Object,
1949        V5 extends @Nullable Object,
1950        U extends @Nullable Object> {
1951      /**
1952       * Applies this function to five inputs, or throws an exception if unable to do so.
1953       *
1954       * <p>Any objects that are passed to {@link DeferredCloser#eventuallyClose(Closeable,
1955       * Executor) closer.eventuallyClose()} will be closed when the {@link ClosingFuture} pipeline
1956       * is done (but not before this method completes), even if this method throws or the pipeline
1957       * is cancelled.
1958       */
1959      ClosingFuture<U> apply(
1960          DeferredCloser closer, V1 value1, V2 value2, V3 value3, V4 value4, V5 value5)
1961          throws Exception;
1962    }
1963
1964    private final ClosingFuture<V1> future1;
1965    private final ClosingFuture<V2> future2;
1966    private final ClosingFuture<V3> future3;
1967    private final ClosingFuture<V4> future4;
1968    private final ClosingFuture<V5> future5;
1969
1970    private Combiner5(
1971        ClosingFuture<V1> future1,
1972        ClosingFuture<V2> future2,
1973        ClosingFuture<V3> future3,
1974        ClosingFuture<V4> future4,
1975        ClosingFuture<V5> future5) {
1976      super(true, ImmutableList.of(future1, future2, future3, future4, future5));
1977      this.future1 = future1;
1978      this.future2 = future2;
1979      this.future3 = future3;
1980      this.future4 = future4;
1981      this.future5 = future5;
1982    }
1983
1984    /**
1985     * Returns a new {@code ClosingFuture} pipeline step derived from the inputs by applying a
1986     * combining function to their values. The function can use a {@link DeferredCloser} to capture
1987     * objects to be closed when the pipeline is done.
1988     *
1989     * <p>If this combiner was returned by {@link #whenAllSucceed(ClosingFuture, ClosingFuture,
1990     * ClosingFuture, ClosingFuture, ClosingFuture)} and any of the inputs fail, so will the
1991     * returned step.
1992     *
1993     * <p>If the function throws a {@code CancellationException}, the pipeline will be cancelled.
1994     *
1995     * <p>If the function throws an {@code ExecutionException}, the cause of the thrown {@code
1996     * ExecutionException} will be extracted and used as the failure of the derived step.
1997     */
1998    public <U extends @Nullable Object> ClosingFuture<U> call(
1999        final ClosingFunction5<V1, V2, V3, V4, V5, U> function, Executor executor) {
2000      return call(
2001          new CombiningCallable<U>() {
2002            @Override
2003            public U call(DeferredCloser closer, Peeker peeker) throws Exception {
2004              return function.apply(
2005                  closer,
2006                  peeker.getDone(future1),
2007                  peeker.getDone(future2),
2008                  peeker.getDone(future3),
2009                  peeker.getDone(future4),
2010                  peeker.getDone(future5));
2011            }
2012
2013            @Override
2014            public String toString() {
2015              return function.toString();
2016            }
2017          },
2018          executor);
2019    }
2020
2021    /**
2022     * Returns a new {@code ClosingFuture} pipeline step derived from the inputs by applying a
2023     * {@code ClosingFuture}-returning function to their values. The function can use a {@link
2024     * DeferredCloser} to capture objects to be closed when the pipeline is done (other than those
2025     * captured by the returned {@link ClosingFuture}).
2026     *
2027     * <p>If this combiner was returned by {@link #whenAllSucceed(ClosingFuture, ClosingFuture,
2028     * ClosingFuture, ClosingFuture, ClosingFuture)} and any of the inputs fail, so will the
2029     * returned step.
2030     *
2031     * <p>If the function throws a {@code CancellationException}, the pipeline will be cancelled.
2032     *
2033     * <p>If the function throws an {@code ExecutionException}, the cause of the thrown {@code
2034     * ExecutionException} will be extracted and used as the failure of the derived step.
2035     *
2036     * <p>If the function throws any other exception, it will be used as the failure of the derived
2037     * step.
2038     *
2039     * <p>If an exception is thrown after the function creates a {@code ClosingFuture}, then none of
2040     * the closeable objects in that {@code ClosingFuture} will be closed.
2041     *
2042     * <p>Usage guidelines for this method:
2043     *
2044     * <ul>
2045     *   <li>Use this method only when calling an API that returns a {@link ListenableFuture} or a
2046     *       {@code ClosingFuture}. If possible, prefer calling {@link #call(CombiningCallable,
2047     *       Executor)} instead, with a function that returns the next value directly.
2048     *   <li>Call {@link DeferredCloser#eventuallyClose(Closeable, Executor)
2049     *       closer.eventuallyClose()} for every closeable object this step creates in order to
2050     *       capture it for later closing.
2051     *   <li>Return a {@code ClosingFuture}. To turn a {@link ListenableFuture} into a {@code
2052     *       ClosingFuture} call {@link #from(ListenableFuture)}.
2053     * </ul>
2054     *
2055     * <p>The same warnings about doing heavyweight operations within {@link
2056     * ClosingFuture#transformAsync(AsyncClosingFunction, Executor)} apply here.
2057     */
2058    public <U extends @Nullable Object> ClosingFuture<U> callAsync(
2059        final AsyncClosingFunction5<V1, V2, V3, V4, V5, U> function, Executor executor) {
2060      return callAsync(
2061          new AsyncCombiningCallable<U>() {
2062            @Override
2063            public ClosingFuture<U> call(DeferredCloser closer, Peeker peeker) throws Exception {
2064              return function.apply(
2065                  closer,
2066                  peeker.getDone(future1),
2067                  peeker.getDone(future2),
2068                  peeker.getDone(future3),
2069                  peeker.getDone(future4),
2070                  peeker.getDone(future5));
2071            }
2072
2073            @Override
2074            public String toString() {
2075              return function.toString();
2076            }
2077          },
2078          executor);
2079    }
2080  }
2081
2082  @Override
2083  public String toString() {
2084    // TODO(dpb): Better toString, in the style of Futures.transform etc.
2085    return toStringHelper(this).add("state", state.get()).addValue(future).toString();
2086  }
2087
2088  @Override
2089  protected void finalize() {
2090    if (state.get().equals(OPEN)) {
2091      logger.log(SEVERE, "Uh oh! An open ClosingFuture has leaked and will close: {0}", this);
2092      FluentFuture<V> unused = finishToFuture();
2093    }
2094  }
2095
2096  private static void closeQuietly(final AutoCloseable closeable, Executor executor) {
2097    if (closeable == null) {
2098      return;
2099    }
2100    try {
2101      executor.execute(
2102          new Runnable() {
2103            @Override
2104            public void run() {
2105              try {
2106                closeable.close();
2107              } catch (Exception e) {
2108                logger.log(WARNING, "thrown by close()", e);
2109              }
2110            }
2111          });
2112    } catch (RejectedExecutionException e) {
2113      if (logger.isLoggable(WARNING)) {
2114        logger.log(
2115            WARNING, String.format("while submitting close to %s; will close inline", executor), e);
2116      }
2117      closeQuietly(closeable, directExecutor());
2118    }
2119  }
2120
2121  private void checkAndUpdateState(State oldState, State newState) {
2122    checkState(
2123        compareAndUpdateState(oldState, newState),
2124        "Expected state to be %s, but it was %s",
2125        oldState,
2126        newState);
2127  }
2128
2129  private boolean compareAndUpdateState(State oldState, State newState) {
2130    return state.compareAndSet(oldState, newState);
2131  }
2132
2133  // TODO(dpb): Should we use a pair of ArrayLists instead of an IdentityHashMap?
2134  private static final class CloseableList extends IdentityHashMap<AutoCloseable, Executor>
2135      implements Closeable {
2136    private final DeferredCloser closer = new DeferredCloser(this);
2137    private volatile boolean closed;
2138    private volatile CountDownLatch whenClosed;
2139
2140    <V, U> ListenableFuture<U> applyClosingFunction(
2141        ClosingFunction<? super V, U> transformation, V input) throws Exception {
2142      // TODO(dpb): Consider ways to defer closing without creating a separate CloseableList.
2143      CloseableList newCloseables = new CloseableList();
2144      try {
2145        return immediateFuture(transformation.apply(newCloseables.closer, input));
2146      } finally {
2147        add(newCloseables, directExecutor());
2148      }
2149    }
2150
2151    <V, U> FluentFuture<U> applyAsyncClosingFunction(
2152        AsyncClosingFunction<V, U> transformation, V input) throws Exception {
2153      // TODO(dpb): Consider ways to defer closing without creating a separate CloseableList.
2154      CloseableList newCloseables = new CloseableList();
2155      try {
2156        ClosingFuture<U> closingFuture = transformation.apply(newCloseables.closer, input);
2157        closingFuture.becomeSubsumedInto(newCloseables);
2158        return closingFuture.future;
2159      } finally {
2160        add(newCloseables, directExecutor());
2161      }
2162    }
2163
2164    @Override
2165    public void close() {
2166      if (closed) {
2167        return;
2168      }
2169      synchronized (this) {
2170        if (closed) {
2171          return;
2172        }
2173        closed = true;
2174      }
2175      for (Map.Entry<AutoCloseable, Executor> entry : entrySet()) {
2176        closeQuietly(entry.getKey(), entry.getValue());
2177      }
2178      clear();
2179      if (whenClosed != null) {
2180        whenClosed.countDown();
2181      }
2182    }
2183
2184    void add(@Nullable AutoCloseable closeable, Executor executor) {
2185      checkNotNull(executor);
2186      if (closeable == null) {
2187        return;
2188      }
2189      synchronized (this) {
2190        if (!closed) {
2191          put(closeable, executor);
2192          return;
2193        }
2194      }
2195      closeQuietly(closeable, executor);
2196    }
2197
2198    /**
2199     * Returns a latch that reaches zero when this objects' deferred closeables have been closed.
2200     */
2201    CountDownLatch whenClosedCountDown() {
2202      if (closed) {
2203        return new CountDownLatch(0);
2204      }
2205      synchronized (this) {
2206        if (closed) {
2207          return new CountDownLatch(0);
2208        }
2209        checkState(whenClosed == null);
2210        return whenClosed = new CountDownLatch(1);
2211      }
2212    }
2213  }
2214
2215  /**
2216   * Returns an object that can be used to wait until this objects' deferred closeables have all had
2217   * {@link Runnable}s that close them submitted to each one's closing {@link Executor}.
2218   */
2219  @VisibleForTesting
2220  CountDownLatch whenClosedCountDown() {
2221    return closeables.whenClosedCountDown();
2222  }
2223
2224  /** The state of a {@link CloseableList}. */
2225  enum State {
2226    /** The {@link CloseableList} has not been subsumed or closed. */
2227    OPEN,
2228
2229    /**
2230     * The {@link CloseableList} has been subsumed into another. It may not be closed or subsumed
2231     * into any other.
2232     */
2233    SUBSUMED,
2234
2235    /**
2236     * Some {@link ListenableFuture} has a callback attached that will close the {@link
2237     * CloseableList}, but it has not yet run. The {@link CloseableList} may not be subsumed.
2238     */
2239    WILL_CLOSE,
2240
2241    /**
2242     * The callback that closes the {@link CloseableList} is running, but it has not completed. The
2243     * {@link CloseableList} may not be subsumed.
2244     */
2245    CLOSING,
2246
2247    /** The {@link CloseableList} has been closed. It may not be further subsumed. */
2248    CLOSED,
2249
2250    /**
2251     * {@link ClosingFuture#finishToValueAndCloser(ValueAndCloserConsumer, Executor)} has been
2252     * called. The step may not be further subsumed, nor may {@link #finishToFuture()} be called.
2253     */
2254    WILL_CREATE_VALUE_AND_CLOSER,
2255  }
2256}