Field Detail

• public static final java.lang.String INPUTS

• private static final java.lang.String LISTENERS

• public static final java.lang.String MAX_FORKS

• public static final java.lang.String OUTPUTS

• public static final java.lang.String STATE_OBJECT

• protected DataflowProcessorActor actor

  The internal actor performing on behalf of the processor

• protected final java.util.Collection<DataflowEventListener> listeners

• protected final java.lang.Object stateObject

  May hold custom state provided at construction time and read within the body

Constructor Detail

• protected DataflowProcessor(java.util.Map<java.lang.String, java.lang.Object> channels, groovy.lang.Closure<?> code)

  Creates a processor After creation the processor needs to be started using the start() method.

  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 processor's body to run each time all inputs have a value to read

Method Detail

• public final void addDataflowEventListener(DataflowEventListener listener)

• public final void bindAllOutputValues(java.lang.Object... values)

  Used by the processor's body to send a value to all output channels. The values passed as arguments will each be sent to an output channel with identical position index.

  If the maxForks value is set to a value greater than 1, calls to bindAllOutputs may result in values written to different channels to be in different order. If this is a problem for the application logic, the bindAllOutputsAtomically method should be considered instead.

  Parameters:

  values - Values to send to output channels of the same position index

• public final void bindAllOutputValuesAtomically(java.lang.Object... values)

  Used by the processor's body to send a value to all output channels, while guaranteeing atomicity of the operation and preventing other calls to bindAllOutputsAtomically() from interfering with one another. The values passed as arguments will each be sent to an output channel with identical position index.

  Parameters:

  values - Values to send to output channels of the same position index

• public final void bindAllOutputs(java.lang.Object value)

  Used by the processor's body to send a value to all output channels. If the maxForks value is set to a value greater than 1, calls to bindAllOutputs may result in values written to different channels to be in different order. If this is a problem for the application logic, the bindAllOutputsAtomically method should be considered instead.

  Parameters:

  value - The value to bind

• public final void bindAllOutputsAtomically(java.lang.Object value)

  Used by the processor's body to send a value to all output channels, while guaranteeing atomicity of the operation and preventing other calls to bindAllOutputsAtomically() from interfering with one another.

  Parameters:

  value - The value to bind

• public final void bindOutput(int idx, java.lang.Object value)

  Used by the processor's body to send a value to the given output channel

  Parameters:

  idx - The index of the channel to bind

  value - The value to bind

• public final void bindOutput(java.lang.Object value)

  Used by the processor's body to send a value to the first / only output channel

  Parameters:

  value - The value to bind

• protected static void checkMaxForks(java.util.Map<?, ?> channels)

• static java.util.List<DataflowReadChannel<?>> extractInputs(java.util.Map<java.lang.String, java.lang.Object> channels)

• private static java.util.Collection<DataflowEventListener> extractListeners(java.util.Map<java.lang.String, java.lang.Object> channels)

• static java.util.List<DataflowWriteChannel<?>> extractOutputs(java.util.Map<java.lang.String, java.lang.Object> channels)

• private static java.lang.Object extractState(java.util.Map<java.lang.String, java.lang.Object> channels)

• protected final void fireAfterRun(java.util.List<java.lang.Object> messages)

• protected final void fireAfterStart()

• protected final void fireAfterStop()

• protected final java.util.List<java.lang.Object> fireBeforeRun(java.util.List<java.lang.Object> messages)

• protected final java.lang.Object fireControlMessageArrived(DataflowReadChannel channel, int index, java.lang.Object message)

• public final java.lang.Object fireCustomEvent(java.lang.Object data)

• protected final java.lang.Object fireMessageArrived(DataflowReadChannel channel, int index, java.lang.Object message)

• protected final java.lang.Object fireMessageSentOut(DataflowWriteChannel channel, int index, java.lang.Object message)

• protected final boolean fireOnException(java.lang.Throwable e)

• public final DataflowWriteChannel<?> getOutput()

  The processor's first / only output channel

  Returns:

  The particular DataflowWriteChannel instance

• public final DataflowWriteChannel<?> getOutputs(int idx)

  The processor's output channel of the given index

  Returns:

  The particular DataflowWriteChannel instance

  Parameters:

  idx - The index of the channel to retrieve

• public final java.util.List<DataflowWriteChannel<?>> getOutputs()

  The processor's all output channels

  Returns:

  A List holding all output channels

• public final java.lang.Object getStateObject()

  Retrieves the custom state object

  Returns:

  The state object associated with the operator

• public final void join()

  Joins the processor waiting for it to finish

  throws:

  InterruptedException If the thread gets interrupted

• public final void registerChannelListenersToAllInputs(DataflowChannelListener<java.lang.Object> handler)

  Registers the provided handler to all input channels

  Parameters:

  handler - The closure to invoke whenever a value gets bound to any of the input channels

• public final void removeDataflowEventListener(DataflowEventListener listener)

• @java.lang.SuppressWarnings({"UseOfSystemOutOrSystemErr"}) final void reportError(java.lang.Throwable e)

  Is invoked in case the actor throws an exception.

  Parameters:

  e - The reported exception

• static boolean shouldBeMultiThreaded(java.util.Map<java.lang.String, java.lang.Object> channels)

• public final DataflowProcessor start(PGroup group)

  Starts a processor using the specified parallel group

  Returns:

  This operator instance

  Parameters:

  group - The parallel group to use with the processor

• public final DataflowProcessor start()

  Starts a processor using the specified parallel group

  Returns:

  This operator instance

• public final void terminate()

  Stops the processor immediately, potentially loosing unhandled messages

• public final void terminateAfterNextRun()

  Gently stops the processor after the next set of messages is handled. Unlike with terminate(), no messages will get lost. If the operator never gets triggered after calling the terminateAfterNextRun() method, the operator never really stops.