001/* 002 * Copyright (C) 2006 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 com.google.common.annotations.Beta; 018import com.google.common.annotations.GwtCompatible; 019import com.google.common.annotations.GwtIncompatible; 020import com.google.common.base.Function; 021import com.google.errorprone.annotations.CanIgnoreReturnValue; 022import java.util.concurrent.ExecutionException; 023import java.util.concurrent.Executor; 024import java.util.concurrent.ScheduledExecutorService; 025import java.util.concurrent.TimeUnit; 026import java.util.concurrent.TimeoutException; 027 028/** 029 * A {@link ListenableFuture} that supports fluent chains of operations. For example: 030 * 031 * <pre>{@code 032 * ListenableFuture<Boolean> adminIsLoggedIn = 033 * FluentFuture.from(usersDatabase.getAdminUser()) 034 * .transform(User::getId, directExecutor()) 035 * .transform(ActivityService::isLoggedIn, threadPool) 036 * .catching(RpcException.class, e -> false, directExecutor()); 037 * }</pre> 038 * 039 * <h3>Alternatives</h3> 040 * 041 * <h4>Frameworks</h4> 042 * 043 * <p>When chaining together a graph of asynchronous operations, you will often find it easier to 044 * use a framework. Frameworks automate the process, often adding features like monitoring, 045 * debugging, and cancellation. Examples of frameworks include: 046 * 047 * <ul> 048 * <li><a href="http://google.github.io/dagger/producers.html">Dagger Producers</a> 049 * </ul> 050 * 051 * <h4>{@link java.util.concurrent.CompletableFuture} / {@link java.util.concurrent.CompletionStage} 052 * </h4> 053 * 054 * <p>Users of {@code CompletableFuture} will likely want to continue using {@code 055 * CompletableFuture}. {@code FluentFuture} is targeted at people who use {@code ListenableFuture}, 056 * who can't use Java 8, or who want an API more focused than {@code CompletableFuture}. (If you 057 * need to adapt between {@code CompletableFuture} and {@code ListenableFuture}, consider <a 058 * href="https://github.com/lukas-krecan/future-converter">Future Converter</a>.) 059 * 060 * <h3>Extension</h3> 061 * 062 * If you want a class like {@code FluentFuture} but with extra methods, we recommend declaring your 063 * own subclass of {@link ListenableFuture}, complete with a method like {@link #from} to adapt an 064 * existing {@code ListenableFuture}, implemented atop a {@link ForwardingListenableFuture} that 065 * forwards to that future and adds the desired methods. 066 * 067 * @since 23.0 068 */ 069@Beta 070@GwtCompatible(emulated = true) 071public abstract class FluentFuture<V> extends GwtFluentFutureCatchingSpecialization<V> { 072 073 /** 074 * A less abstract subclass of AbstractFuture. This can be used to optimize setFuture by ensuring 075 * that {@link #get} calls exactly the implementation of {@link AbstractFuture#get}. 076 */ 077 abstract static class TrustedFuture<V> extends FluentFuture<V> 078 implements AbstractFuture.Trusted<V> { 079 @CanIgnoreReturnValue 080 @Override 081 public final V get() throws InterruptedException, ExecutionException { 082 return super.get(); 083 } 084 085 @CanIgnoreReturnValue 086 @Override 087 public final V get(long timeout, TimeUnit unit) 088 throws InterruptedException, ExecutionException, TimeoutException { 089 return super.get(timeout, unit); 090 } 091 092 @Override 093 public final boolean isDone() { 094 return super.isDone(); 095 } 096 097 @Override 098 public final boolean isCancelled() { 099 return super.isCancelled(); 100 } 101 102 @Override 103 public final void addListener(Runnable listener, Executor executor) { 104 super.addListener(listener, executor); 105 } 106 107 @CanIgnoreReturnValue 108 @Override 109 public final boolean cancel(boolean mayInterruptIfRunning) { 110 return super.cancel(mayInterruptIfRunning); 111 } 112 } 113 114 FluentFuture() {} 115 116 /** 117 * Converts the given {@code ListenableFuture} to an equivalent {@code FluentFuture}. 118 * 119 * <p>If the given {@code ListenableFuture} is already a {@code FluentFuture}, it is returned 120 * directly. If not, it is wrapped in a {@code FluentFuture} that delegates all calls to the 121 * original {@code ListenableFuture}. 122 */ 123 public static <V> FluentFuture<V> from(ListenableFuture<V> future) { 124 return future instanceof FluentFuture 125 ? (FluentFuture<V>) future 126 : new ForwardingFluentFuture<V>(future); 127 } 128 129 /** 130 * Returns a {@code Future} whose result is taken from this {@code Future} or, if this {@code 131 * Future} fails with the given {@code exceptionType}, from the result provided by the {@code 132 * fallback}. {@link Function#apply} is not invoked until the primary input has failed, so if the 133 * primary input succeeds, it is never invoked. If, during the invocation of {@code fallback}, an 134 * exception is thrown, this exception is used as the result of the output {@code Future}. 135 * 136 * <p>Usage example: 137 * 138 * <pre>{@code 139 * // Falling back to a zero counter in case an exception happens when processing the RPC to fetch 140 * // counters. 141 * ListenableFuture<Integer> faultTolerantFuture = 142 * fetchCounters().catching(FetchException.class, x -> 0, directExecutor()); 143 * }</pre> 144 * 145 * <p>When selecting an executor, note that {@code directExecutor} is dangerous in some cases. See 146 * the discussion in the {@link #addListener} documentation. All its warnings about heavyweight 147 * listeners are also applicable to heavyweight functions passed to this method. 148 * 149 * <p>This method is similar to {@link java.util.concurrent.CompletableFuture#exceptionally}. It 150 * can also serve some of the use cases of {@link java.util.concurrent.CompletableFuture#handle} 151 * and {@link java.util.concurrent.CompletableFuture#handleAsync} when used along with {@link 152 * #transform}. 153 * 154 * @param exceptionType the exception type that triggers use of {@code fallback}. The exception 155 * type is matched against the input's exception. "The input's exception" means the cause of 156 * the {@link ExecutionException} thrown by {@code input.get()} or, if {@code get()} throws a 157 * different kind of exception, that exception itself. To avoid hiding bugs and other 158 * unrecoverable errors, callers should prefer more specific types, avoiding {@code 159 * Throwable.class} in particular. 160 * @param fallback the {@link Function} to be called if the input fails with the expected 161 * exception type. The function's argument is the input's exception. "The input's exception" 162 * means the cause of the {@link ExecutionException} thrown by {@code this.get()} or, if 163 * {@code get()} throws a different kind of exception, that exception itself. 164 * @param executor the executor that runs {@code fallback} if the input fails 165 */ 166 @Partially.GwtIncompatible("AVAILABLE but requires exceptionType to be Throwable.class") 167 public final <X extends Throwable> FluentFuture<V> catching( 168 Class<X> exceptionType, Function<? super X, ? extends V> fallback, Executor executor) { 169 return (FluentFuture<V>) Futures.catching(this, exceptionType, fallback, executor); 170 } 171 172 /** 173 * Returns a {@code Future} whose result is taken from this {@code Future} or, if this {@code 174 * Future} fails with the given {@code exceptionType}, from the result provided by the {@code 175 * fallback}. {@link AsyncFunction#apply} is not invoked until the primary input has failed, so if 176 * the primary input succeeds, it is never invoked. If, during the invocation of {@code fallback}, 177 * an exception is thrown, this exception is used as the result of the output {@code Future}. 178 * 179 * <p>Usage examples: 180 * 181 * <pre>{@code 182 * // Falling back to a zero counter in case an exception happens when processing the RPC to fetch 183 * // counters. 184 * ListenableFuture<Integer> faultTolerantFuture = 185 * fetchCounters().catchingAsync( 186 * FetchException.class, x -> immediateFuture(0), directExecutor()); 187 * }</pre> 188 * 189 * <p>The fallback can also choose to propagate the original exception when desired: 190 * 191 * <pre>{@code 192 * // Falling back to a zero counter only in case the exception was a 193 * // TimeoutException. 194 * ListenableFuture<Integer> faultTolerantFuture = 195 * fetchCounters().catchingAsync( 196 * FetchException.class, 197 * e -> { 198 * if (omitDataOnFetchFailure) { 199 * return immediateFuture(0); 200 * } 201 * throw e; 202 * }, 203 * directExecutor()); 204 * }</pre> 205 * 206 * <p>When selecting an executor, note that {@code directExecutor} is dangerous in some cases. See 207 * the discussion in the {@link #addListener} documentation. All its warnings about heavyweight 208 * listeners are also applicable to heavyweight functions passed to this method. (Specifically, 209 * {@code directExecutor} functions should avoid heavyweight operations inside {@code 210 * AsyncFunction.apply}. Any heavyweight operations should occur in other threads responsible for 211 * completing the returned {@code Future}.) 212 * 213 * <p>This method is similar to {@link java.util.concurrent.CompletableFuture#exceptionally}. It 214 * can also serve some of the use cases of {@link java.util.concurrent.CompletableFuture#handle} 215 * and {@link java.util.concurrent.CompletableFuture#handleAsync} when used along with {@link 216 * #transform}. 217 * 218 * @param exceptionType the exception type that triggers use of {@code fallback}. The exception 219 * type is matched against the input's exception. "The input's exception" means the cause of 220 * the {@link ExecutionException} thrown by {@code this.get()} or, if {@code get()} throws a 221 * different kind of exception, that exception itself. To avoid hiding bugs and other 222 * unrecoverable errors, callers should prefer more specific types, avoiding {@code 223 * Throwable.class} in particular. 224 * @param fallback the {@link AsyncFunction} to be called if the input fails with the expected 225 * exception type. The function's argument is the input's exception. "The input's exception" 226 * means the cause of the {@link ExecutionException} thrown by {@code input.get()} or, if 227 * {@code get()} throws a different kind of exception, that exception itself. 228 * @param executor the executor that runs {@code fallback} if the input fails 229 */ 230 @Partially.GwtIncompatible("AVAILABLE but requires exceptionType to be Throwable.class") 231 public final <X extends Throwable> FluentFuture<V> catchingAsync( 232 Class<X> exceptionType, AsyncFunction<? super X, ? extends V> fallback, Executor executor) { 233 return (FluentFuture<V>) Futures.catchingAsync(this, exceptionType, fallback, executor); 234 } 235 236 /** 237 * Returns a future that delegates to this future but will finish early (via a {@link 238 * TimeoutException} wrapped in an {@link ExecutionException}) if the specified timeout expires. 239 * If the timeout expires, not only will the output future finish, but also the input future 240 * ({@code this}) will be cancelled and interrupted. 241 * 242 * @param timeout when to time out the future 243 * @param unit the time unit of the time parameter 244 * @param scheduledExecutor The executor service to enforce the timeout. 245 */ 246 @GwtIncompatible // ScheduledExecutorService 247 @SuppressWarnings("GoodTime") // should accept a java.time.Duration 248 public final FluentFuture<V> withTimeout( 249 long timeout, TimeUnit unit, ScheduledExecutorService scheduledExecutor) { 250 return (FluentFuture<V>) Futures.withTimeout(this, timeout, unit, scheduledExecutor); 251 } 252 253 /** 254 * Returns a new {@code Future} whose result is asynchronously derived from the result of this 255 * {@code Future}. If the input {@code Future} fails, the returned {@code Future} fails with the 256 * same exception (and the function is not invoked). 257 * 258 * <p>More precisely, the returned {@code Future} takes its result from a {@code Future} produced 259 * by applying the given {@code AsyncFunction} to the result of the original {@code Future}. 260 * Example usage: 261 * 262 * <pre>{@code 263 * FluentFuture<RowKey> rowKeyFuture = FluentFuture.from(indexService.lookUp(query)); 264 * ListenableFuture<QueryResult> queryFuture = 265 * rowKeyFuture.transformAsync(dataService::readFuture, executor); 266 * }</pre> 267 * 268 * <p>When selecting an executor, note that {@code directExecutor} is dangerous in some cases. See 269 * the discussion in the {@link #addListener} documentation. All its warnings about heavyweight 270 * listeners are also applicable to heavyweight functions passed to this method. (Specifically, 271 * {@code directExecutor} functions should avoid heavyweight operations inside {@code 272 * AsyncFunction.apply}. Any heavyweight operations should occur in other threads responsible for 273 * completing the returned {@code Future}.) 274 * 275 * <p>The returned {@code Future} attempts to keep its cancellation state in sync with that of the 276 * input future and that of the future returned by the chain function. That is, if the returned 277 * {@code Future} is cancelled, it will attempt to cancel the other two, and if either of the 278 * other two is cancelled, the returned {@code Future} will receive a callback in which it will 279 * attempt to cancel itself. 280 * 281 * <p>This method is similar to {@link java.util.concurrent.CompletableFuture#thenCompose} and 282 * {@link java.util.concurrent.CompletableFuture#thenComposeAsync}. It can also serve some of the 283 * use cases of {@link java.util.concurrent.CompletableFuture#handle} and {@link 284 * java.util.concurrent.CompletableFuture#handleAsync} when used along with {@link #catching}. 285 * 286 * @param function A function to transform the result of this future to the result of the output 287 * future 288 * @param executor Executor to run the function in. 289 * @return A future that holds result of the function (if the input succeeded) or the original 290 * input's failure (if not) 291 */ 292 public final <T> FluentFuture<T> transformAsync( 293 AsyncFunction<? super V, T> function, Executor executor) { 294 return (FluentFuture<T>) Futures.transformAsync(this, function, executor); 295 } 296 297 /** 298 * Returns a new {@code Future} whose result is derived from the result of this {@code Future}. If 299 * this input {@code Future} fails, the returned {@code Future} fails with the same exception (and 300 * the function is not invoked). Example usage: 301 * 302 * <pre>{@code 303 * ListenableFuture<List<Row>> rowsFuture = 304 * queryFuture.transform(QueryResult::getRows, executor); 305 * }</pre> 306 * 307 * <p>When selecting an executor, note that {@code directExecutor} is dangerous in some cases. See 308 * the discussion in the {@link #addListener} documentation. All its warnings about heavyweight 309 * listeners are also applicable to heavyweight functions passed to this method. 310 * 311 * <p>The returned {@code Future} attempts to keep its cancellation state in sync with that of the 312 * input future. That is, if the returned {@code Future} is cancelled, it will attempt to cancel 313 * the input, and if the input is cancelled, the returned {@code Future} will receive a callback 314 * in which it will attempt to cancel itself. 315 * 316 * <p>An example use of this method is to convert a serializable object returned from an RPC into 317 * a POJO. 318 * 319 * <p>This method is similar to {@link java.util.concurrent.CompletableFuture#thenApply} and 320 * {@link java.util.concurrent.CompletableFuture#thenApplyAsync}. It can also serve some of the 321 * use cases of {@link java.util.concurrent.CompletableFuture#handle} and {@link 322 * java.util.concurrent.CompletableFuture#handleAsync} when used along with {@link #catching}. 323 * 324 * @param function A Function to transform the results of this future to the results of the 325 * returned future. 326 * @param executor Executor to run the function in. 327 * @return A future that holds result of the transformation. 328 */ 329 public final <T> FluentFuture<T> transform(Function<? super V, T> function, Executor executor) { 330 return (FluentFuture<T>) Futures.transform(this, function, executor); 331 } 332 333 /** 334 * Registers separate success and failure callbacks to be run when this {@code Future}'s 335 * computation is {@linkplain java.util.concurrent.Future#isDone() complete} or, if the 336 * computation is already complete, immediately. 337 * 338 * <p>The callback is run on {@code executor}. There is no guaranteed ordering of execution of 339 * callbacks, but any callback added through this method is guaranteed to be called once the 340 * computation is complete. 341 * 342 * <p>Example: 343 * 344 * <pre>{@code 345 * future.addCallback( 346 * new FutureCallback<QueryResult>() { 347 * public void onSuccess(QueryResult result) { 348 * storeInCache(result); 349 * } 350 * public void onFailure(Throwable t) { 351 * reportError(t); 352 * } 353 * }, executor); 354 * }</pre> 355 * 356 * <p>When selecting an executor, note that {@code directExecutor} is dangerous in some cases. See 357 * the discussion in the {@link #addListener} documentation. All its warnings about heavyweight 358 * listeners are also applicable to heavyweight callbacks passed to this method. 359 * 360 * <p>For a more general interface to attach a completion listener, see {@link #addListener}. 361 * 362 * <p>This method is similar to {@link java.util.concurrent.CompletableFuture#whenComplete} and 363 * {@link java.util.concurrent.CompletableFuture#whenCompleteAsync}. It also serves the use case 364 * of {@link java.util.concurrent.CompletableFuture#thenAccept} and {@link 365 * java.util.concurrent.CompletableFuture#thenAcceptAsync}. 366 * 367 * @param callback The callback to invoke when this {@code Future} is completed. 368 * @param executor The executor to run {@code callback} when the future completes. 369 */ 370 public final void addCallback(FutureCallback<? super V> callback, Executor executor) { 371 Futures.addCallback(this, callback, executor); 372 } 373}