1 /*
2 * Copyright (C) 2009 The Guava Authors
3 *
4 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
5 * in compliance with the License. You may obtain a copy of the License at
6 *
7 * http://www.apache.org/licenses/LICENSE-2.0
8 *
9 * Unless required by applicable law or agreed to in writing, software distributed under the License
10 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
11 * or implied. See the License for the specific language governing permissions and limitations under
12 * the License.
13 */
14
15 package com.google.common.util.concurrent;
16
17 import static com.google.common.base.Preconditions.checkNotNull;
18 import static com.google.common.util.concurrent.Uninterruptibles.getUninterruptibly;
19
20 import com.google.common.annotations.Beta;
21 import com.google.common.annotations.GwtIncompatible;
22 import java.util.concurrent.Executor;
23 import java.util.concurrent.Executors;
24 import java.util.concurrent.Future;
25 import java.util.concurrent.ThreadFactory;
26 import java.util.concurrent.atomic.AtomicBoolean;
27 import org.checkerframework.checker.nullness.qual.Nullable;
28
29 /**
30 * Utilities necessary for working with libraries that supply plain {@link Future} instances. Note
31 * that, whenever possible, it is strongly preferred to modify those libraries to return {@code
32 * ListenableFuture} directly.
33 *
34 * @author Sven Mawson
35 * @since 10.0 (replacing {@code Futures.makeListenable}, which existed in 1.0)
36 */
37 @Beta
38 @GwtIncompatible
39 @ElementTypesAreNonnullByDefault
40 public final class JdkFutureAdapters {
41 /**
42 * Assigns a thread to the given {@link Future} to provide {@link ListenableFuture} functionality.
43 *
44 * <p><b>Warning:</b> If the input future does not already implement {@code ListenableFuture}, the
45 * returned future will emulate {@link ListenableFuture#addListener} by taking a thread from an
46 * internal, unbounded pool at the first call to {@code addListener} and holding it until the
47 * future is {@linkplain Future#isDone() done}.
48 *
49 * <p>Prefer to create {@code ListenableFuture} instances with {@link SettableFuture}, {@link
50 * MoreExecutors#listeningDecorator( java.util.concurrent.ExecutorService)}, {@link
51 * ListenableFutureTask}, {@link AbstractFuture}, and other utilities over creating plain {@code
52 * Future} instances to be upgraded to {@code ListenableFuture} after the fact.
53 */
54 public static <V extends @Nullable Object> ListenableFuture<V> listenInPoolThread(
55 Future<V> future) {
56 if (future instanceof ListenableFuture) {
57 return (ListenableFuture<V>) future;
58 }
59 return new ListenableFutureAdapter<V>(future);
60 }
61
62 /**
63 * Submits a blocking task for the given {@link Future} to provide {@link ListenableFuture}
64 * functionality.
65 *
66 * <p><b>Warning:</b> If the input future does not already implement {@code ListenableFuture}, the
67 * returned future will emulate {@link ListenableFuture#addListener} by submitting a task to the
68 * given executor at the first call to {@code addListener}. The task must be started by the
69 * executor promptly, or else the returned {@code ListenableFuture} may fail to work. The task's
70 * execution consists of blocking until the input future is {@linkplain Future#isDone() done}, so
71 * each call to this method may claim and hold a thread for an arbitrary length of time. Use of
72 * bounded executors or other executors that may fail to execute a task promptly may result in
73 * deadlocks.
74 *
75 * <p>Prefer to create {@code ListenableFuture} instances with {@link SettableFuture}, {@link
76 * MoreExecutors#listeningDecorator( java.util.concurrent.ExecutorService)}, {@link
77 * ListenableFutureTask}, {@link AbstractFuture}, and other utilities over creating plain {@code
78 * Future} instances to be upgraded to {@code ListenableFuture} after the fact.
79 *
80 * @since 12.0
81 */
82 public static <V extends @Nullable Object> ListenableFuture<V> listenInPoolThread(
83 Future<V> future, Executor executor) {
84 checkNotNull(executor);
85 if (future instanceof ListenableFuture) {
86 return (ListenableFuture<V>) future;
87 }
88 return new ListenableFutureAdapter<V>(future, executor);
89 }
90
91 /**
92 * An adapter to turn a {@link Future} into a {@link ListenableFuture}. This will wait on the
93 * future to finish, and when it completes, run the listeners. This implementation will wait on
94 * the source future indefinitely, so if the source future never completes, the adapter will never
95 * complete either.
96 *
97 * <p>If the delegate future is interrupted or throws an unexpected unchecked exception, the
98 * listeners will not be invoked.
99 */
100 private static class ListenableFutureAdapter<V extends @Nullable Object>
101 extends ForwardingFuture<V> implements ListenableFuture<V> {
102
103 private static final ThreadFactory threadFactory =
104 new ThreadFactoryBuilder()
105 .setDaemon(true)
106 .setNameFormat("ListenableFutureAdapter-thread-%d")
107 .build();
108 private static final Executor defaultAdapterExecutor =
109 Executors.newCachedThreadPool(threadFactory);
110
111 private final Executor adapterExecutor;
112
113 // The execution list to hold our listeners.
114 private final ExecutionList executionList = new ExecutionList();
115
116 // This allows us to only start up a thread waiting on the delegate future when the first
117 // listener is added.
118 private final AtomicBoolean hasListeners = new AtomicBoolean(false);
119
120 // The delegate future.
121 private final Future<V> delegate;
122
123 ListenableFutureAdapter(Future<V> delegate) {
124 this(delegate, defaultAdapterExecutor);
125 }
126
127 ListenableFutureAdapter(Future<V> delegate, Executor adapterExecutor) {
128 this.delegate = checkNotNull(delegate);
129 this.adapterExecutor = checkNotNull(adapterExecutor);
130 }
131
132 @Override
133 protected Future<V> delegate() {
134 return delegate;
135 }
136
137 @Override
138 public void addListener(Runnable listener, Executor exec) {
139 executionList.add(listener, exec);
140
141 // When a listener is first added, we run a task that will wait for the delegate to finish,
142 // and when it is done will run the listeners.
143 if (hasListeners.compareAndSet(false, true)) {
144 if (delegate.isDone()) {
145 // If the delegate is already done, run the execution list immediately on the current
146 // thread.
147 executionList.execute();
148 return;
149 }
150
151 // TODO(lukes): handle RejectedExecutionException
152 adapterExecutor.execute(
153 new Runnable() {
154 @Override
155 public void run() {
156 try {
157 /*
158 * Threads from our private pool are never interrupted. Threads from a
159 * user-supplied executor might be, but... what can we do? This is another reason
160 * to return a proper ListenableFuture instead of using listenInPoolThread.
161 */
162 getUninterruptibly(delegate);
163 } catch (Throwable e) {
164 // ExecutionException / CancellationException / RuntimeException / Error
165 // The task is presumably done, run the listeners.
166 }
167 executionList.execute();
168 }
169 });
170 }
171 }
172 }
173
174 private JdkFutureAdapters() {}
175 }