groovyx.gpars.actor.impl
Class SequentialProcessingActor

java.lang.Object
  groovyx.gpars.serial.WithSerialId
      groovyx.gpars.actor.impl.MessageStream
          groovyx.gpars.actor.Actor
              groovyx.gpars.actor.impl.ReplyingMessageStream
                  groovyx.gpars.actor.impl.SequentialProcessingActor

All Implemented Interfaces:

java.io.Serializable, java.lang.Runnable

Direct Known Subclasses:

BlockingActor

public abstract class SequentialProcessingActor
extends ReplyingMessageStream
implements java.lang.Runnable

Author:

Alex Tkachman, Vaclav Pech

See Also:

Serialized Form

Nested Class Summary

private static class

SequentialProcessingActor.Node Represents an element in the message queue.

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

private SequentialProcessingActor.Node	inputQueue Stored incoming messages.
private static java.util.concurrent.atomic.AtomicReferenceFieldUpdater<SequentialProcessingActor,SequentialProcessingActor.Node>	inputQueueUpdater
private java.util.concurrent.atomic.AtomicBoolean	ongoingThreadTermination
private SequentialProcessingActor.Node	outputQueue Stores messages ready for processing by the actor.
protected static int	S_ACTIVE_MASK
protected static int	S_FINISHED_MASK
protected static int	S_FINISHING_MASK
protected static int	S_NOT_STARTED
protected static int	S_RUNNING
protected static int	S_STOP_TERMINATE_MASK
protected static int	S_STOPPED
protected static int	S_STOPPING
protected static int	S_TERMINATED
protected static int	S_TERMINATING
private static long	serialVersionUID
private static java.lang.String	SHOULD_NOT_REACH_HERE
protected int	stopFlag Indicates whether the actor should terminate
protected static java.util.concurrent.atomic.AtomicIntegerFieldUpdater<SequentialProcessingActor>	stopFlagUpdater
private java.lang.Thread	waitingThread

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

protected

SequentialProcessingActor() Creates a new instance, sets the default actor group.

Method Summary

protected abstract void	act() This method represents the body of the actor.
private ActorMessage	awaitNextMessage(long endTime) Holds common functionality for takeMessage() methods.
protected void	handleTermination()
protected boolean	hasBeenStopped()
protected void	checkStopTerminate()
boolean	isActive() Checks the current status of the Actor.
protected ActorMessage	pollMessage() Polls a message from the queues
protected java.lang.Object	receive() Retrieves a message from the message queue, waiting, if necessary, for a message to arrive.
protected java.lang.Object	receive(groovy.time.BaseDuration duration) Retrieves a message from the message queue, waiting, if necessary, for a message to arrive.
protected java.lang.Object	receive(long timeout, java.util.concurrent.TimeUnit units) Retrieves a message from the message queue, waiting, if necessary, for a message to arrive.
protected abstract java.lang.Object	receiveImpl() Retrieves a message from the message queue, waiting, if necessary, for a message to arrive.
protected abstract java.lang.Object	receiveImpl(long timeout, java.util.concurrent.TimeUnit units) Retrieves a message from the message queue, waiting, if necessary, for a message to arrive.
private ActorMessage	retrieveNextMessage() Takes the next message from the outputQueue, decrements the counter and possibly throws control exceptions
void	run()
MessageStream	send(java.lang.Object message) Send message to stream and return immediately
void	setParallelGroup(PGroup group) Sets the actor's group.
Actor	silentStart() Starts the Actor without sending the START_MESSAGE message to speed the start-up.
SequentialProcessingActor	start() Starts the Actor.
Actor	stop() Send message to stop to the actor.
protected ActorMessage	sweepNextMessage() Removes the head of the message queue
protected ActorMessage	takeMessage() Takes a message from the queues.
protected ActorMessage	takeMessage(long timeout, java.util.concurrent.TimeUnit timeUnit) Takes a message from the queues.
Actor	terminate() Terminate the Actor.
private void	throwIfNeeded(ActorMessage toProcess) Checks the supplied message and throws either STOP or TERMINATE, if the message is a Stop or Terminate message respectively.
private void	transferQueues() Transfers messages from the input queue into the output queue, reverting the order of the elements.
private static java.lang.Object	unwrapMessage(java.lang.Object msg)

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

getSender, reply, replyIfExists, setSender

Methods inherited from class groovyx.gpars.actor.Actor

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

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

call, getRemoteClass, leftShift, reInterrupt, 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

serialVersionUID

private static final long serialVersionUID

See Also:

Constant Field Values

inputQueue

private volatile SequentialProcessingActor.Node inputQueue

Stored incoming messages. The most recently received message is in the head of the list.

outputQueue

private SequentialProcessingActor.Node outputQueue

Stores messages ready for processing by the actor. The oldest message is in the head of the list. Messages are transferred from the inputQueue into the output queue in the transferQueues() method.

ongoingThreadTermination

private final java.util.concurrent.atomic.AtomicBoolean ongoingThreadTermination

inputQueueUpdater

private static final java.util.concurrent.atomic.AtomicReferenceFieldUpdater<SequentialProcessingActor,SequentialProcessingActor.Node> inputQueueUpdater

waitingThread

private volatile java.lang.Thread waitingThread

S_ACTIVE_MASK

protected static final int S_ACTIVE_MASK

See Also:

Constant Field Values

S_FINISHING_MASK

protected static final int S_FINISHING_MASK

See Also:

Constant Field Values

S_FINISHED_MASK

protected static final int S_FINISHED_MASK

See Also:

Constant Field Values

S_STOP_TERMINATE_MASK

protected static final int S_STOP_TERMINATE_MASK

See Also:

Constant Field Values

S_NOT_STARTED

protected static final int S_NOT_STARTED

See Also:

Constant Field Values

S_RUNNING

protected static final int S_RUNNING

See Also:

Constant Field Values

S_STOPPING

protected static final int S_STOPPING

See Also:

Constant Field Values

S_TERMINATING

protected static final int S_TERMINATING

See Also:

Constant Field Values

S_STOPPED

protected static final int S_STOPPED

See Also:

Constant Field Values

S_TERMINATED

protected static final int S_TERMINATED

See Also:

Constant Field Values

stopFlag

protected volatile int stopFlag

Indicates whether the actor should terminate

stopFlagUpdater

protected static final java.util.concurrent.atomic.AtomicIntegerFieldUpdater<SequentialProcessingActor> stopFlagUpdater

SHOULD_NOT_REACH_HERE

private static final java.lang.String SHOULD_NOT_REACH_HERE

See Also:

Constant Field Values

Constructor Detail

SequentialProcessingActor

protected SequentialProcessingActor()

Creates a new instance, sets the default actor group.

Method Detail

isActive

public final boolean isActive()

Checks the current status of the Actor.

Specified by:

isActive in class Actor

Returns:

current status of the Actor.

throwIfNeeded

private void throwIfNeeded(ActorMessage toProcess)

Checks the supplied message and throws either STOP or TERMINATE, if the message is a Stop or Terminate message respectively.

Parameters:

toProcess - The next message to process by the actors

pollMessage

protected final ActorMessage pollMessage()

Polls a message from the queues

Returns:

The message

takeMessage

protected final ActorMessage takeMessage()
                                  throws java.lang.InterruptedException

Takes a message from the queues. Blocks until a message is available.

Returns:

The message

Throws:

java.lang.InterruptedException - If the thread gets interrupted.

takeMessage

protected ActorMessage takeMessage(long timeout,
                                   java.util.concurrent.TimeUnit timeUnit)
                            throws java.lang.InterruptedException

Takes a message from the queues. Blocks until a message is available.

Parameters:

timeout - Max time to wait for a message

timeUnit - The units for the timeout

Returns:

The message

Throws:

java.lang.InterruptedException - If the thread gets interrupted.

awaitNextMessage

private ActorMessage awaitNextMessage(long endTime)
                               throws java.lang.InterruptedException

Holds common functionality for takeMessage() methods.

Parameters:

endTime - End of the timeout, 0 if no timeout was set

Returns:

The next message

Throws:

java.lang.InterruptedException - If the thread has been interrupted

retrieveNextMessage

private ActorMessage retrieveNextMessage()

Takes the next message from the outputQueue, decrements the counter and possibly throws control exceptions

Returns:

The next message

transferQueues

private void transferQueues()

Transfers messages from the input queue into the output queue, reverting the order of the elements.

setParallelGroup

public final void setParallelGroup(PGroup group)

Sets the actor's group. It can only be invoked before the actor is started.

Overrides:

setParallelGroup in class Actor

Parameters:

group - new group

send

public final MessageStream send(java.lang.Object message)

Description copied from class: MessageStream

Send message to stream and return immediately

Specified by:

send in class MessageStream

Parameters:

message - message to send

Returns:

always return message stream itself

hasBeenStopped

protected final boolean hasBeenStopped()

Specified by:

hasBeenStopped in class Actor

handleTermination

protected void handleTermination()

Overrides:

handleTermination in class Actor

silentStart

public Actor silentStart()

Description copied from class: Actor

Starts the Actor without sending the START_MESSAGE message to speed the start-up. The potential custom afterStart handlers won't be run. No messages can be sent or received before an Actor is started.

Specified by:

silentStart in class Actor

Returns:

same actor

start

public final SequentialProcessingActor start()

Starts the Actor. No messages can be send or received before an Actor is started.

Specified by:

start in class Actor

Returns:

this (the actor itself) to allow method chaining

stop

public final Actor stop()

Send message to stop to the actor. All messages in queue will be processed before stopped but no new messages will be accepted after that point

Specified by:

stop in class Actor

Returns:

this (the actor itself) to allow method chaining

terminate

public final Actor terminate()

Terminate the Actor. The background thread will be interrupted, unprocessed messages will be passed to the afterStop method, if exists. Has no effect if the Actor is not started.

Specified by:

terminate in class Actor

Returns:

this (the actor itself) to allow method chaining

receiveImpl

protected abstract java.lang.Object receiveImpl()
                                         throws java.lang.InterruptedException

Retrieves a message from the message queue, waiting, if necessary, for a message to arrive.

Returns:

The message retrieved from the queue, or null, if the timeout expires.

Throws:

java.lang.InterruptedException - If the thread is interrupted during the wait. Should propagate up to stop the thread.

receiveImpl

protected abstract java.lang.Object receiveImpl(long timeout,
                                                java.util.concurrent.TimeUnit units)
                                         throws java.lang.InterruptedException

Retrieves a message from the message queue, waiting, if necessary, for a message to arrive.

Parameters:

timeout - how long to wait before giving up, in units of unit

units - a TimeUnit determining how to interpret the timeout parameter

Returns:

The message retrieved from the queue, or null, if the timeout expires.

Throws:

java.lang.InterruptedException - If the thread is interrupted during the wait. Should propagate up to stop the thread.

receive

protected final java.lang.Object receive()
                                  throws java.lang.InterruptedException

Retrieves a message from the message queue, waiting, if necessary, for a message to arrive.

Returns:

The message retrieved from the queue, or null, if the timeout expires.

Throws:

java.lang.InterruptedException - If the thread is interrupted during the wait. Should propagate up to stop the thread.

receive

protected final java.lang.Object receive(long timeout,
                                         java.util.concurrent.TimeUnit units)
                                  throws java.lang.InterruptedException

Retrieves a message from the message queue, waiting, if necessary, for a message to arrive.

Parameters:

timeout - how long to wait before giving up, in units of unit

units - a TimeUnit determining how to interpret the timeout parameter

Returns:

The message retrieved from the queue, or null, if the timeout expires.

Throws:

java.lang.InterruptedException - If the thread is interrupted during the wait. Should propagate up to stop the thread.

unwrapMessage

private static java.lang.Object unwrapMessage(java.lang.Object msg)

receive

protected final java.lang.Object receive(groovy.time.BaseDuration duration)
                                  throws java.lang.InterruptedException

Retrieves a message from the message queue, waiting, if necessary, for a message to arrive.

Parameters:

duration - how long to wait before giving up, in units of unit

Returns:

The message retrieved from the queue, or null, if the timeout expires.

Throws:

java.lang.InterruptedException - If the thread is interrupted during the wait. Should propagate up to stop the thread.

sweepNextMessage

protected final ActorMessage sweepNextMessage()

Removes the head of the message queue

Specified by:

sweepNextMessage in class Actor

Returns:

The head message, or null, if the message queue is empty

act

protected abstract void act()

This method represents the body of the actor. It is called upon actor's start and can exit either normally by return or due to actor being stopped through the stop() method, which cancels the current actor action. Provides an extension point for subclasses to provide their custom Actor's message handling code.

run

public void run()

Specified by:

run in interface java.lang.Runnable

checkStopTerminate

protected final void checkStopTerminate()