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.GwtIncompatible;
021import com.google.common.annotations.J2ktIncompatible;
022import java.util.concurrent.Executor;
023import java.util.concurrent.Executors;
024import java.util.concurrent.Future;
025import java.util.concurrent.ThreadFactory;
026import java.util.concurrent.atomic.AtomicBoolean;
027import org.jspecify.annotations.Nullable;
028
029/**
030 * Utilities necessary for working with libraries that supply plain {@link Future} instances. Note
031 * that, whenever possible, it is strongly preferred to modify those libraries to return {@code
032 * ListenableFuture} directly.
033 *
034 * <p>For interoperability between {@code ListenableFuture} and <b>{@code CompletableFuture}</b>,
035 * consider <a href="https://github.com/lukas-krecan/future-converter">Future Converter</a>.
036 *
037 * @author Sven Mawson
038 * @since 10.0 (replacing {@code Futures.makeListenable}, which existed in 1.0)
039 */
040@J2ktIncompatible
041@GwtIncompatible
042public final class JdkFutureAdapters {
043  /**
044   * Assigns a thread to the given {@link Future} to provide {@link ListenableFuture} functionality.
045   *
046   * <p><b>Warning:</b> If the input future does not already implement {@code ListenableFuture}, the
047   * returned future will emulate {@link ListenableFuture#addListener} by taking a thread from an
048   * internal, unbounded pool at the first call to {@code addListener} and holding it until the
049   * future is {@linkplain Future#isDone() done}.
050   *
051   * <p>Prefer to create {@code ListenableFuture} instances with {@link SettableFuture}, {@link
052   * MoreExecutors#listeningDecorator( java.util.concurrent.ExecutorService)}, {@link
053   * ListenableFutureTask}, {@link AbstractFuture}, and other utilities over creating plain {@code
054   * Future} instances to be upgraded to {@code ListenableFuture} after the fact.
055   */
056  public static <V extends @Nullable Object> ListenableFuture<V> listenInPoolThread(
057      Future<V> future) {
058    if (future instanceof ListenableFuture) {
059      return (ListenableFuture<V>) future;
060    }
061    return new ListenableFutureAdapter<>(future);
062  }
063
064  /**
065   * Submits a blocking task for the given {@link Future} to provide {@link ListenableFuture}
066   * functionality.
067   *
068   * <p><b>Warning:</b> If the input future does not already implement {@code ListenableFuture}, the
069   * returned future will emulate {@link ListenableFuture#addListener} by submitting a task to the
070   * given executor at the first call to {@code addListener}. The task must be started by the
071   * executor promptly, or else the returned {@code ListenableFuture} may fail to work. The task's
072   * execution consists of blocking until the input future is {@linkplain Future#isDone() done}, so
073   * each call to this method may claim and hold a thread for an arbitrary length of time. Use of
074   * bounded executors or other executors that may fail to execute a task promptly may result in
075   * deadlocks.
076   *
077   * <p>Prefer to create {@code ListenableFuture} instances with {@link SettableFuture}, {@link
078   * MoreExecutors#listeningDecorator( java.util.concurrent.ExecutorService)}, {@link
079   * ListenableFutureTask}, {@link AbstractFuture}, and other utilities over creating plain {@code
080   * Future} instances to be upgraded to {@code ListenableFuture} after the fact.
081   *
082   * @since 12.0
083   */
084  public static <V extends @Nullable Object> ListenableFuture<V> listenInPoolThread(
085      Future<V> future, Executor executor) {
086    checkNotNull(executor);
087    if (future instanceof ListenableFuture) {
088      return (ListenableFuture<V>) future;
089    }
090    return new ListenableFutureAdapter<>(future, executor);
091  }
092
093  /**
094   * An adapter to turn a {@link Future} into a {@link ListenableFuture}. This will wait on the
095   * future to finish, and when it completes, run the listeners. This implementation will wait on
096   * the source future indefinitely, so if the source future never completes, the adapter will never
097   * complete either.
098   *
099   * <p>If the delegate future is interrupted or throws an unexpected unchecked exception, the
100   * listeners will not be invoked.
101   */
102  private static class ListenableFutureAdapter<V extends @Nullable Object>
103      extends ForwardingFuture<V> implements ListenableFuture<V> {
104
105    private static final ThreadFactory threadFactory =
106        new ThreadFactoryBuilder()
107            .setDaemon(true)
108            .setNameFormat("ListenableFutureAdapter-thread-%d")
109            .build();
110    private static final Executor defaultAdapterExecutor =
111        Executors.newCachedThreadPool(threadFactory);
112
113    private final Executor adapterExecutor;
114
115    // The execution list to hold our listeners.
116    private final ExecutionList executionList = new ExecutionList();
117
118    // This allows us to only start up a thread waiting on the delegate future when the first
119    // listener is added.
120    private final AtomicBoolean hasListeners = new AtomicBoolean(false);
121
122    // The delegate future.
123    private final Future<V> delegate;
124
125    ListenableFutureAdapter(Future<V> delegate) {
126      this(delegate, defaultAdapterExecutor);
127    }
128
129    ListenableFutureAdapter(Future<V> delegate, Executor adapterExecutor) {
130      this.delegate = checkNotNull(delegate);
131      this.adapterExecutor = checkNotNull(adapterExecutor);
132    }
133
134    @Override
135    protected Future<V> delegate() {
136      return delegate;
137    }
138
139    @Override
140    public void addListener(Runnable listener, Executor exec) {
141      executionList.add(listener, exec);
142
143      // When a listener is first added, we run a task that will wait for the delegate to finish,
144      // and when it is done will run the listeners.
145      if (hasListeners.compareAndSet(false, true)) {
146        if (delegate.isDone()) {
147          // If the delegate is already done, run the execution list immediately on the current
148          // thread.
149          executionList.execute();
150          return;
151        }
152
153        // TODO(lukes): handle RejectedExecutionException
154        adapterExecutor.execute(
155            () -> {
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 t) {
164                // (including CancellationException and sneaky checked exception)
165                // The task is presumably done, run the listeners.
166                // TODO(cpovirk): Do *something* in case of Error (and maybe
167                // non-CancellationException, non-ExecutionException exceptions)?
168              }
169              executionList.execute();
170            });
171      }
172    }
173  }
174
175  private JdkFutureAdapters() {}
176}