Overview  Package   Class  Deprecated  Index  Help 
org.codehaus.gpars
FRAMES    NO FRAMES    
SUMMARY: NESTED | FIELD | CONSTR | METHOD DETAIL: FIELD | CONSTR | METHOD

groovyx.gpars.actor.impl
[Java] 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.lang.Runnable
@SuppressWarnings({"UnqualifiedStaticUsage"})
public abstract class SequentialProcessingActor
extends ReplyingMessageStream

Authors:
Alex Tkachman, Vaclav Pech

Nested Class Summary
private static class SequentialProcessingActor.Node

Represents an element in the message queue.

 
Field Summary
private static java.lang.String SHOULD_NOT_REACH_HERE

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_STOPPED

protected static int S_STOPPING

protected static int S_STOP_TERMINATE_MASK

protected static int S_TERMINATED

protected static int S_TERMINATING

private SequentialProcessingActor.Node inputQueue

Stored incoming messages.

private static java.util.concurrent.atomic.AtomicReferenceFieldUpdater inputQueueUpdater

private java.util.concurrent.atomic.AtomicBoolean ongoingThreadTermination

private SequentialProcessingActor.Node outputQueue

Stores messages ready for processing by the actor.

private static long serialVersionUID

protected int stopFlag

Indicates whether the actor should terminate
protected static java.util.concurrent.atomic.AtomicIntegerFieldUpdater stopFlagUpdater

private java.lang.Thread waitingThread

 
Fields inherited from class ReplyingMessageStream
sender, serialVersionUID
 
Fields inherited from class Actor
ACTOR_HAS_ALREADY_BEEN_STARTED, AFTER_START, CANNOT_SEND_REPLIES_NO_SENDER_HAS_BEEN_REGISTERED, EMPTY_ARGUMENTS, ON_DELIVERY_ERROR, RESPONDS_TO, START_MESSAGE, STOP_MESSAGE, TERMINATE_MESSAGE, TIMEOUT, TIMEOUT_MESSAGE, currentActorPerThread, currentThread, joinLatch, onStop, parallelGroup, serialVersionUID, timer
 
Fields inherited from class MessageStream
serialVersionUID
 
Fields inherited from class WithSerialId
serialHandle, serialVersionUID
 
Constructor Summary
protected SequentialProcessingActor()

Creates a new instance, sets the default actor group.
 
Method Summary
protected void act()

This method represents the body of the actor.

private ActorMessage awaitNextMessage(long endTime)

Holds common functionality for takeMessage() methods.
protected void checkStopTerminate()

protected void handleTermination()

protected boolean hasBeenStopped()

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(long timeout, java.util.concurrent.TimeUnit units)

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 receiveImpl()

Retrieves a message from the message queue, waiting, if necessary, for a message to arrive.
protected 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)

void setParallelGroup(PGroup group)

Sets the actor's group.

Actor silentStart()

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 ReplyingMessageStream
getSender, reply, replyIfExists, setSender
 
Methods inherited from class Actor
callDynamic, createActorMessage, createRemoteHandle, deregisterCurrentActorWithThread, getJoinLatch, getParallelGroup, handleException, handleInterrupt, handleStart, handleTermination, handleTimeout, hasBeenStopped, isActive, isActorThread, join, join, join, join, onStop, registerCurrentActorWithThread, sendAndContinue, sendAndPromise, setParallelGroup, silentStart, start, stop, sweepNextMessage, sweepQueue, terminate, threadBoundActor
 
Methods inherited from class MessageStream
call, getRemoteClass, leftShift, reInterrupt, send, send, send, sendAndWait, sendAndWait, sendAndWait
 
Methods inherited from class WithSerialId
createRemoteHandle, getOrCreateSerialHandle, getRemoteClass, writeReplace
 

Field Detail

SHOULD_NOT_REACH_HERE

private static final java.lang.String SHOULD_NOT_REACH_HERE

S_ACTIVE_MASK

protected static final int S_ACTIVE_MASK

S_FINISHED_MASK

protected static final int S_FINISHED_MASK

S_FINISHING_MASK

protected static final int S_FINISHING_MASK

S_NOT_STARTED

protected static final int S_NOT_STARTED

S_RUNNING

protected static final int S_RUNNING

S_STOPPED

protected static final int S_STOPPED

S_STOPPING

protected static final int S_STOPPING

S_STOP_TERMINATE_MASK

protected static final int S_STOP_TERMINATE_MASK

S_TERMINATED

protected static final int S_TERMINATED

S_TERMINATING

protected static final int S_TERMINATING

inputQueue

@SuppressWarnings({"UnusedDeclaration"})
    //modified through inputQueryUpdater
private SequentialProcessingActor.Node inputQueue
Stored incoming messages. The most recently received message is in the head of the list.

inputQueueUpdater

private static final java.util.concurrent.atomic.AtomicReferenceFieldUpdater inputQueueUpdater

ongoingThreadTermination

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

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.

serialVersionUID

private static final long serialVersionUID

stopFlag

protected int stopFlag
Indicates whether the actor should terminate

stopFlagUpdater

protected static final java.util.concurrent.atomic.AtomicIntegerFieldUpdater stopFlagUpdater

waitingThread

private java.lang.Thread waitingThread

 
Constructor Detail

SequentialProcessingActor

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

 
Method Detail

act

protected 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.

awaitNextMessage

private ActorMessage awaitNextMessage(long endTime)
Holds common functionality for takeMessage() methods.
throws:
InterruptedException If the thread has been interrupted
Parameters:
endTime - End of the timeout, 0 if no timeout was set
Returns:
The next message

checkStopTerminate

protected final void checkStopTerminate()

handleTermination

@Override
protected void handleTermination()

hasBeenStopped

@Override
protected final boolean hasBeenStopped()

isActive

@Override
public final boolean isActive()
Checks the current status of the Actor.

pollMessage

protected final ActorMessage pollMessage()
Polls a message from the queues
Returns:
The message

receive

protected final java.lang.Object receive()
Retrieves a message from the message queue, waiting, if necessary, for a message to arrive.
throws:
InterruptedException If the thread is interrupted during the wait. Should propagate up to stop the thread.
Returns:
The message retrieved from the queue, or null, if the timeout expires.

receive

protected final 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.
throws:
InterruptedException If the thread is interrupted during the wait. Should propagate up to stop the thread.
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.

receive

protected final java.lang.Object receive(groovy.time.BaseDuration duration)
Retrieves a message from the message queue, waiting, if necessary, for a message to arrive.
throws:
InterruptedException If the thread is interrupted during the wait. Should propagate up to stop the thread.
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.

receiveImpl

protected java.lang.Object receiveImpl()
Retrieves a message from the message queue, waiting, if necessary, for a message to arrive.
throws:
InterruptedException If the thread is interrupted during the wait. Should propagate up to stop the thread.
Returns:
The message retrieved from the queue, or null, if the timeout expires.

receiveImpl

protected 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.
throws:
InterruptedException If the thread is interrupted during the wait. Should propagate up to stop the thread.
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.

retrieveNextMessage

private ActorMessage retrieveNextMessage()
Takes the next message from the outputQueue, decrements the counter and possibly throws control exceptions
Returns:
The next message

run

@Override
@SuppressWarnings({"ThrowCaughtLocally", "OverlyLongMethod"})
public void run()

send

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

setParallelGroup

@Override
public final void setParallelGroup(PGroup group)
Sets the actor's group. It can only be invoked before the actor is started.
Parameters:
group - new group

silentStart

@Override
public Actor silentStart()

start

@Override
public final SequentialProcessingActor start()
Starts the Actor. No messages can be send or received before an Actor is started.
Returns:
this (the actor itself) to allow method chaining

stop

@Override
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
Returns:
this (the actor itself) to allow method chaining

sweepNextMessage

@Override
protected final ActorMessage sweepNextMessage()
Removes the head of the message queue
Returns:
The head message, or null, if the message queue is empty

takeMessage

protected final ActorMessage takeMessage()
Takes a message from the queues. Blocks until a message is available.
throws:
InterruptedException If the thread gets interrupted.
Returns:
The message

takeMessage

protected ActorMessage takeMessage(long timeout, java.util.concurrent.TimeUnit timeUnit)
Takes a message from the queues. Blocks until a message is available.
throws:
InterruptedException If the thread gets interrupted.
Parameters:
timeout - Max time to wait for a message
timeUnit - The units for the timeout
Returns:
The message

terminate

@Override
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.
Returns:
this (the actor itself) to allow method chaining

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

transferQueues

private void transferQueues()
Transfers messages from the input queue into the output queue, reverting the order of the elements.

unwrapMessage

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

 

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