001/*
002 * Copyright (C) 2009 The Guava Authors
003 *
004 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
005 * in compliance with the License. You may obtain a copy of the License at
006 *
007 * http://www.apache.org/licenses/LICENSE-2.0
008 *
009 * Unless required by applicable law or agreed to in writing, software distributed under the License
010 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
011 * or implied. See the License for the specific language governing permissions and limitations under
012 * the License.
013 */
014
015package com.google.common.util.concurrent;
016
017import static com.google.common.base.Preconditions.checkNotNull;
018import static com.google.common.util.concurrent.Uninterruptibles.getUninterruptibly;
019
020import com.google.common.annotations.Beta;
021import com.google.common.annotations.GwtIncompatible;
022import java.util.concurrent.ExecutionException;
023import java.util.concurrent.Executor;
024import java.util.concurrent.Executors;
025import java.util.concurrent.Future;
026import java.util.concurrent.ThreadFactory;
027import java.util.concurrent.atomic.AtomicBoolean;
028import org.checkerframework.checker.nullness.qual.Nullable;
029
030/**
031 * Utilities necessary for working with libraries that supply plain {@link Future} instances. Note
032 * that, whenever possible, it is strongly preferred to modify those libraries to return {@code
033 * ListenableFuture} directly.
034 *
035 * @author Sven Mawson
036 * @since 10.0 (replacing {@code Futures.makeListenable}, which existed in 1.0)
037 */
038@Beta
039@GwtIncompatible
040@ElementTypesAreNonnullByDefault
041public final class JdkFutureAdapters {
042  /**
043   * Assigns a thread to the given {@link Future} to provide {@link ListenableFuture} functionality.
044   *
045   * <p><b>Warning:</b> If the input future does not already implement {@code ListenableFuture}, the
046   * returned future will emulate {@link ListenableFuture#addListener} by taking a thread from an
047   * internal, unbounded pool at the first call to {@code addListener} and holding it until the
048   * future is {@linkplain Future#isDone() done}.
049   *
050   * <p>Prefer to create {@code ListenableFuture} instances with {@link SettableFuture}, {@link
051   * MoreExecutors#listeningDecorator( java.util.concurrent.ExecutorService)}, {@link
052   * ListenableFutureTask}, {@link AbstractFuture}, and other utilities over creating plain {@code
053   * Future} instances to be upgraded to {@code ListenableFuture} after the fact.
054   */
055  public static <V extends @Nullable Object> ListenableFuture<V> listenInPoolThread(
056      Future<V> future) {
057    if (future instanceof ListenableFuture) {
058      return (ListenableFuture<V>) future;
059    }
060    return new ListenableFutureAdapter<>(future);
061  }
062
063  /**
064   * Submits a blocking task for the given {@link Future} to provide {@link ListenableFuture}
065   * functionality.
066   *
067   * <p><b>Warning:</b> If the input future does not already implement {@code ListenableFuture}, the
068   * returned future will emulate {@link ListenableFuture#addListener} by submitting a task to the
069   * given executor at the first call to {@code addListener}. The task must be started by the
070   * executor promptly, or else the returned {@code ListenableFuture} may fail to work. The task's
071   * execution consists of blocking until the input future is {@linkplain Future#isDone() done}, so
072   * each call to this method may claim and hold a thread for an arbitrary length of time. Use of
073   * bounded executors or other executors that may fail to execute a task promptly may result in
074   * deadlocks.
075   *
076   * <p>Prefer to create {@code ListenableFuture} instances with {@link SettableFuture}, {@link
077   * MoreExecutors#listeningDecorator( java.util.concurrent.ExecutorService)}, {@link
078   * ListenableFutureTask}, {@link AbstractFuture}, and other utilities over creating plain {@code
079   * Future} instances to be upgraded to {@code ListenableFuture} after the fact.
080   *
081   * @since 12.0
082   */
083  public static <V extends @Nullable Object> ListenableFuture<V> listenInPoolThread(
084      Future<V> future, Executor executor) {
085    checkNotNull(executor);
086    if (future instanceof ListenableFuture) {
087      return (ListenableFuture<V>) future;
088    }
089    return new ListenableFutureAdapter<>(future, executor);
090  }
091
092  /**
093   * An adapter to turn a {@link Future} into a {@link ListenableFuture}. This will wait on the
094   * future to finish, and when it completes, run the listeners. This implementation will wait on
095   * the source future indefinitely, so if the source future never completes, the adapter will never
096   * complete either.
097   *
098   * <p>If the delegate future is interrupted or throws an unexpected unchecked exception, the
099   * listeners will not be invoked.
100   */
101  private static class ListenableFutureAdapter<V extends @Nullable Object>
102      extends ForwardingFuture<V> implements ListenableFuture<V> {
103
104    private static final ThreadFactory threadFactory =
105        new ThreadFactoryBuilder()
106            .setDaemon(true)
107            .setNameFormat("ListenableFutureAdapter-thread-%d")
108            .build();
109    private static final Executor defaultAdapterExecutor =
110        Executors.newCachedThreadPool(threadFactory);
111
112    private final Executor adapterExecutor;
113
114    // The execution list to hold our listeners.
115    private final ExecutionList executionList = new ExecutionList();
116
117    // This allows us to only start up a thread waiting on the delegate future when the first
118    // listener is added.
119    private final AtomicBoolean hasListeners = new AtomicBoolean(false);
120
121    // The delegate future.
122    private final Future<V> delegate;
123
124    ListenableFutureAdapter(Future<V> delegate) {
125      this(delegate, defaultAdapterExecutor);
126    }
127
128    ListenableFutureAdapter(Future<V> delegate, Executor adapterExecutor) {
129      this.delegate = checkNotNull(delegate);
130      this.adapterExecutor = checkNotNull(adapterExecutor);
131    }
132
133    @Override
134    protected Future<V> delegate() {
135      return delegate;
136    }
137
138    @Override
139    public void addListener(Runnable listener, Executor exec) {
140      executionList.add(listener, exec);
141
142      // When a listener is first added, we run a task that will wait for the delegate to finish,
143      // and when it is done will run the listeners.
144      if (hasListeners.compareAndSet(false, true)) {
145        if (delegate.isDone()) {
146          // If the delegate is already done, run the execution list immediately on the current
147          // thread.
148          executionList.execute();
149          return;
150        }
151
152        // TODO(lukes): handle RejectedExecutionException
153        adapterExecutor.execute(
154            () -> {
155              try {
156                /*
157                 * Threads from our private pool are never interrupted. Threads from a
158                 * user-supplied executor might be, but... what can we do? This is another reason
159                 * to return a proper ListenableFuture instead of using listenInPoolThread.
160                 */
161                getUninterruptibly(delegate);
162              } catch (ExecutionException | RuntimeException | Error e) {
163                // (including CancellationException)
164                // The task is presumably done, run the listeners.
165                // TODO(cpovirk): Do *something* in case of Error (and maybe RuntimeException)?
166              }
167              executionList.execute();
168            });
169      }
170    }
171  }
172
173  private JdkFutureAdapters() {}
174}