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