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