11d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert/* 21d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * Copyright (C) 2009 The Guava Authors 31d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * 41d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * Licensed under the Apache License, Version 2.0 (the "License"); 51d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * you may not use this file except in compliance with the License. 61d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * You may obtain a copy of the License at 71d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * 81d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * http://www.apache.org/licenses/LICENSE-2.0 91d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * 101d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * Unless required by applicable law or agreed to in writing, software 111d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * distributed under the License is distributed on an "AS IS" BASIS, 121d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 131d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * See the License for the specific language governing permissions and 141d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * limitations under the License. 151d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert */ 161d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert 171d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringertpackage com.google.common.util.concurrent; 181d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert 191d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringertimport static com.google.common.base.Preconditions.checkNotNull; 201d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert 211d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringertimport com.google.common.annotations.Beta; 221d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringertimport com.google.common.annotations.VisibleForTesting; 231d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert 241d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringertimport java.util.concurrent.Executor; 251d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringertimport java.util.concurrent.Executors; 261d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringertimport java.util.concurrent.Future; 271d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringertimport java.util.concurrent.ThreadFactory; 281d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringertimport java.util.concurrent.atomic.AtomicBoolean; 291d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert 301d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert/** 311d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * Utilities necessary for working with libraries that supply plain {@link 321d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * Future} instances. Note that, whenver possible, it is strongly preferred to 331d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * modify those libraries to return {@code ListenableFuture} directly. 341d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * 351d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * @author Sven Mawson 361d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * @since 10.0 (replacing {@code Futures.makeListenable}, which 371d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * existed in 1.0) 381d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert */ 391d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert@Beta 401d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringertpublic final class JdkFutureAdapters { 411d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert /** 421d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * Assigns a thread to the given {@link Future} to provide {@link 431d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * ListenableFuture} functionality. 441d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * 451d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * <p><b>Warning:</b> If the input future does not already implement {@link 461d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * ListenableFuture}, the returned future will emulate {@link 471d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * ListenableFuture#addListener} by taking a thread from an internal, 481d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * unbounded pool at the first call to {@code addListener} and holding it 491d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * until the future is {@linkplain Future#isDone() done}. 501d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * 511d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * <p>Prefer to create {@code ListenableFuture} instances with {@link 521d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * SettableFuture}, {@link MoreExecutors#listeningDecorator( 531d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * java.util.concurrent.ExecutorService)}, {@link ListenableFutureTask}, 541d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * {@link AbstractFuture}, and other utilities over creating plain {@code 551d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * Future} instances to be upgraded to {@code ListenableFuture} after the 561d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * fact. 571d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert */ 581d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert public static <V> ListenableFuture<V> listenInPoolThread( 591d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert Future<V> future) { 601d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert if (future instanceof ListenableFuture<?>) { 611d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert return (ListenableFuture<V>) future; 621d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert } 631d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert return new ListenableFutureAdapter<V>(future); 641d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert } 651d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert 661d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert @VisibleForTesting 671d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert static <V> ListenableFuture<V> listenInPoolThread( 681d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert Future<V> future, Executor executor) { 691d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert checkNotNull(executor); 701d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert if (future instanceof ListenableFuture<?>) { 711d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert return (ListenableFuture<V>) future; 721d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert } 731d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert return new ListenableFutureAdapter<V>(future, executor); 741d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert } 751d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert 761d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert /** 771d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * An adapter to turn a {@link Future} into a {@link ListenableFuture}. This 781d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * will wait on the future to finish, and when it completes, run the 791d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * listeners. This implementation will wait on the source future 801d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * indefinitely, so if the source future never completes, the adapter will 811d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * never complete either. 821d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * 831d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * <p>If the delegate future is interrupted or throws an unexpected unchecked 841d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert * exception, the listeners will not be invoked. 851d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert */ 861d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert private static class ListenableFutureAdapter<V> extends ForwardingFuture<V> 871d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert implements ListenableFuture<V> { 881d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert 891d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert private static final ThreadFactory threadFactory = 901d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert new ThreadFactoryBuilder() 911d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert .setDaemon(true) 921d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert .setNameFormat("ListenableFutureAdapter-thread-%d") 931d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert .build(); 941d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert private static final Executor defaultAdapterExecutor = 951d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert Executors.newCachedThreadPool(threadFactory); 961d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert 971d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert private final Executor adapterExecutor; 981d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert 991d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert // The execution list to hold our listeners. 1001d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert private final ExecutionList executionList = new ExecutionList(); 1011d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert 1021d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert // This allows us to only start up a thread waiting on the delegate future 1031d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert // when the first listener is added. 1041d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert private final AtomicBoolean hasListeners = new AtomicBoolean(false); 1051d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert 1061d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert // The delegate future. 1071d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert private final Future<V> delegate; 1081d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert 1091d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert ListenableFutureAdapter(Future<V> delegate) { 1101d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert this(delegate, defaultAdapterExecutor); 1111d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert } 1121d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert 1131d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert ListenableFutureAdapter(Future<V> delegate, Executor adapterExecutor) { 1141d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert this.delegate = checkNotNull(delegate); 1151d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert this.adapterExecutor = checkNotNull(adapterExecutor); 1161d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert } 1171d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert 1181d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert @Override 1191d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert protected Future<V> delegate() { 1201d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert return delegate; 1211d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert } 1221d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert 1231d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert @Override 1241d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert public void addListener(Runnable listener, Executor exec) { 1251d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert executionList.add(listener, exec); 1261d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert 1271d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert // When a listener is first added, we run a task that will wait for 1281d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert // the delegate to finish, and when it is done will run the listeners. 1291d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert if (hasListeners.compareAndSet(false, true)) { 1301d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert if (delegate.isDone()) { 1311d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert // If the delegate is already done, run the execution list 1321d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert // immediately on the current thread. 1331d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert executionList.execute(); 1341d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert return; 1351d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert } 1361d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert 1371d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert adapterExecutor.execute(new Runnable() { 1381d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert @Override 1391d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert public void run() { 1401d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert try { 1411d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert delegate.get(); 1421d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert } catch (Error e) { 1431d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert throw e; 1441d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert } catch (InterruptedException e) { 1451d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert Thread.currentThread().interrupt(); 1461d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert // Threads from our private pool are never interrupted. 1471d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert throw new AssertionError(e); 1481d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert } catch (Throwable e) { 1491d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert // ExecutionException / CancellationException / RuntimeException 1501d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert // The task is done, run the listeners. 1511d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert } 1521d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert executionList.execute(); 1531d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert } 1541d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert }); 1551d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert } 1561d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert } 1571d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert } 1581d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert 1591d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert private JdkFutureAdapters() {} 1601d580d0f6ee4f21eb309ba7b509d2c6d671c4044Bjorn Bringert} 161