groovyx.gpars.dataflow
Class Dataflow

java.lang.Object
  groovyx.gpars.dataflow.Dataflow

public abstract class Dataflow
extends java.lang.Object

Contains factory methods to create dataflow actors and starting them.

Author:

Vaclav Pech, Dierk Koenig Date: Jun 4, 2009

Field Summary

static java.lang.ThreadLocal<PGroup>	activeParallelGroup Maps threads/tasks to parallel groups they belong to
static PGroup	DATA_FLOW_GROUP The parallel group used by all Dataflow Concurrency actors by default.

Constructor Summary

Dataflow()

Method Summary

static DataflowProcessor	operator(DataflowReadChannel input, DataflowWriteChannel output, groovy.lang.Closure code) Creates an operator using the current parallel group
static DataflowProcessor	operator(DataflowReadChannel input, DataflowWriteChannel output, int maxForks, groovy.lang.Closure code) Creates an operator using the current parallel group
static DataflowProcessor	operator(java.util.List inputChannels, java.util.List outputChannels, groovy.lang.Closure code) Creates an operator using the current parallel group
static DataflowProcessor	operator(java.util.List inputChannels, java.util.List outputChannels, int maxForks, groovy.lang.Closure code) Creates an operator using the current parallel group
static DataflowProcessor	operator(java.util.Map channels, groovy.lang.Closure code) Creates an operator using the default dataflow parallel group
static DataflowProcessor	prioritySelector(java.util.List inputChannels, java.util.List outputChannels) Creates a prioritizing selector using the default dataflow parallel group.
static DataflowProcessor	prioritySelector(java.util.List inputChannels, java.util.List outputChannels, groovy.lang.Closure code) Creates a prioritizing selector using the default dataflow parallel group Input with lower position index have higher priority.
static DataflowProcessor	prioritySelector(java.util.Map channels) Creates a prioritizing selector using the default dataflow parallel group.
static DataflowProcessor	prioritySelector(java.util.Map channels, groovy.lang.Closure code) Creates a prioritizing selector using the default dataflow parallel group Input with lower position index have higher priority.
static PGroup	retrieveCurrentDFPGroup() Retrieves the thread-local value of the active PGroup or the default DataflowGroup
static Select<?>	select(DataflowReadChannel<?>... channels) Creates a select using the default dataflow parallel group.
static Select<?>	select(java.util.List<DataflowReadChannel> channels) Creates a select using the default dataflow parallel group.
static DataflowProcessor	selector(java.util.List inputChannels, java.util.List outputChannels) Creates a selector using the default dataflow parallel group.
static DataflowProcessor	selector(java.util.List inputChannels, java.util.List outputChannels, groovy.lang.Closure code) Creates a selector using the default dataflow parallel group
static DataflowProcessor	selector(java.util.Map channels) Creates a selector using the default dataflow parallel group.
static DataflowProcessor	selector(java.util.Map channels, groovy.lang.Closure code) Creates a selector using the default dataflow parallel group
static DataflowProcessor	splitter(DataflowReadChannel inputChannel, java.util.List<DataflowWriteChannel> outputChannels) Creates a splitter copying its single input channel into all of its output channels.
static DataflowProcessor	splitter(DataflowReadChannel inputChannel, java.util.List<DataflowWriteChannel> outputChannels, int maxForks) Creates a splitter copying its single input channel into all of its output channels.
static <T> Promise<T>	task(java.util.concurrent.Callable<T> callable) Creates a new task assigned to a thread from the current parallel group.
static <T> Promise<T>	task(groovy.lang.Closure<T> code) Creates a new task assigned to a thread from the default dataflow parallel group.
static Promise<java.lang.Object>	task(java.lang.Runnable runnable) Creates a new task assigned to a thread from the current parallel group.
static java.lang.Object	usingGroup(PGroup group, groovy.lang.Closure code) Sets the supplied PGroup as the default for the given block of code.
static <T> Promise<T>	whenAllBound(java.util.List<Promise> promises, groovy.lang.Closure<T> code) Without blocking the thread waits for all the promises to get bound and then passes them to the supplied closure.
static <T> Promise<T>	whenAllBound(java.util.List<Promise> promises, groovy.lang.Closure<T> code, groovy.lang.Closure<T> errorHandler) Without blocking the thread waits for all the promises to get bound and then passes them to the supplied closure.
static <T> Promise<T>	whenAllBound(Promise promise1, groovy.lang.Closure<T> code) Without blocking the thread waits for all the promises to get bound and then passes them to the supplied closure.
static <T> Promise<T>	whenAllBound(Promise promise1, groovy.lang.Closure<T> code, groovy.lang.Closure<T> errorHandler) Without blocking the thread waits for all the promises to get bound and then passes them to the supplied closure.
static <T> Promise<T>	whenAllBound(Promise promise1, Promise promise2, groovy.lang.Closure<T> code) Without blocking the thread waits for all the promises to get bound and then passes them to the supplied closure.
static <T> Promise<T>	whenAllBound(Promise promise1, Promise promise2, groovy.lang.Closure<T> code, groovy.lang.Closure<T> errorHandler) Without blocking the thread waits for all the promises to get bound and then passes them to the supplied closure.
static <T> Promise<T>	whenAllBound(Promise promise1, Promise promise2, Promise promise3, groovy.lang.Closure<T> code) Without blocking the thread waits for all the promises to get bound and then passes them to the supplied closure.
static <T> Promise<T>	whenAllBound(Promise promise1, Promise promise2, Promise promise3, groovy.lang.Closure<T> code, groovy.lang.Closure<T> errorHandler) Without blocking the thread waits for all the promises to get bound and then passes them to the supplied closure.
static <T> Promise<T>	whenAllBound(Promise promise1, Promise promise2, Promise promise3, Promise promise4, groovy.lang.Closure<T> code) Without blocking the thread waits for all the promises to get bound and then passes them to the supplied closure.
static <T> Promise<T>	whenAllBound(Promise promise1, Promise promise2, Promise promise3, Promise promise4, groovy.lang.Closure<T> code, groovy.lang.Closure<T> errorHandler) Without blocking the thread waits for all the promises to get bound and then passes them to the supplied closure.

Methods inherited from class java.lang.Object

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

Field Detail

DATA_FLOW_GROUP

public static final PGroup DATA_FLOW_GROUP

The parallel group used by all Dataflow Concurrency actors by default.

activeParallelGroup

public static final java.lang.ThreadLocal<PGroup> activeParallelGroup

Maps threads/tasks to parallel groups they belong to

Constructor Detail

Dataflow

public Dataflow()

Method Detail

retrieveCurrentDFPGroup

public static PGroup retrieveCurrentDFPGroup()

Retrieves the thread-local value of the active PGroup or the default DataflowGroup

Returns:

The PGroup to use for DF within the current thread

usingGroup

public static java.lang.Object usingGroup(PGroup group,
                                          groovy.lang.Closure code)

Sets the supplied PGroup as the default for the given block of code. All dataflow functions, such as task or operator and callback handlers, will use the PGroup and its thread pool for their scheduling.

Parameters:

group - The group to make the default inside the block

code - The code to run with overriden default

Returns:

The value returned from the supplied code block

task

public static <T> Promise<T> task(groovy.lang.Closure<T> code)

Creates a new task assigned to a thread from the default dataflow parallel group. Tasks are a lightweight version of dataflow operators, which do not define their communication channels explicitly, but can only exchange data using explicit DataflowVariables and Streams.

Parameters:

code - The task body to run

Returns:

A DataflowVariable, which gets assigned the value returned from the supplied code

task

public static <T> Promise<T> task(java.util.concurrent.Callable<T> callable)

Creates a new task assigned to a thread from the current parallel group. Tasks are a lightweight version of dataflow operators, which do not define their communication channels explicitly, but can only exchange data using explicit DataflowVariables and Streams. Registers itself with Dataflow for nested 'whenBound' handlers to use the same group.

Parameters:

callable - The task body to run

Returns:

A DataflowVariable, which gets assigned the value returned from the supplied code

task

public static Promise<java.lang.Object> task(java.lang.Runnable runnable)

Creates a new task assigned to a thread from the current parallel group. Tasks are a lightweight version of dataflow operators, which do not define their communication channels explicitly, but can only exchange data using explicit DataflowVariables and Streams. Registers itself with Dataflow for nested 'whenBound' handlers to use the same group.

Parameters:

runnable - The task body to run

Returns:

A DataflowVariable, which gets bound to null once the supplied code finishes

operator

public static DataflowProcessor operator(java.util.Map channels,
                                         groovy.lang.Closure code)

Creates an operator using the default dataflow parallel group

Parameters:

channels - A map specifying "inputs" and "outputs" - dataflow channels (instances of the DataflowQueue or DataflowVariable classes) to use for inputs and outputs

code - The operator's body to run each time all inputs have a value to read

Returns:

A new active operator instance

operator

public static DataflowProcessor operator(java.util.List inputChannels,
                                         java.util.List outputChannels,
                                         groovy.lang.Closure code)

Creates an operator using the current parallel group

Parameters:

inputChannels - dataflow channels to use for input

outputChannels - dataflow channels to use for output

code - The operator's body to run each time all inputs have a value to read

Returns:

A new active operator instance

operator

public static DataflowProcessor operator(java.util.List inputChannels,
                                         java.util.List outputChannels,
                                         int maxForks,
                                         groovy.lang.Closure code)

Creates an operator using the current parallel group

Parameters:

inputChannels - dataflow channels to use for input

outputChannels - dataflow channels to use for output

maxForks - Number of parallel threads running operator's body, defaults to 1

code - The operator's body to run each time all inputs have a value to read

Returns:

A new active operator instance

operator

public static DataflowProcessor operator(DataflowReadChannel input,
                                         DataflowWriteChannel output,
                                         groovy.lang.Closure code)

Creates an operator using the current parallel group

Parameters:

input - a dataflow channel to use for input

output - a dataflow channel to use for output

code - The operator's body to run each time all inputs have a value to read

Returns:

A new active operator instance

operator

public static DataflowProcessor operator(DataflowReadChannel input,
                                         DataflowWriteChannel output,
                                         int maxForks,
                                         groovy.lang.Closure code)

Creates an operator using the current parallel group

Parameters:

input - a dataflow channel to use for input

output - a dataflow channel to use for output

maxForks - Number of parallel threads running operator's body, defaults to 1

code - The operator's body to run each time all inputs have a value to read

Returns:

A new active operator instance

selector

public static DataflowProcessor selector(java.util.Map channels,
                                         groovy.lang.Closure code)

Creates a selector using the default dataflow parallel group

Parameters:

channels - A map specifying "inputs" and "outputs" - dataflow channels (instances of the DataflowQueue or DataflowVariable classes) to use for inputs and outputs

code - The selector's body to run each time a value is available in any of the inputs channels

Returns:

A new active selector instance

selector

public static DataflowProcessor selector(java.util.List inputChannels,
                                         java.util.List outputChannels,
                                         groovy.lang.Closure code)

Creates a selector using the default dataflow parallel group

Parameters:

inputChannels - dataflow channels to use for input

outputChannels - dataflow channels to use for output

code - The selector's body to run each time a value is available in any of the inputs channels

Returns:

A new active selector instance

selector

public static DataflowProcessor selector(java.util.Map channels)

Creates a selector using the default dataflow parallel group. Since no body is provided, the selector will simply copy the incoming values to all output channels.

Parameters:

channels - A map specifying "inputs" and "outputs" - dataflow channels (instances of the DataflowQueue or DataflowVariable classes) to use for inputs and outputs

Returns:

A new active selector instance

selector

public static DataflowProcessor selector(java.util.List inputChannels,
                                         java.util.List outputChannels)

Creates a selector using the default dataflow parallel group. Since no body is provided, the selector will simply copy the incoming values to all output channels.

Parameters:

inputChannels - dataflow channels to use for input

outputChannels - dataflow channels to use for output

Returns:

A new active selector instance

prioritySelector

public static DataflowProcessor prioritySelector(java.util.Map channels,
                                                 groovy.lang.Closure code)

Creates a prioritizing selector using the default dataflow parallel group Input with lower position index have higher priority.

Parameters:

channels - A map specifying "inputs" and "outputs" - dataflow channels (instances of the DataflowQueue or DataflowVariable classes) to use for inputs and outputs

code - The selector's body to run each time a value is available in any of the inputs channels

Returns:

A new active selector instance

prioritySelector

public static DataflowProcessor prioritySelector(java.util.List inputChannels,
                                                 java.util.List outputChannels,
                                                 groovy.lang.Closure code)

Creates a prioritizing selector using the default dataflow parallel group Input with lower position index have higher priority.

Parameters:

inputChannels - dataflow channels to use for input

outputChannels - dataflow channels to use for output

code - The selector's body to run each time a value is available in any of the inputs channels

Returns:

A new active selector instance

prioritySelector

public static DataflowProcessor prioritySelector(java.util.Map channels)

Creates a prioritizing selector using the default dataflow parallel group. Since no body is provided, the selector will simply copy the incoming values to all output channels. Input with lower position index have higher priority.

Parameters:

channels - A map specifying "inputs" and "outputs" - dataflow channels (instances of the DataflowQueue or DataflowVariable classes) to use for inputs and outputs

Returns:

A new active selector instance

prioritySelector

public static DataflowProcessor prioritySelector(java.util.List inputChannels,
                                                 java.util.List outputChannels)

Creates a prioritizing selector using the default dataflow parallel group. Since no body is provided, the selector will simply copy the incoming values to all output channels. Input with lower position index have higher priority.

Parameters:

inputChannels - dataflow channels to use for input

outputChannels - dataflow channels to use for output

Returns:

A new active selector instance

splitter

public static DataflowProcessor splitter(DataflowReadChannel inputChannel,
                                         java.util.List<DataflowWriteChannel> outputChannels)

Creates a splitter copying its single input channel into all of its output channels. The created splitter will be part of the default parallel group Input with lower position index have higher priority.

Parameters:

inputChannel - The channel to read values from

outputChannels - A list of channels to output to

Returns:

A new active splitter instance

splitter

public static DataflowProcessor splitter(DataflowReadChannel inputChannel,
                                         java.util.List<DataflowWriteChannel> outputChannels,
                                         int maxForks)

Creates a splitter copying its single input channel into all of its output channels. The created splitter will be part of this parallel group Input with lower position index have higher priority.

Parameters:

inputChannel - The channel to read values from

outputChannels - A list of channels to output to

maxForks - Number of threads running the splitter's body, defaults to 1

Returns:

A new active splitter instance

select

public static Select<?> select(DataflowReadChannel<?>... channels)

Creates a select using the default dataflow parallel group. The returns Select instance will allow the user to obtain values from the supplied dataflow variables or streams as they become available.

Parameters:

channels - Dataflow variables or streams to wait for values on

Returns:

A new select instance

select

public static Select<?> select(java.util.List<DataflowReadChannel> channels)

Creates a select using the default dataflow parallel group. The returns Select instance will allow the user to obtain values from the supplied dataflow variables or streams as they become available.

Parameters:

channels - Dataflow variables or streams to wait for values on

Returns:

A new select instance

whenAllBound

public static <T> Promise<T> whenAllBound(java.util.List<Promise> promises,
                                          groovy.lang.Closure<T> code)

Without blocking the thread waits for all the promises to get bound and then passes them to the supplied closure.

Type Parameters:

T - The type of the final result

Parameters:

promises - The promises to wait for

code - A closure to execute with concrete values for each of the supplied promises

Returns:

A promise for the final result

whenAllBound

public static <T> Promise<T> whenAllBound(Promise promise1,
                                          groovy.lang.Closure<T> code)

Without blocking the thread waits for all the promises to get bound and then passes them to the supplied closure.

Type Parameters:

T - The type of the final result

Parameters:

promise1 - The promises to wait for

code - A closure to execute with concrete values for each of the supplied promises

Returns:

A promise for the final result

whenAllBound

public static <T> Promise<T> whenAllBound(Promise promise1,
                                          Promise promise2,
                                          groovy.lang.Closure<T> code)

Without blocking the thread waits for all the promises to get bound and then passes them to the supplied closure.

Type Parameters:

T - The type of the final result

Parameters:

promise1 - The promises to wait for

promise2 - The promises to wait for

code - A closure to execute with concrete values for each of the supplied promises

Returns:

A promise for the final result

whenAllBound

public static <T> Promise<T> whenAllBound(Promise promise1,
                                          Promise promise2,
                                          Promise promise3,
                                          groovy.lang.Closure<T> code)

Without blocking the thread waits for all the promises to get bound and then passes them to the supplied closure.

Type Parameters:

T - The type of the final result

Parameters:

promise1 - The promises to wait for

promise2 - The promises to wait for

promise3 - The promises to wait for

code - A closure to execute with concrete values for each of the supplied promises

Returns:

A promise for the final result

whenAllBound

public static <T> Promise<T> whenAllBound(Promise promise1,
                                          Promise promise2,
                                          Promise promise3,
                                          Promise promise4,
                                          groovy.lang.Closure<T> code)

Without blocking the thread waits for all the promises to get bound and then passes them to the supplied closure.

Type Parameters:

T - The type of the final result

Parameters:

promise1 - The promises to wait for

promise2 - The promises to wait for

promise3 - The promises to wait for

promise4 - The promises to wait for

code - A closure to execute with concrete values for each of the supplied promises

Returns:

A promise for the final result

whenAllBound

public static <T> Promise<T> whenAllBound(java.util.List<Promise> promises,
                                          groovy.lang.Closure<T> code,
                                          groovy.lang.Closure<T> errorHandler)

Without blocking the thread waits for all the promises to get bound and then passes them to the supplied closure.

Type Parameters:

T - The type of the final result

Parameters:

promises - The promises to wait for

code - A closure to execute with concrete values for each of the supplied promises

errorHandler - A closure handling an exception (an instance of Throwable), if if it gets bound

Returns:

A promise for the final result

whenAllBound

public static <T> Promise<T> whenAllBound(Promise promise1,
                                          groovy.lang.Closure<T> code,
                                          groovy.lang.Closure<T> errorHandler)

Without blocking the thread waits for all the promises to get bound and then passes them to the supplied closure.

Type Parameters:

T - The type of the final result

Parameters:

promise1 - The promises to wait for

code - A closure to execute with concrete values for each of the supplied promises

errorHandler - A closure handling an exception (an instance of Throwable), if if it gets bound

Returns:

A promise for the final result

whenAllBound

public static <T> Promise<T> whenAllBound(Promise promise1,
                                          Promise promise2,
                                          groovy.lang.Closure<T> code,
                                          groovy.lang.Closure<T> errorHandler)

Without blocking the thread waits for all the promises to get bound and then passes them to the supplied closure.

Type Parameters:

T - The type of the final result

Parameters:

promise1 - The promises to wait for

promise2 - The promises to wait for

code - A closure to execute with concrete values for each of the supplied promises

errorHandler - A closure handling an exception (an instance of Throwable), if if it gets bound

Returns:

A promise for the final result

whenAllBound

public static <T> Promise<T> whenAllBound(Promise promise1,
                                          Promise promise2,
                                          Promise promise3,
                                          groovy.lang.Closure<T> code,
                                          groovy.lang.Closure<T> errorHandler)

Without blocking the thread waits for all the promises to get bound and then passes them to the supplied closure.

Type Parameters:

T - The type of the final result

Parameters:

promise1 - The promises to wait for

promise2 - The promises to wait for

promise3 - The promises to wait for

code - A closure to execute with concrete values for each of the supplied promises

errorHandler - A closure handling an exception (an instance of Throwable), if if it gets bound

Returns:

A promise for the final result

whenAllBound

public static <T> Promise<T> whenAllBound(Promise promise1,
                                          Promise promise2,
                                          Promise promise3,
                                          Promise promise4,
                                          groovy.lang.Closure<T> code,
                                          groovy.lang.Closure<T> errorHandler)

Without blocking the thread waits for all the promises to get bound and then passes them to the supplied closure.

Type Parameters:

T - The type of the final result

Parameters:

promise1 - The promises to wait for

promise2 - The promises to wait for

promise3 - The promises to wait for

promise4 - The promises to wait for

code - A closure to execute with concrete values for each of the supplied promises

errorHandler - A closure handling an exception (an instance of Throwable), if if it gets bound

Returns:

A promise for the final result