bayou.async

Interface Async<T>

All Known Subinterfaces:

AutoAsync<T>, Result<T>

All Known Implementing Classes:

AsyncBundle, HttpResponseImpl, Promise, SimpleHttpResponse, WebSocketResponse.Accept, WebSocketResponse.Reject

public interface Async<T>

Represents an async action that eventually completes with a Result<T>.

The completion status and result can be polled by `pollResult()` method; or callbacks can be registered through `onCompletion()` method.

Sequential Actions

After an action completes, often we want to start another action, and so on, forming a sequence of actions. This is modeled through sequencing methods, which include transform(), map(), peek(), then(), catch_(), and finally_().

Every sequencing method takes in a subsequent action, and returns the action sequence (also an Async) of the two actions. After `this` action completes, the subsequent action is started; when the subsequent action completes, the action sequence completes with the same result.

The following is an example of a sequence of 5 actions:

     Async<Integer> action1(){...}
     Async<String>  action2(Integer x){...}

     Async<Integer> sequence =
         action1()
         .map ( x->x*2 )
         .then( x->action2(x) )
         .catch_( IOException.class, e->"default string" )
         .finally_( resource::close );
 

Cancelling a sequence is equivalent to cancelling every action in the sequence (after it's started).

Covariance

Conceptually, Async<T> is covariant, for example, Async<String> "is" an Async<CharSequence>, which "is" an Async<Object>.

Even if we use wildcards profusely, sometimes we'll run into situations where explicit cast is necessary. We provide covary() to do covariance cast, so that, if necessary, an Async<String> can be used as an Async<CharSequence>.

See Also

• Promise -- a mutable implementation of Async for result producers
• AsyncIterator -- for loops of async actions
• AsyncBundle -- to bundle parallel async actions
• Fiber -- async analogue of Thread

Field Summary

Fields

Modifier and Type	Field and Description
static Async<Void>	VOID A constant Async<Void> that immediately succeeds with value null.

Method Summary

Abstract Methods
Result<T>	pollResult() Return the result of the action if it is completed; otherwise return null.
void	onCompletion(Consumer<Result<T>> callback) Register a callback that will be invoked after the action is completed.
void	cancel(Exception reason) Send a cancellation request to this async action.
Default Methods
boolean	isCompleted() Whether the action is completed.
T	sync() Block the current thread util this async action completes; return the result or throw the exception.
Async<T>	timeout(Duration duration) Fail this action with TimeoutException if it's not completed within the specific duration.
<R> Async<R>	covary() Do a covariance cast.
<R> Async<R>	transform(FunctionX<Result<T>,Async<R>> func) After `this` action completes, invoke a function to transform the result.
<R> Async<R>	map(FunctionX<T,R> onSuccess) After `this` action succeeds, map the value to another value.
<R> Async<R>	map(FunctionX<T,R> onSuccess, FunctionX<Exception,R> onFailure) After `this` action completes, map the success value or the failure exception to another value.
Async<T>	peek(ConsumerX<T> onSuccess) After `this` action succeeds, perform an action on the value.
Async<T>	peek(ConsumerX<T> onSuccess, ConsumerX<Exception> onFailure) After `this` action completes, perform an action on the result.
<R> Async<R>	then(FunctionX<T,Async<R>> onSuccess) After `this` action succeeds, invoke a subsequent action.
<R> Async<R>	then(FunctionX<T,Async<R>> onSuccess, FunctionX<Exception,Async<R>> onFailure) After `this` action completes, invoke a subsequence action.
<X extends Exception> Async<T>	catch_(Class<X> exceptionType, FunctionX<X,T> exceptionHandler) If `this` action fails with a specific exception type, invoke an exception handler.
<X extends Exception> Async<T>	catch__(Class<X> exceptionType, FunctionX<X,Async<T>> exceptionHandler) If `this` action fails with a specific exception type, invoke an exception handler.
Async<T>	finally_(RunnableX action) After `this` action completes, regardless of success or failure, execute a subsequent action.
Async<T>	finally_(Callable<Async<?>> action) After `this` action completes, regardless of success or failure, execute a subsequent action.
Static Methods
<T> Async<T>	success(T value) Create an `Async<T>` that immediately succeeds with `value`.
<T> Async<T>	failure(Exception e) Create an `Async<T>` that immediately fails with `e`.
<T> Async<T>	execute(ExecutorService executor, Callable<T> action) Execute the action asynchronously, return an Async that completes when the action completes.
<T> Async<T>	execute(Callable<T> action) Execute the action asynchronously in a default executor.
<T1,T2,R> Async<R>	invoke(BiFunctionX<T1,T2,R> func, Async<T1> async1, Async<T2> async2) Invoke `func` with async args.
<T1,T2,R> Async<R>	invoke_(BiFunctionX<T1,T2,Async<R>> func, Async<T1> async1, Async<T2> async2) Similar to invoke(bayou.util.function.BiFunctionX, Async, Async), except that `func` returns `Async<R>` instead.

Field Detail

VOID

static final Async<Void> VOID

A constant Async<Void> that immediately succeeds with value null. This object is equivalent to the return value of `Async.success( (Void)null )`.

Example Usage:

     Async<Void> action()
     {
         if(condition)
             return Async.VOID;
         ...
     }
 

Method Detail

pollResult

Result<T> pollResult()

Return the result of the action if it is completed; otherwise return null.

isCompleted

default boolean isCompleted()

Whether the action is completed.

This method is equivalent to `pollResult()!= null`

sync

default T sync()
        throws Exception

Block the current thread util this async action completes; return the result or throw the exception.

If the current thread is interrupted while it's being blocked by this method, this async action will receive a cancellation request with `InterruptedException` as reason.

This method does not have a timeout parameter; if timeout is needed, consider `action.timeout(duration).sync()`.

Caution: This is a blocking method, which usually should not be called in an async application. Because this method blocks the current thread, deadlock is possible if the current thread is needed for completion of this async action. Be very careful if you use this method on a production system.

Throws:

Exception

onCompletion

void onCompletion(Consumer<Result<T>> callback)

Register a callback that will be invoked after the action is completed. The result of the action will be fed to the callback.

Multiple callbacks can be registered to an async action; there is no guaranteed in which order they will be invoked.

The `callback` will be invoked in the current executor.

cancel

void cancel(Exception reason)

Send a cancellation request to this async action.

If the action has not been completed, it should be aborted as soon as possible, and the result should be a failure with `reason` as the failure exception. Multiple cancellation requests can be sent; only the first one is effective.

If the action has been completed, the cancellation request is ignored.

The `reason` should usually be a checked exception. If there is no particularly reason, consider `new Exception("cancelled")`.

timeout

default Async<T> timeout(Duration duration)

Fail this action with TimeoutException if it's not completed within the specific duration.

The default implementation calls cancel(TimeoutException) after `duration`, unless this action is completed before that.

Example usage:

     action().timeout(Duration.ofSeconds(2)).then(...);
 

Multiple timeouts on the same action is ok, the effect is that of the shortest timeout. Some actions have default timeout, caller can then use timeout() to set a shorter timeout.

This method returns `this`, or an equivalent Async (that behaves the same as this one).

The timeout is relative to when `timeout()` is called, not when the action is started.

Returns:

`this` or an equivalent Async<T>, for method chaining.

covary

default <R> Async<R> covary()

Do a covariance cast.

For example

     Async<String> asyncString = ...;

     Async<CharSequence> asyncChars = asyncString.covary();

     asyncString
         .<CharSequence>covary()
         .catch_(Exception.class, e->someCharSequence)

 

R must be a supertype of T. Unfortunately, Java doe not support <R super T>, so users of this method must take care to uphold that constraint.

transform

default <R> Async<R> transform(FunctionX<Result<T>,Async<R>> func)

After `this` action completes, invoke a function to transform the result.

This is a sequencing method. After `this` action completes with result `r` (success or failure), invoke `func.apply(r)`

• if it returns `Async<R> a2`, `a2` is the subsequence action.
• if it throws `Exception e`, the sequence fails with `e`.

The `func` will be invoked in the current executor.

map

default <R> Async<R> map(FunctionX<T,R> onSuccess)

After `this` action succeeds, map the value to another value.

This is a sequencing method.

• If `this` action succeeds with value `t`, invoke `onSuccess.apply(t)` --
  • if it returns `r`, the sequence succeeds with `r`
  • if it throws `e`, the sequence fails with `e`
• If `this` action fails with `e`, the sequence fails with `e`

Invoking `map(onSuccess)` is equivalent to `map(onSuccess, onFailure)` with `onFailure = e->{throw e;}`.

(If you need "flat-map", see then() methods.)

The `onSuccess` will be invoked in the current executor.

map

default <R> Async<R> map(FunctionX<T,R> onSuccess,
                         FunctionX<Exception,R> onFailure)

After `this` action completes, map the success value or the failure exception to another value.

This is a sequencing method. After `this` action succeeds with `v` or fails with `e`, invoke `onSuccess.apply(v)` or `onFailure.apply(e)` respectively,

• if it returns `r`, the sequence succeeds with `r`
• if it throws `e`, the sequence fails with `e`

If you only care about the failure case, see catch_() method.

(If you need "flat-map", see then() methods.)

The `onSuccess` or 'onFailure' will be invoked in the current executor.

peek

default Async<T> peek(ConsumerX<T> onSuccess)

After `this` action succeeds, perform an action on the value.

This is a sequencing method equivalent to

     map( t->{ onSuccess.accept(t); return t; } );
 

useful for performing an action without changing the result.

Example usage:

     Async
         .success("Hello")
         .map( String::toUpperCase )
         .peek( System.out::println )
         ...
 

peek

default Async<T> peek(ConsumerX<T> onSuccess,
                      ConsumerX<Exception> onFailure)

After `this` action completes, perform an action on the result.

This is a sequencing method equivalent to

     map(
         t->{ onSuccess.accept(t); return t; },
         e->{ onFailure.accept(e); throw  e; }
     );
 

useful for performing an action without changing the result.

Example usage:

         ...
         .peek( System.out::println, Exception::printStackTrace )
         ...
 

then

default <R> Async<R> then(FunctionX<T,Async<R>> onSuccess)

After `this` action succeeds, invoke a subsequent action.

This is a sequencing method.

• If `this` action succeeds with value `t`, invoke `onSuccess.apply(t)` --
  • if it returns `a2`, `a2` is the subsequence action.
  • if it throws `e`, the sequence fails with `e`
• If `this` action fails with `e`, the sequence fails with `e`

Invoking `then(onSuccess)` is equivalent to `then(onSuccess, onFailure)` with `onFailure = e->{throw e;}`.

(This method is also known as "flat-map".)

The `onSuccess` will be invoked in the current executor.

then

default <R> Async<R> then(FunctionX<T,Async<R>> onSuccess,
                          FunctionX<Exception,Async<R>> onFailure)

After `this` action completes, invoke a subsequence action.

This is a sequencing method. After `this` action succeeds with `v` or fails with `e`, invoke `onSuccess.apply(v)` or `onFailure.apply(e)` respectively,

• if it returns `a2`, `a2` is the subsequence action.
• if it throws `e`, the sequence fails with `e`

(This method is also known as "flat-map".)

The `onSuccess` or 'onFailure' will be invoked in the current executor.

catch_

default <X extends Exception> Async<T> catch_(Class<X> exceptionType,
                                              FunctionX<X,T> exceptionHandler)

If `this` action fails with a specific exception type, invoke an exception handler.

This is a sequencing method.

• if `this` action fails with `x` of type `X`, invoke `exceptionHandler.apply(x)`
  • if it returns `t`, the sequence succeeds with `t`
  • if it throws `e`, the sequence fails with `e`
• otherwise, the sequence completes with the result of `this` action.

Example Usage:

     action.catch_(IOException.class, e->"surrogate");

     action
         .then( ... )
         .catch_(IOException.class, e->{
             log(e);
             throw new MyException(e);
         })
         .finally_( ... );
 

If you need an exceptionHandler that returns a supertype of T, consider covary() before catch_().

See also catch__(Class, FunctionX<X, Async<T>>) for async exception handlers. Note that that method has 2 underscores, while this method has 1 underscore.

The `exceptionHandler` will be invoked in the current executor.

catch__

default <X extends Exception> Async<T> catch__(Class<X> exceptionType,
                                               FunctionX<X,Async<T>> exceptionHandler)

If `this` action fails with a specific exception type, invoke an exception handler.

This is a sequencing method.

• if `this` action fails with `x` of type `X`, run the async `exceptionHandler`
  • if it succeeds with `t`, the sequence succeeds with `t`
  • if it fails with `e`, the sequence fails with `e`
• otherwise, the sequence completes with the result of `this` action.

Example Usage:

     action.catch__(IOException.class, e->Fiber.sleep(duration, "surrogate") );

     action
         .then( ... )
         .catch__(IOException.class, e->
             someAsyncAction()
                 .then( v->Async.failure(e) ) // rethrow
         )
         .finally_( ... );
 

If you need an exceptionHandler that returns Async<X> where X is a supertype of T, consider <X>covary() before catch__().

See also catch_(Class, FunctionX<X, T>). Note that that method has 1 underscore, while this method has 2 underscores.

The `exceptionHandler` will be invoked in the current executor.

finally_

default Async<T> finally_(RunnableX action)

After `this` action completes, regardless of success or failure, execute a subsequent action.

This is a sequencing method. After `this` action completes with result `r` (success or failure), invoke `action.run()`

• if it returns normally, the sequence completes with result `r`
• if it throws `e`, the sequence fails with `e`.

Example Usage:

     action
         .then(...)
         .finally_( resource::close );
 

The `action` will be invoked in the current executor.

finally_

default Async<T> finally_(Callable<Async<?>> action)

After `this` action completes, regardless of success or failure, execute a subsequent action.

This is a sequencing method. After `this` action completes with result `r` (success or failure), start the async `action`,

• if it completes successfully, the sequence completes with result `r`
• if it fails with `e`, the sequence fails with `e`.

Example Usage:

     action
         .then(...)
         .finally_( wsChannel::close );
 

The `action` will be invoked in the current executor.

success

static <T> Async<T> success(T value)

Create an `Async<T>` that immediately succeeds with `value`.

This method is equivalent to Result.success(value).

failure

static <T> Async<T> failure(Exception e)

Create an `Async<T>` that immediately fails with `e`.

This method is equivalent to Result.failure(e).

execute

static <T> Async<T> execute(ExecutorService executor,
                            Callable<T> action)

Execute the action asynchronously, return an Async that completes when the action completes.

This method is usually to convert a legacy blocking action to an async action. For example

     Async.execute( executor, ()->Files.readAllBytes(path) );
 

If the returned Async is cancelled with `reason=e` before the action is completed, the Async fails with `e`, and the thread executing the action is interrupted.

The action will inherit the current fiber, i.e. when it's executed, Fiber.current() will return the same value as the caller of this method would see.

See Callable_Void in case the `action` argument is a void-returning lambda expression or method reference.

execute

static <T> Async<T> execute(Callable<T> action)

Execute the action asynchronously in a default executor.

This method is usually used to convert a legacy blocking action to an async action. For example

     Async.execute( ()->Files.readAllBytes(path) );
 

This method is equivalent to execute(ExecutorService, Callable) with a default executor.

Caution: the action is submitted to a system default executor with unlimited threads. That is probably inappropriate in a production system.

invoke

static <T1,T2,R> Async<R> invoke(BiFunctionX<T1,T2,R> func,
                                 Async<T1> async1,
                                 Async<T2> async2)

Invoke `func` with async args.

If both `async1` and `async2` are successful, their values will be fed to `func`, and this `invoke` action completes with the result of `func`.

If either `async1` or `async2` fails, this `invoke` action fails too with the same exception; the uncompleted `async1` or `async2` will be canceled.

Cancelling this `invoke` action is equivalent to canceling both `async1` and `async2`.

See Also:

invoke_(bayou.util.function.BiFunctionX, Async, Async)

invoke_

static <T1,T2,R> Async<R> invoke_(BiFunctionX<T1,T2,Async<R>> func,
                                  Async<T1> async1,
                                  Async<T2> async2)

Similar to invoke(bayou.util.function.BiFunctionX, Async, Async), except that `func` returns `Async<R>` instead.