groovyx.gpars.actor.impl
Class SequentialProcessingActor

java.lang.Object
  extended by groovyx.gpars.serial.WithSerialId
      extended by groovyx.gpars.actor.impl.MessageStream
          extended by groovyx.gpars.actor.Actor
              extended by groovyx.gpars.actor.impl.ReplyingMessageStream
                  extended by 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()

Copyright © 2008–2012 Václav Pech. All Rights Reserved.