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