groovyx.gpars.dataflow.operator
Class DataflowProcessorActor

java.lang.Object
  groovyx.gpars.serial.WithSerialId
      groovyx.gpars.actor.impl.MessageStream
          groovyx.gpars.actor.Actor
              groovyx.gpars.actor.AbstractLoopingActor
                  groovyx.gpars.actor.StaticDispatchActor<java.lang.Object>
                      groovyx.gpars.dataflow.operator.DataflowProcessorActor

All Implemented Interfaces:

java.io.Serializable

Direct Known Subclasses:

DataflowOperatorActor, DataflowSelectorActor

abstract class DataflowProcessorActor
extends StaticDispatchActor<java.lang.Object>

A base actor class for operators' and selectors' actors

Author:

Vaclav Pech

Nested Class Summary

Nested classes/interfaces inherited from class groovyx.gpars.actor.Actor

Actor.MyRemoteHandle, Actor.RemoteActor

Nested classes/interfaces inherited from class groovyx.gpars.actor.impl.MessageStream

MessageStream.RemoteMessageStream, MessageStream.SendTo

Field Summary

protected static java.lang.String	CANNOT_OBTAIN_THE_SEMAPHORE_TO_FORK_OPERATOR_S_BODY
protected groovy.lang.Closure	code
protected java.util.List	inputs
protected java.util.List	outputs
protected DataflowProcessor	owningProcessor
protected boolean	stoppingGently

Fields inherited from class groovyx.gpars.actor.Actor

ACTOR_HAS_ALREADY_BEEN_STARTED, CANNOT_SEND_REPLIES_NO_SENDER_HAS_BEEN_REGISTERED, currentThread, parallelGroup, START_MESSAGE, STOP_MESSAGE, TERMINATE_MESSAGE, TIMEOUT, TIMEOUT_MESSAGE, timer

Fields inherited from class groovyx.gpars.serial.WithSerialId

serialHandle

Constructor Summary

DataflowProcessorActor(DataflowProcessor owningProcessor, PGroup group, java.util.List outputs, java.util.List inputs, groovy.lang.Closure code)

Method Summary

(package private) void	afterStart()
(package private) void	afterStop()
protected java.lang.Object	fireMessageArrived(java.lang.Object result, int index, boolean controlMessage)
protected void	forwardPoisonPill(java.lang.Object data)
(package private) void	checkPoison(java.lang.Object data) Handles the poisson message.
(package private) static boolean	isControlMessage(java.lang.Object message)
(package private) void	onException(java.lang.Throwable e)
void	onMessage(java.lang.Object message) All messages unhandled by sub-classes will result in an exception being thrown
(package private) void	registerChannelListenersToAllInputs(DataflowChannelListener<java.lang.Object> handler) Registers the provided handler to all input channels
(package private) void	reportException(java.lang.Throwable e)
MessageStream	send(java.lang.Object message) Sends the message, ignoring exceptions caused by the actor not being active anymore

Methods inherited from class groovyx.gpars.actor.AbstractLoopingActor

getSender, hasBeenStopped, initialize, isActive, isFair, makeFair, reply, replyIfExists, setParallelGroup, setTimeout, silentStart, start, stop, sweepNextMessage, terminate

Methods inherited from class groovyx.gpars.actor.Actor

createActorMessage, createRemoteHandle, deregisterCurrentActorWithThread, getJoinLatch, getParallelGroup, handleException, handleInterrupt, handleStart, handleTermination, handleTimeout, isActorThread, join, join, join, join, onStop, registerCurrentActorWithThread, sendAndContinue, sendAndPromise, threadBoundActor

Methods inherited from class groovyx.gpars.actor.impl.MessageStream

call, getRemoteClass, leftShift, send, send, sendAndWait, sendAndWait, sendAndWait

Methods inherited from class groovyx.gpars.serial.WithSerialId

getOrCreateSerialHandle, writeReplace

Methods inherited from class java.lang.Object

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

Field Detail

CANNOT_OBTAIN_THE_SEMAPHORE_TO_FORK_OPERATOR_S_BODY

protected static final java.lang.String CANNOT_OBTAIN_THE_SEMAPHORE_TO_FORK_OPERATOR_S_BODY

See Also:

Constant Field Values

inputs

protected final java.util.List inputs

outputs

protected final java.util.List outputs

code

protected final groovy.lang.Closure code

owningProcessor

protected final DataflowProcessor owningProcessor

stoppingGently

protected boolean stoppingGently

Constructor Detail

DataflowProcessorActor

DataflowProcessorActor(DataflowProcessor owningProcessor,
                       PGroup group,
                       java.util.List outputs,
                       java.util.List inputs,
                       groovy.lang.Closure code)

Method Detail

afterStart

void afterStart()

afterStop

void afterStop()

registerChannelListenersToAllInputs

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

onException

final void onException(java.lang.Throwable e)

send

public MessageStream send(java.lang.Object message)

Sends the message, ignoring exceptions caused by the actor not being active anymore

Overrides:

send in class AbstractLoopingActor

Parameters:

message - The message to send

Returns:

The current actor

onMessage

public void onMessage(java.lang.Object message)

All messages unhandled by sub-classes will result in an exception being thrown

Specified by:

onMessage in class StaticDispatchActor<java.lang.Object>

Parameters:

message - The unhandled message

isControlMessage

static boolean isControlMessage(java.lang.Object message)

checkPoison

final void checkPoison(java.lang.Object data)

Handles the poisson message. After receiving the poisson a dataflow operator will send the poisson to all its output channels and terminate.

Parameters:

data - The poisson to re-send

forwardPoisonPill

protected void forwardPoisonPill(java.lang.Object data)

reportException

final void reportException(java.lang.Throwable e)

fireMessageArrived

protected java.lang.Object fireMessageArrived(java.lang.Object result,
                                              int index,
                                              boolean controlMessage)