-
Notifications
You must be signed in to change notification settings - Fork 10.9k
/
ClosingFuture.java
2277 lines (2142 loc) · 95.9 KB
/
ClosingFuture.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
/*
* Copyright (C) 2017 The Guava Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.google.common.util.concurrent;
import static com.google.common.base.Functions.constant;
import static com.google.common.base.MoreObjects.toStringHelper;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;
import static com.google.common.collect.Lists.asList;
import static com.google.common.util.concurrent.ClosingFuture.State.CLOSED;
import static com.google.common.util.concurrent.ClosingFuture.State.CLOSING;
import static com.google.common.util.concurrent.ClosingFuture.State.OPEN;
import static com.google.common.util.concurrent.ClosingFuture.State.SUBSUMED;
import static com.google.common.util.concurrent.ClosingFuture.State.WILL_CLOSE;
import static com.google.common.util.concurrent.ClosingFuture.State.WILL_CREATE_VALUE_AND_CLOSER;
import static com.google.common.util.concurrent.Futures.getDone;
import static com.google.common.util.concurrent.Futures.immediateFuture;
import static com.google.common.util.concurrent.Futures.nonCancellationPropagating;
import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
import static java.util.logging.Level.FINER;
import static java.util.logging.Level.SEVERE;
import static java.util.logging.Level.WARNING;
import com.google.common.annotations.Beta;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Function;
import com.google.common.collect.FluentIterable;
import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.ClosingFuture.Combiner.AsyncCombiningCallable;
import com.google.common.util.concurrent.ClosingFuture.Combiner.CombiningCallable;
import com.google.common.util.concurrent.Futures.FutureCombiner;
import com.google.errorprone.annotations.CanIgnoreReturnValue;
import com.google.errorprone.annotations.DoNotMock;
import com.google.j2objc.annotations.RetainedWith;
import java.io.Closeable;
import java.io.IOException;
import java.util.IdentityHashMap;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.logging.Logger;
import org.checkerframework.checker.nullness.compatqual.NullableDecl;
/**
* A step in a pipeline of an asynchronous computation. When the last step in the computation is
* complete, some objects captured during the computation are closed.
*
* <p>A pipeline of {@code ClosingFuture}s is a tree of steps. Each step represents either an
* asynchronously-computed intermediate value, or else an exception that indicates the failure or
* cancellation of the operation so far. The only way to extract the value or exception from a step
* is by declaring that step to be the last step of the pipeline. Nevertheless, we refer to the
* "value" of a successful step or the "result" (value or exception) of any step.
*
* <ol>
* <li>A pipeline starts at its leaf step (or steps), which is created from either a callable
* block or a {@link ListenableFuture}.
* <li>Each other step is derived from one or more input steps. At each step, zero or more objects
* can be captured for later closing.
* <li>There is one last step (the root of the tree), from which you can extract the final result
* of the computation. After that result is available (or the computation fails), all objects
* captured by any of the steps in the pipeline are closed.
* </ol>
*
* <h3>Starting a pipeline</h3>
*
* Start a {@code ClosingFuture} pipeline {@linkplain #submit(ClosingCallable, Executor) from a
* callable block} that may capture objects for later closing. To start a pipeline from a {@link
* ListenableFuture} that doesn't create resources that should be closed later, you can use {@link
* #from(ListenableFuture)} instead.
*
* <h3>Derived steps</h3>
*
* A {@code ClosingFuture} step can be derived from one or more input {@code ClosingFuture} steps in
* ways similar to {@link FluentFuture}s:
*
* <ul>
* <li>by transforming the value from a successful input step,
* <li>by catching the exception from a failed input step, or
* <li>by combining the results of several input steps.
* </ul>
*
* Each derivation can capture the next value or any intermediate objects for later closing.
*
* <p>A step can be the input to at most one derived step. Once you transform its value, catch its
* exception, or combine it with others, you cannot do anything else with it, including declare it
* to be the last step of the pipeline.
*
* <h4>Transforming</h4>
*
* To derive the next step by asynchronously applying a function to an input step's value, call
* {@link #transform(ClosingFunction, Executor)} or {@link #transformAsync(AsyncClosingFunction,
* Executor)} on the input step.
*
* <h4>Catching</h4>
*
* To derive the next step from a failed input step, call {@link #catching(Class, ClosingFunction,
* Executor)} or {@link #catchingAsync(Class, AsyncClosingFunction, Executor)} on the input step.
*
* <h4>Combining</h4>
*
* To derive a {@code ClosingFuture} from two or more input steps, pass the input steps to {@link
* #whenAllComplete(Iterable)} or {@link #whenAllSucceed(Iterable)} or its overloads.
*
* <h3>Cancelling</h3>
*
* Any step in a pipeline can be {@linkplain #cancel(boolean) cancelled}, even after another step
* has been derived, with the same semantics as cancelling a {@link Future}. In addition, a
* successfully cancelled step will immediately start closing all objects captured for later closing
* by it and by its input steps.
*
* <h3>Ending a pipeline</h3>
*
* Each {@code ClosingFuture} pipeline must be ended. To end a pipeline, decide whether you want to
* close the captured objects automatically or manually.
*
* <h4>Automatically closing</h4>
*
* You can extract a {@link Future} that represents the result of the last step in the pipeline by
* calling {@link #finishToFuture()}. All objects the pipeline has captured for closing will begin
* to be closed asynchronously <b>after</b> the returned {@code Future} is done: the future
* completes before closing starts, rather than once it has finished.
*
* <pre>{@code
* FluentFuture<UserName> userName =
* ClosingFuture.submit(
* closer -> closer.eventuallyClose(database.newTransaction(), closingExecutor),
* executor)
* .transformAsync((closer, transaction) -> transaction.queryClosingFuture("..."), executor)
* .transform((closer, result) -> result.get("userName"), directExecutor())
* .catching(DBException.class, e -> "no user", directExecutor())
* .finishToFuture();
* }</pre>
*
* In this example, when the {@code userName} {@link Future} is done, the transaction and the query
* result cursor will both be closed, even if the operation is cancelled or fails.
*
* <h4>Manually closing</h4>
*
* If you want to close the captured objects manually, after you've used the final result, call
* {@link #finishToValueAndCloser(ValueAndCloserConsumer, Executor)} to get an object that holds the
* final result. You then call {@link ValueAndCloser#closeAsync()} to close the captured objects.
*
* <pre>{@code
* ClosingFuture.submit(
* closer -> closer.eventuallyClose(database.newTransaction(), closingExecutor),
* executor)
* .transformAsync((closer, transaction) -> transaction.queryClosingFuture("..."), executor)
* .transform((closer, result) -> result.get("userName"), directExecutor())
* .catching(DBException.class, e -> "no user", directExecutor())
* .finishToValueAndCloser(
* valueAndCloser -> this.userNameValueAndCloser = valueAndCloser, executor);
*
* // later
* try { // get() will throw if the operation failed or was cancelled.
* UserName userName = userNameValueAndCloser.get();
* // do something with userName
* } finally {
* userNameValueAndCloser.closeAsync();
* }
* }</pre>
*
* In this example, when {@code userNameValueAndCloser.closeAsync()} is called, the transaction and
* the query result cursor will both be closed, even if the operation is cancelled or fails.
*
* <p>Note that if you don't call {@code closeAsync()}, the captured objects will not be closed. The
* automatic-closing approach described above is safer.
*
* @param <V> the type of the value of this step
* @since 30.0
*/
// TODO(dpb): Consider reusing one CloseableList for the entire pipeline, modulo combinations.
@Beta // @Beta for one release.
@DoNotMock("Use ClosingFuture.from(Futures.immediate*Future)")
// TODO(dpb): GWT compatibility.
public final class ClosingFuture<V> {
private static final Logger logger = Logger.getLogger(ClosingFuture.class.getName());
/**
* An object that can capture objects to be closed later, when a {@link ClosingFuture} pipeline is
* done.
*/
public static final class DeferredCloser {
@RetainedWith private final CloseableList list;
DeferredCloser(CloseableList list) {
this.list = list;
}
/**
* Captures an object to be closed when a {@link ClosingFuture} pipeline is done.
*
* <p>For users of the {@code -jre} flavor of Guava, the object can be any {@code
* AutoCloseable}. For users of the {@code -android} flavor, the object must be a {@code
* Closeable}. (For more about the flavors, see <a
* href="https://github.com/google/guava#adding-guava-to-your-build">Adding Guava to your
* build</a>.)
*
* <p>Be careful when targeting an older SDK than you are building against (most commonly when
* building for Android): Ensure that any object you pass implements the interface not just in
* your current SDK version but also at the oldest version you support. For example, <a
* href="https://developer.android.com/sdk/api_diff/16/">API Level 16</a> is the first version
* in which {@code Cursor} is {@code Closeable}. To support older versions, pass a wrapper
* {@code Closeable} with a method reference like {@code cursor::close}.
*
* <p>Note that this method is still binary-compatible between flavors because the erasure of
* its parameter type is {@code Object}, not {@code AutoCloseable} or {@code Closeable}.
*
* @param closeable the object to be closed (see notes above)
* @param closingExecutor the object will be closed on this executor
* @return the first argument
*/
@CanIgnoreReturnValue
@NullableDecl
// TODO(b/163345357): Widen bound to AutoCloseable once we require API Level 19.
public <C extends Object & Closeable> C eventuallyClose(
@NullableDecl C closeable, Executor closingExecutor) {
checkNotNull(closingExecutor);
if (closeable != null) {
list.add(closeable, closingExecutor);
}
return closeable;
}
}
/**
* An operation that computes a result.
*
* @param <V> the type of the result
*/
public interface ClosingCallable<V extends Object> {
/**
* Computes a result, or throws an exception if unable to do so.
*
* <p>Any objects that are passed to {@link DeferredCloser#eventuallyClose(Closeable, Executor)
* closer.eventuallyClose()} will be closed when the {@link ClosingFuture} pipeline is done (but
* not before this method completes), even if this method throws or the pipeline is cancelled.
*/
@NullableDecl
V call(DeferredCloser closer) throws Exception;
}
/**
* An operation that computes a {@link ClosingFuture} of a result.
*
* @param <V> the type of the result
* @since 30.1
*/
public interface AsyncClosingCallable<V extends Object> {
/**
* Computes a result, or throws an exception if unable to do so.
*
* <p>Any objects that are passed to {@link DeferredCloser#eventuallyClose(Closeable, Executor)
* closer.eventuallyClose()} will be closed when the {@link ClosingFuture} pipeline is done (but
* not before this method completes), even if this method throws or the pipeline is cancelled.
*/
ClosingFuture<V> call(DeferredCloser closer) throws Exception;
}
/**
* A function from an input to a result.
*
* @param <T> the type of the input to the function
* @param <U> the type of the result of the function
*/
public interface ClosingFunction<T extends Object, U extends Object> {
/**
* Applies this function to an input, or throws an exception if unable to do so.
*
* <p>Any objects that are passed to {@link DeferredCloser#eventuallyClose(Closeable, Executor)
* closer.eventuallyClose()} will be closed when the {@link ClosingFuture} pipeline is done (but
* not before this method completes), even if this method throws or the pipeline is cancelled.
*/
@NullableDecl
U apply(DeferredCloser closer, @NullableDecl T input) throws Exception;
}
/**
* A function from an input to a {@link ClosingFuture} of a result.
*
* @param <T> the type of the input to the function
* @param <U> the type of the result of the function
*/
public interface AsyncClosingFunction<T extends Object, U extends Object> {
/**
* Applies this function to an input, or throws an exception if unable to do so.
*
* <p>Any objects that are passed to {@link DeferredCloser#eventuallyClose(Closeable, Executor)
* closer.eventuallyClose()} will be closed when the {@link ClosingFuture} pipeline is done (but
* not before this method completes), even if this method throws or the pipeline is cancelled.
*/
ClosingFuture<U> apply(DeferredCloser closer, @NullableDecl T input) throws Exception;
}
/**
* An object that holds the final result of an asynchronous {@link ClosingFuture} operation and
* allows the user to close all the closeable objects that were captured during it for later
* closing.
*
* <p>The asynchronous operation will have completed before this object is created.
*
* @param <V> the type of the value of a successful operation
* @see ClosingFuture#finishToValueAndCloser(ValueAndCloserConsumer, Executor)
*/
public static final class ValueAndCloser<V> {
private final ClosingFuture<? extends V> closingFuture;
ValueAndCloser(ClosingFuture<? extends V> closingFuture) {
this.closingFuture = checkNotNull(closingFuture);
}
/**
* Returns the final value of the associated {@link ClosingFuture}, or throws an exception as
* {@link Future#get()} would.
*
* <p>Because the asynchronous operation has already completed, this method is synchronous and
* returns immediately.
*
* @throws CancellationException if the computation was cancelled
* @throws ExecutionException if the computation threw an exception
*/
@NullableDecl
public V get() throws ExecutionException {
return getDone(closingFuture.future);
}
/**
* Starts closing all closeable objects captured during the {@link ClosingFuture}'s asynchronous
* operation on the {@link Executor}s specified by calls to {@link
* DeferredCloser#eventuallyClose(Closeable, Executor)}.
*
* <p>If any such calls specified {@link MoreExecutors#directExecutor()}, those objects will be
* closed synchronously.
*
* <p>Idempotent: objects will be closed at most once.
*/
public void closeAsync() {
closingFuture.close();
}
}
/**
* Represents an operation that accepts a {@link ValueAndCloser} for the last step in a {@link
* ClosingFuture} pipeline.
*
* @param <V> the type of the final value of a successful pipeline
* @see ClosingFuture#finishToValueAndCloser(ValueAndCloserConsumer, Executor)
*/
public interface ValueAndCloserConsumer<V> {
/** Accepts a {@link ValueAndCloser} for the last step in a {@link ClosingFuture} pipeline. */
void accept(ValueAndCloser<V> valueAndCloser);
}
/**
* Starts a {@link ClosingFuture} pipeline by submitting a callable block to an executor.
*
* @throws java.util.concurrent.RejectedExecutionException if the task cannot be scheduled for
* execution
*/
public static <V> ClosingFuture<V> submit(ClosingCallable<V> callable, Executor executor) {
return new ClosingFuture<>(callable, executor);
}
/**
* Starts a {@link ClosingFuture} pipeline by submitting a callable block to an executor.
*
* @throws java.util.concurrent.RejectedExecutionException if the task cannot be scheduled for
* execution
* @since 30.1
*/
public static <V> ClosingFuture<V> submitAsync(
AsyncClosingCallable<V> callable, Executor executor) {
return new ClosingFuture<>(callable, executor);
}
/**
* Starts a {@link ClosingFuture} pipeline with a {@link ListenableFuture}.
*
* <p>{@code future}'s value will not be closed when the pipeline is done even if {@code V}
* implements {@link Closeable}. In order to start a pipeline with a value that will be closed
* when the pipeline is done, use {@link #submit(ClosingCallable, Executor)} instead.
*/
public static <V> ClosingFuture<V> from(ListenableFuture<V> future) {
return new ClosingFuture<V>(future);
}
/**
* Starts a {@link ClosingFuture} pipeline with a {@link ListenableFuture}.
*
* <p>If {@code future} succeeds, its value will be closed (using {@code closingExecutor)} when
* the pipeline is done, even if the pipeline is canceled or fails.
*
* <p>Cancelling the pipeline will not cancel {@code future}, so that the pipeline can access its
* value in order to close it.
*
* @param future the future to create the {@code ClosingFuture} from. For discussion of the
* future's result type {@code C}, see {@link DeferredCloser#eventuallyClose(Closeable,
* Executor)}.
* @param closingExecutor the future's result will be closed on this executor
* @deprecated Creating {@link Future}s of closeable types is dangerous in general because the
* underlying value may never be closed if the {@link Future} is canceled after its operation
* begins. Consider replacing code that creates {@link ListenableFuture}s of closeable types,
* including those that pass them to this method, with {@link #submit(ClosingCallable,
* Executor)} in order to ensure that resources do not leak. Or, to start a pipeline with a
* {@link ListenableFuture} that doesn't create values that should be closed, use {@link
* ClosingFuture#from}.
*/
@Deprecated
// TODO(b/163345357): Widen bound to AutoCloseable once we require API Level 19.
public static <C extends Object & Closeable> ClosingFuture<C> eventuallyClosing(
ListenableFuture<C> future, final Executor closingExecutor) {
checkNotNull(closingExecutor);
final ClosingFuture<C> closingFuture = new ClosingFuture<>(nonCancellationPropagating(future));
Futures.addCallback(
future,
new FutureCallback<Closeable>() {
@Override
public void onSuccess(@NullableDecl Closeable result) {
closingFuture.closeables.closer.eventuallyClose(result, closingExecutor);
}
@Override
public void onFailure(Throwable t) {}
},
directExecutor());
return closingFuture;
}
/**
* Starts specifying how to combine {@link ClosingFuture}s into a single pipeline.
*
* @throws IllegalStateException if a {@code ClosingFuture} has already been derived from any of
* the {@code futures}, or if any has already been {@linkplain #finishToFuture() finished}
*/
public static Combiner whenAllComplete(Iterable<? extends ClosingFuture<?>> futures) {
return new Combiner(false, futures);
}
/**
* Starts specifying how to combine {@link ClosingFuture}s into a single pipeline.
*
* @throws IllegalStateException if a {@code ClosingFuture} has already been derived from any of
* the arguments, or if any has already been {@linkplain #finishToFuture() finished}
*/
public static Combiner whenAllComplete(
ClosingFuture<?> future1, ClosingFuture<?>... moreFutures) {
return whenAllComplete(asList(future1, moreFutures));
}
/**
* Starts specifying how to combine {@link ClosingFuture}s into a single pipeline, assuming they
* all succeed. If any fail, the resulting pipeline will fail.
*
* @throws IllegalStateException if a {@code ClosingFuture} has already been derived from any of
* the {@code futures}, or if any has already been {@linkplain #finishToFuture() finished}
*/
public static Combiner whenAllSucceed(Iterable<? extends ClosingFuture<?>> futures) {
return new Combiner(true, futures);
}
/**
* Starts specifying how to combine two {@link ClosingFuture}s into a single pipeline, assuming
* they all succeed. If any fail, the resulting pipeline will fail.
*
* <p>Calling this method allows you to use lambdas or method references typed with the types of
* the input {@link ClosingFuture}s.
*
* @throws IllegalStateException if a {@code ClosingFuture} has already been derived from any of
* the arguments, or if any has already been {@linkplain #finishToFuture() finished}
*/
public static <V1, V2> Combiner2<V1, V2> whenAllSucceed(
ClosingFuture<V1> future1, ClosingFuture<V2> future2) {
return new Combiner2<>(future1, future2);
}
/**
* Starts specifying how to combine three {@link ClosingFuture}s into a single pipeline, assuming
* they all succeed. If any fail, the resulting pipeline will fail.
*
* <p>Calling this method allows you to use lambdas or method references typed with the types of
* the input {@link ClosingFuture}s.
*
* @throws IllegalStateException if a {@code ClosingFuture} has already been derived from any of
* the arguments, or if any has already been {@linkplain #finishToFuture() finished}
*/
public static <V1, V2, V3> Combiner3<V1, V2, V3> whenAllSucceed(
ClosingFuture<V1> future1, ClosingFuture<V2> future2, ClosingFuture<V3> future3) {
return new Combiner3<>(future1, future2, future3);
}
/**
* Starts specifying how to combine four {@link ClosingFuture}s into a single pipeline, assuming
* they all succeed. If any fail, the resulting pipeline will fail.
*
* <p>Calling this method allows you to use lambdas or method references typed with the types of
* the input {@link ClosingFuture}s.
*
* @throws IllegalStateException if a {@code ClosingFuture} has already been derived from any of
* the arguments, or if any has already been {@linkplain #finishToFuture() finished}
*/
public static <V1, V2, V3, V4> Combiner4<V1, V2, V3, V4> whenAllSucceed(
ClosingFuture<V1> future1,
ClosingFuture<V2> future2,
ClosingFuture<V3> future3,
ClosingFuture<V4> future4) {
return new Combiner4<>(future1, future2, future3, future4);
}
/**
* Starts specifying how to combine five {@link ClosingFuture}s into a single pipeline, assuming
* they all succeed. If any fail, the resulting pipeline will fail.
*
* <p>Calling this method allows you to use lambdas or method references typed with the types of
* the input {@link ClosingFuture}s.
*
* @throws IllegalStateException if a {@code ClosingFuture} has already been derived from any of
* the arguments, or if any has already been {@linkplain #finishToFuture() finished}
*/
public static <V1, V2, V3, V4, V5> Combiner5<V1, V2, V3, V4, V5> whenAllSucceed(
ClosingFuture<V1> future1,
ClosingFuture<V2> future2,
ClosingFuture<V3> future3,
ClosingFuture<V4> future4,
ClosingFuture<V5> future5) {
return new Combiner5<>(future1, future2, future3, future4, future5);
}
/**
* Starts specifying how to combine {@link ClosingFuture}s into a single pipeline, assuming they
* all succeed. If any fail, the resulting pipeline will fail.
*
* @throws IllegalStateException if a {@code ClosingFuture} has already been derived from any of
* the arguments, or if any has already been {@linkplain #finishToFuture() finished}
*/
public static Combiner whenAllSucceed(
ClosingFuture<?> future1,
ClosingFuture<?> future2,
ClosingFuture<?> future3,
ClosingFuture<?> future4,
ClosingFuture<?> future5,
ClosingFuture<?> future6,
ClosingFuture<?>... moreFutures) {
return whenAllSucceed(
FluentIterable.of(future1, future2, future3, future4, future5, future6)
.append(moreFutures));
}
private final AtomicReference<State> state = new AtomicReference<>(OPEN);
private final CloseableList closeables = new CloseableList();
private final FluentFuture<V> future;
private ClosingFuture(ListenableFuture<V> future) {
this.future = FluentFuture.from(future);
}
private ClosingFuture(final ClosingCallable<V> callable, Executor executor) {
checkNotNull(callable);
TrustedListenableFutureTask<V> task =
TrustedListenableFutureTask.create(
new Callable<V>() {
@Override
public V call() throws Exception {
return callable.call(closeables.closer);
}
@Override
public String toString() {
return callable.toString();
}
});
executor.execute(task);
this.future = task;
}
private ClosingFuture(final AsyncClosingCallable<V> callable, Executor executor) {
checkNotNull(callable);
TrustedListenableFutureTask<V> task =
TrustedListenableFutureTask.create(
new AsyncCallable<V>() {
@Override
public ListenableFuture<V> call() throws Exception {
CloseableList newCloseables = new CloseableList();
try {
ClosingFuture<V> closingFuture = callable.call(newCloseables.closer);
closingFuture.becomeSubsumedInto(closeables);
return closingFuture.future;
} finally {
closeables.add(newCloseables, directExecutor());
}
}
@Override
public String toString() {
return callable.toString();
}
});
executor.execute(task);
this.future = task;
}
/**
* Returns a future that finishes when this step does. Calling {@code get()} on the returned
* future returns {@code null} if the step is successful or throws the same exception that would
* be thrown by calling {@code finishToFuture().get()} if this were the last step. Calling {@code
* cancel()} on the returned future has no effect on the {@code ClosingFuture} pipeline.
*
* <p>{@code statusFuture} differs from most methods on {@code ClosingFuture}: You can make calls
* to {@code statusFuture} <i>in addition to</i> the call you make to {@link #finishToFuture()} or
* a derivation method <i>on the same instance</i>. This is important because calling {@code
* statusFuture} alone does not provide a way to close the pipeline.
*/
public ListenableFuture<?> statusFuture() {
return nonCancellationPropagating(future.transform(constant(null), directExecutor()));
}
/**
* Returns a new {@code ClosingFuture} pipeline step derived from this one by applying a function
* to its value. The function can use a {@link DeferredCloser} to capture objects to be closed
* when the pipeline is done.
*
* <p>If this {@code ClosingFuture} fails, the function will not be called, and the derived {@code
* ClosingFuture} will be equivalent to this one.
*
* <p>If the function throws an exception, that exception is used as the result of the derived
* {@code ClosingFuture}.
*
* <p>Example usage:
*
* <pre>{@code
* ClosingFuture<List<Row>> rowsFuture =
* queryFuture.transform((closer, result) -> result.getRows(), executor);
* }</pre>
*
* <p>When selecting an executor, note that {@code directExecutor} is dangerous in some cases. See
* the discussion in the {@link ListenableFuture#addListener} documentation. All its warnings
* about heavyweight listeners are also applicable to heavyweight functions passed to this method.
*
* <p>After calling this method, you may not call {@link #finishToFuture()}, {@link
* #finishToValueAndCloser(ValueAndCloserConsumer, Executor)}, or any other derivation method on
* this {@code ClosingFuture}.
*
* @param function transforms the value of this step to the value of the derived step
* @param executor executor to run the function in
* @return the derived step
* @throws IllegalStateException if a {@code ClosingFuture} has already been derived from this
* one, or if this {@code ClosingFuture} has already been {@linkplain #finishToFuture()
* finished}
*/
public <U> ClosingFuture<U> transform(
final ClosingFunction<? super V, U> function, Executor executor) {
checkNotNull(function);
AsyncFunction<V, U> applyFunction =
new AsyncFunction<V, U>() {
@Override
public ListenableFuture<U> apply(V input) throws Exception {
return closeables.applyClosingFunction(function, input);
}
@Override
public String toString() {
return function.toString();
}
};
// TODO(dpb): Switch to future.transformSync when that exists (passing a throwing function).
return derive(future.transformAsync(applyFunction, executor));
}
/**
* Returns a new {@code ClosingFuture} pipeline step derived from this one by applying a function
* that returns a {@code ClosingFuture} to its value. The function can use a {@link
* DeferredCloser} to capture objects to be closed when the pipeline is done (other than those
* captured by the returned {@link ClosingFuture}).
*
* <p>If this {@code ClosingFuture} succeeds, the derived one will be equivalent to the one
* returned by the function.
*
* <p>If this {@code ClosingFuture} fails, the function will not be called, and the derived {@code
* ClosingFuture} will be equivalent to this one.
*
* <p>If the function throws an exception, that exception is used as the result of the derived
* {@code ClosingFuture}. But if the exception is thrown after the function creates a {@code
* ClosingFuture}, then none of the closeable objects in that {@code ClosingFuture} will be
* closed.
*
* <p>Usage guidelines for this method:
*
* <ul>
* <li>Use this method only when calling an API that returns a {@link ListenableFuture} or a
* {@code ClosingFuture}. If possible, prefer calling {@link #transform(ClosingFunction,
* Executor)} instead, with a function that returns the next value directly.
* <li>Call {@link DeferredCloser#eventuallyClose(Closeable, Executor) closer.eventuallyClose()}
* for every closeable object this step creates in order to capture it for later closing.
* <li>Return a {@code ClosingFuture}. To turn a {@link ListenableFuture} into a {@code
* ClosingFuture} call {@link #from(ListenableFuture)}.
* <li>In case this step doesn't create new closeables, you can adapt an API that returns a
* {@link ListenableFuture} to return a {@code ClosingFuture} by wrapping it with a call to
* {@link #withoutCloser(AsyncFunction)}
* </ul>
*
* <p>Example usage:
*
* <pre>{@code
* // Result.getRowsClosingFuture() returns a ClosingFuture.
* ClosingFuture<List<Row>> rowsFuture =
* queryFuture.transformAsync((closer, result) -> result.getRowsClosingFuture(), executor);
*
* // Result.writeRowsToOutputStreamFuture() returns a ListenableFuture that resolves to the
* // number of written rows. openOutputFile() returns a FileOutputStream (which implements
* // Closeable).
* ClosingFuture<Integer> rowsFuture2 =
* queryFuture.transformAsync(
* (closer, result) -> {
* FileOutputStream fos = closer.eventuallyClose(openOutputFile(), closingExecutor);
* return ClosingFuture.from(result.writeRowsToOutputStreamFuture(fos));
* },
* executor);
*
* // Result.getRowsFuture() returns a ListenableFuture (no new closeables are created).
* ClosingFuture<List<Row>> rowsFuture3 =
* queryFuture.transformAsync(withoutCloser(Result::getRowsFuture), executor);
*
* }</pre>
*
* <p>When selecting an executor, note that {@code directExecutor} is dangerous in some cases. See
* the discussion in the {@link ListenableFuture#addListener} documentation. All its warnings
* about heavyweight listeners are also applicable to heavyweight functions passed to this method.
* (Specifically, {@code directExecutor} functions should avoid heavyweight operations inside
* {@code AsyncClosingFunction.apply}. Any heavyweight operations should occur in other threads
* responsible for completing the returned {@code ClosingFuture}.)
*
* <p>After calling this method, you may not call {@link #finishToFuture()}, {@link
* #finishToValueAndCloser(ValueAndCloserConsumer, Executor)}, or any other derivation method on
* this {@code ClosingFuture}.
*
* @param function transforms the value of this step to a {@code ClosingFuture} with the value of
* the derived step
* @param executor executor to run the function in
* @return the derived step
* @throws IllegalStateException if a {@code ClosingFuture} has already been derived from this
* one, or if this {@code ClosingFuture} has already been {@linkplain #finishToFuture()
* finished}
*/
public <U> ClosingFuture<U> transformAsync(
final AsyncClosingFunction<? super V, U> function, Executor executor) {
checkNotNull(function);
AsyncFunction<V, U> applyFunction =
new AsyncFunction<V, U>() {
@Override
public ListenableFuture<U> apply(V input) throws Exception {
return closeables.applyAsyncClosingFunction(function, input);
}
@Override
public String toString() {
return function.toString();
}
};
return derive(future.transformAsync(applyFunction, executor));
}
/**
* Returns an {@link AsyncClosingFunction} that applies an {@link AsyncFunction} to an input,
* ignoring the DeferredCloser and returning a {@code ClosingFuture} derived from the returned
* {@link ListenableFuture}.
*
* <p>Use this method to pass a transformation to {@link #transformAsync(AsyncClosingFunction,
* Executor)} or to {@link #catchingAsync(Class, AsyncClosingFunction, Executor)} as long as it
* meets these conditions:
*
* <ul>
* <li>It does not need to capture any {@link Closeable} objects by calling {@link
* DeferredCloser#eventuallyClose(Closeable, Executor)}.
* <li>It returns a {@link ListenableFuture}.
* </ul>
*
* <p>Example usage:
*
* <pre>{@code
* // Result.getRowsFuture() returns a ListenableFuture.
* ClosingFuture<List<Row>> rowsFuture =
* queryFuture.transformAsync(withoutCloser(Result::getRowsFuture), executor);
* }</pre>
*
* @param function transforms the value of a {@code ClosingFuture} step to a {@link
* ListenableFuture} with the value of a derived step
*/
public static <V, U> AsyncClosingFunction<V, U> withoutCloser(
final AsyncFunction<V, U> function) {
checkNotNull(function);
return new AsyncClosingFunction<V, U>() {
@Override
public ClosingFuture<U> apply(DeferredCloser closer, V input) throws Exception {
return ClosingFuture.from(function.apply(input));
}
};
}
/**
* Returns a new {@code ClosingFuture} pipeline step derived from this one by applying a function
* to its exception if it is an instance of a given exception type. The function can use a {@link
* DeferredCloser} to capture objects to be closed when the pipeline is done.
*
* <p>If this {@code ClosingFuture} succeeds or fails with a different exception type, the
* function will not be called, and the derived {@code ClosingFuture} will be equivalent to this
* one.
*
* <p>If the function throws an exception, that exception is used as the result of the derived
* {@code ClosingFuture}.
*
* <p>Example usage:
*
* <pre>{@code
* ClosingFuture<QueryResult> queryFuture =
* queryFuture.catching(
* QueryException.class, (closer, x) -> Query.emptyQueryResult(), executor);
* }</pre>
*
* <p>When selecting an executor, note that {@code directExecutor} is dangerous in some cases. See
* the discussion in the {@link ListenableFuture#addListener} documentation. All its warnings
* about heavyweight listeners are also applicable to heavyweight functions passed to this method.
*
* <p>After calling this method, you may not call {@link #finishToFuture()}, {@link
* #finishToValueAndCloser(ValueAndCloserConsumer, Executor)}, or any other derivation method on
* this {@code ClosingFuture}.
*
* @param exceptionType the exception type that triggers use of {@code fallback}. The exception
* type is matched against this step's exception. "This step's exception" means the cause of
* the {@link ExecutionException} thrown by {@link Future#get()} on the {@link Future}
* underlying this step or, if {@code get()} throws a different kind of exception, that
* exception itself. To avoid hiding bugs and other unrecoverable errors, callers should
* prefer more specific types, avoiding {@code Throwable.class} in particular.
* @param fallback the function to be called if this step fails with the expected exception type.
* The function's argument is this step's exception. "This step's exception" means the cause
* of the {@link ExecutionException} thrown by {@link Future#get()} on the {@link Future}
* underlying this step or, if {@code get()} throws a different kind of exception, that
* exception itself.
* @param executor the executor that runs {@code fallback} if the input fails
*/
public <X extends Throwable> ClosingFuture<V> catching(
Class<X> exceptionType, ClosingFunction<? super X, ? extends V> fallback, Executor executor) {
return catchingMoreGeneric(exceptionType, fallback, executor);
}
// Avoids generic type capture inconsistency problems where |? extends V| is incompatible with V.
private <X extends Throwable, W extends V> ClosingFuture<V> catchingMoreGeneric(
Class<X> exceptionType, final ClosingFunction<? super X, W> fallback, Executor executor) {
checkNotNull(fallback);
AsyncFunction<X, W> applyFallback =
new AsyncFunction<X, W>() {
@Override
public ListenableFuture<W> apply(X exception) throws Exception {
return closeables.applyClosingFunction(fallback, exception);
}
@Override
public String toString() {
return fallback.toString();
}
};
// TODO(dpb): Switch to future.catchingSync when that exists (passing a throwing function).
return derive(future.catchingAsync(exceptionType, applyFallback, executor));
}
/**
* Returns a new {@code ClosingFuture} pipeline step derived from this one by applying a function
* that returns a {@code ClosingFuture} to its exception if it is an instance of a given exception
* type. The function can use a {@link DeferredCloser} to capture objects to be closed when the
* pipeline is done (other than those captured by the returned {@link ClosingFuture}).
*
* <p>If this {@code ClosingFuture} fails with an exception of the given type, the derived {@code
* ClosingFuture} will be equivalent to the one returned by the function.
*
* <p>If this {@code ClosingFuture} succeeds or fails with a different exception type, the
* function will not be called, and the derived {@code ClosingFuture} will be equivalent to this
* one.
*
* <p>If the function throws an exception, that exception is used as the result of the derived
* {@code ClosingFuture}. But if the exception is thrown after the function creates a {@code
* ClosingFuture}, then none of the closeable objects in that {@code ClosingFuture} will be
* closed.
*
* <p>Usage guidelines for this method:
*
* <ul>
* <li>Use this method only when calling an API that returns a {@link ListenableFuture} or a
* {@code ClosingFuture}. If possible, prefer calling {@link #catching(Class,
* ClosingFunction, Executor)} instead, with a function that returns the next value
* directly.
* <li>Call {@link DeferredCloser#eventuallyClose(Closeable, Executor) closer.eventuallyClose()}
* for every closeable object this step creates in order to capture it for later closing.
* <li>Return a {@code ClosingFuture}. To turn a {@link ListenableFuture} into a {@code
* ClosingFuture} call {@link #from(ListenableFuture)}.
* <li>In case this step doesn't create new closeables, you can adapt an API that returns a
* {@link ListenableFuture} to return a {@code ClosingFuture} by wrapping it with a call to
* {@link #withoutCloser(AsyncFunction)}
* </ul>
*
* <p>Example usage:
*
* <pre>{@code
* // Fall back to a secondary input stream in case of IOException.
* ClosingFuture<InputStream> inputFuture =
* firstInputFuture.catchingAsync(
* IOException.class, (closer, x) -> secondaryInputStreamClosingFuture(), executor);
* }
* }</pre>
*
* <p>When selecting an executor, note that {@code directExecutor} is dangerous in some cases. See
* the discussion in the {@link ListenableFuture#addListener} documentation. All its warnings
* about heavyweight listeners are also applicable to heavyweight functions passed to this method.
* (Specifically, {@code directExecutor} functions should avoid heavyweight operations inside
* {@code AsyncClosingFunction.apply}. Any heavyweight operations should occur in other threads
* responsible for completing the returned {@code ClosingFuture}.)
*
* <p>After calling this method, you may not call {@link #finishToFuture()}, {@link
* #finishToValueAndCloser(ValueAndCloserConsumer, Executor)}, or any other derivation method on
* this {@code ClosingFuture}.
*
* @param exceptionType the exception type that triggers use of {@code fallback}. The exception
* type is matched against this step's exception. "This step's exception" means the cause of
* the {@link ExecutionException} thrown by {@link Future#get()} on the {@link Future}
* underlying this step or, if {@code get()} throws a different kind of exception, that
* exception itself. To avoid hiding bugs and other unrecoverable errors, callers should
* prefer more specific types, avoiding {@code Throwable.class} in particular.
* @param fallback the function to be called if this step fails with the expected exception type.
* The function's argument is this step's exception. "This step's exception" means the cause
* of the {@link ExecutionException} thrown by {@link Future#get()} on the {@link Future}
* underlying this step or, if {@code get()} throws a different kind of exception, that
* exception itself.
* @param executor the executor that runs {@code fallback} if the input fails
*/
// TODO(dpb): Should this do something special if the function throws CancellationException or
// ExecutionException?
public <X extends Throwable> ClosingFuture<V> catchingAsync(
Class<X> exceptionType,
AsyncClosingFunction<? super X, ? extends V> fallback,
Executor executor) {
return catchingAsyncMoreGeneric(exceptionType, fallback, executor);
}
// Avoids generic type capture inconsistency problems where |? extends V| is incompatible with V.
private <X extends Throwable, W extends V> ClosingFuture<V> catchingAsyncMoreGeneric(
Class<X> exceptionType,
final AsyncClosingFunction<? super X, W> fallback,
Executor executor) {
checkNotNull(fallback);
AsyncFunction<X, W> asyncFunction =
new AsyncFunction<X, W>() {
@Override
public ListenableFuture<W> apply(X exception) throws Exception {
return closeables.applyAsyncClosingFunction(fallback, exception);
}
@Override
public String toString() {
return fallback.toString();
}
};
return derive(future.catchingAsync(exceptionType, asyncFunction, executor));
}
/**
* Marks this step as the last step in the {@code ClosingFuture} pipeline.
*
* <p>The returned {@link Future} is completed when the pipeline's computation completes, or when
* the pipeline is cancelled.
*
* <p>All objects the pipeline has captured for closing will begin to be closed asynchronously
* <b>after</b> the returned {@code Future} is done: the future completes before closing starts,
* rather than once it has finished.
*
* <p>After calling this method, you may not call {@link
* #finishToValueAndCloser(ValueAndCloserConsumer, Executor)}, this method, or any other
* derivation method on this {@code ClosingFuture}.
*
* @return a {@link Future} that represents the final value or exception of the pipeline