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