bayou.async

Class AsyncBundle<T,R>

java.lang.Object

bayou.async.AsyncBundle<T,R>

Type Parameters:

T - value type of child tasks

R - value type of the bundle

All Implemented Interfaces:

Async<R>

public class AsyncBundle<T,R>
extends Object
implements Async<R>

Bundles multiple parallel Async<T> tasks as one Async<R> action.

An AsyncBundle<T,R> consists of multiple child tasks of type Async<T> ; the bundle itself is an Async<R> , which completes when all or some of the tasks are completed (determined by the trigger function).

For common usages, see static methods anyOf(), allOf(), someOf(). For example

     Stream<Async<String>> tasks = ...;
     Async<List<String>> bundle = AsyncBundle.someOf(tasks, 2);
     // the bundle succeeds as soon as 2 of the tasks succeeds
 

Trigger Function

The trigger function determines when and how the bundle completes. The input to the function is the results from currently completed tasks; the output of the function can be one of:

• return null
  the bundle is not yet completed; awaiting more task completions
• return `r` (non-null, type R)
  the bundle completes with success value `r`
• throw `e`
  the bundle completes with failure exception `e`

The bundle first invokes the trigger function with an empty list, then with a list of one result after the first task completion, and so on, until the bundle completes.

Cancellation

When a bundle completes, the remaining uncompleted tasks are cancelled; they are no longer needed by the bundle.

Constructor Summary

Constructors

Constructor and Description
AsyncBundle(Stream<Async<T>> tasks, FunctionX<List<Result<T>>,R> trigger) Create a bundle over the tasks, with the trigger function.

Method Summary

Instance Methods
List<Result<T>>	pollTaskResults() Return an immutable list of results from currently completed tasks.
Result<R>	pollResult() Implements Async<R>.pollResult().
void	onCompletion(Consumer<Result<R>> callback) Implements Async<R>.onCompletion().
void	cancel(Exception reason) Implements Async<R>.cancel().
Static Methods
<T> AsyncBundle<T,T>	anyOf(Stream<Async<T>> tasks) Create a bundle that succeeds as soon as any one of the tasks succeeds.
<T> AsyncBundle<T,List<T>>	allOf(Stream<Async<T>> tasks) Create a bundle that succeeds if all of the tasks succeeds.
<T> AsyncBundle<T,List<T>>	someOf(Stream<Async<T>> tasks, int successThreshold) Create a bundle that succeeds as soon as a number of the tasks succeeds.
<T> FunctionX<List<Result<T>>,List<T>>	triggerOf(int successThreshold, int taskCount) Return a simple trigger function that's based the number of success results.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface bayou.async.Async

catch__, catch_, covary, finally_, finally_, isCompleted, map, map, peek, peek, sync, then, then, timeout, transform

Constructor Detail

AsyncBundle

public AsyncBundle(Stream<Async<T>> tasks,
                   FunctionX<List<Result<T>>,R> trigger)

Create a bundle over the tasks, with the trigger function.

Method Detail

pollTaskResults

public List<Result<T>> pollTaskResults()

Return an immutable list of results from currently completed tasks.

pollResult

public Result<R> pollResult()

Implements Async<R>.pollResult().

Specified by:

pollResult in interface Async<R>

onCompletion

public void onCompletion(Consumer<Result<R>> callback)

Implements Async<R>.onCompletion().

Specified by:

onCompletion in interface Async<R>

cancel

public void cancel(Exception reason)

Implements Async<R>.cancel().

The cancel request will be forwarded to all uncompleted tasks.

Specified by:

cancel in interface Async<R>

anyOf

public static <T> AsyncBundle<T,T> anyOf(Stream<Async<T>> tasks)

Create a bundle that succeeds as soon as any one of the tasks succeeds.

The bundle fails if none of the tasks succeeds.

This method is similar to someOf() with successThreshold=1.

allOf

public static <T> AsyncBundle<T,List<T>> allOf(Stream<Async<T>> tasks)

Create a bundle that succeeds if all of the tasks succeeds.

The bundle fails as soon as one of the task fails.

This method is similar to someOf() with successThreshold=number_of_tasks.

someOf

public static <T> AsyncBundle<T,List<T>> someOf(Stream<Async<T>> tasks,
                                                int successThreshold)

Create a bundle that succeeds as soon as a number of the tasks succeeds.

When the number of tasks that have succeeded reaches successThreshold , the bundle succeeds, with the list of values from these tasks.

The bundle fails if successThreshold becomes unreachable (because too many tasks have failed). See triggerOf(successThreshold, taskCount) for details.

triggerOf

public static <T> FunctionX<List<Result<T>>,List<T>> triggerOf(int successThreshold,
                                                               int taskCount)

Return a simple trigger function that's based the number of success results.

Define failureMax=taskCount-successThreshold . Define successCount/failureCount as number of success/failure results. The trigger function behaves as the following

• if(successCount >= successThreshold)
  Return the list of values from success results. The bundle succeeds.
• else if(failureCount > failureMax)
  Throw an OverLimitException (limitName="failureMax"). The bundle fails.
• else
  Return null. The bundle is not yet completed.

Note that successThreshold is impossible to reach if successThreshold>taskCount . In that case the trigger function fails on any input because failureCount>failureMax is always true. The bundle therefore fails immediately on construction.

Parameters:

taskCount - total number of tasks in the bundle