SequentialProcessingActor (org.codehaus.gpars)

groovyx.gpars.actor.impl.SequentialProcessingActor

All Implemented Interfaces and Traits:: java.lang.Runnable

@java.lang.SuppressWarnings({"UnqualifiedStaticUsage"})
public abstract class SequentialProcessingActor
extends ReplyingMessageStream

Authors:: Alex Tkachman, Vaclav Pech

Nested Class Summary

Nested classes
Modifiers	Name	Description
`private static class`	`SequentialProcessingActor.Node`	Represents an element in the message queue.

Field Summary

Fields
Modifiers	Name	Description
`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<SequentialProcessingActor, Node>`	`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<SequentialProcessingActor>`	`stopFlagUpdater`
`private java.lang.Thread`	`waitingThread`

Inherited fields
Fields inherited from class	Fields
`class ReplyingMessageStream`	`sender, serialVersionUID`
`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`
`class MessageStream`	`serialVersionUID`
`class WithSerialId`	`serialHandle, serialVersionUID`

Constructor Summary

Constructors
Constructor and description
`protected SequentialProcessingActor ()` Creates a new instance, sets the default actor group.

Methods Summary

Methods
Type	Name and description
`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)`

Inherited Methods Summary

Inherited Methods
Methods inherited from class	Name
`class ReplyingMessageStream`	`getSender, reply, replyIfExists, setSender`
`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`
`class MessageStream`	`call, getRemoteClass, leftShift, reInterrupt, send, send, send, sendAndWait, sendAndWait, sendAndWait`
`class WithSerialId`	`createRemoteHandle, getOrCreateSerialHandle, getRemoteClass, writeReplace`
`class java.lang.Object`	`java.lang.Object#wait(), java.lang.Object#wait(long, int), java.lang.Object#wait(long), java.lang.Object#equals(java.lang.Object), java.lang.Object#toString(), java.lang.Object#hashCode(), java.lang.Object#getClass(), java.lang.Object#notify(), java.lang.Object#notifyAll()`

- Field Detail
  - private static final java.lang.String SHOULD_NOT_REACH_HERE
  - protected static final int S_ACTIVE_MASK
  - protected static final int S_FINISHED_MASK
  - protected static final int S_FINISHING_MASK
  - protected static final int S_NOT_STARTED
  - protected static final int S_RUNNING
  - protected static final int S_STOPPED
  - protected static final int S_STOPPING
  - protected static final int S_STOP_TERMINATE_MASK
  - protected static final int S_TERMINATED
  - protected static final int S_TERMINATING
  - @java.lang.SuppressWarnings({"UnusedDeclaration"}) //modified through inputQueryUpdater private SequentialProcessingActor.Node inputQueue
    
    Stored incoming messages. The most recently received message is in the head of the list.
  - private static final java.util.concurrent.atomic.AtomicReferenceFieldUpdater<SequentialProcessingActor, Node> inputQueueUpdater
  - private final java.util.concurrent.atomic.AtomicBoolean ongoingThreadTermination
  - 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.
  - private static final long serialVersionUID
  - protected int stopFlag
    
    Indicates whether the actor should terminate
  - protected static final java.util.concurrent.atomic.AtomicIntegerFieldUpdater<SequentialProcessingActor> stopFlagUpdater
  - private java.lang.Thread waitingThread
- Constructor Detail
  - protected SequentialProcessingActor()
    
    Creates a new instance, sets the default actor group.
- Method Detail
  - 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.
  - private ActorMessage awaitNextMessage(long endTime)
    
    Holds common functionality for takeMessage() methods.
    throws:
    InterruptedException If the thread has been interrupted
    
    Returns:
    The next message
    Parameters:
    endTime - End of the timeout, 0 if no timeout was set
  - protected final void checkStopTerminate()
  - @java.lang.Override protected void handleTermination()
  - @java.lang.Override protected final boolean hasBeenStopped()
  - @java.lang.Override public final boolean isActive()
    
    Checks the current status of the Actor.
  - protected final ActorMessage pollMessage()
    
    Polls a message from the queues
    Returns:
    The message
  - 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.
  - 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.
    
    Returns:
    The message retrieved from the queue, or null, if the timeout expires.
    Parameters:
    timeout - how long to wait before giving up, in units of unit
    units - a TimeUnit determining how to interpret the timeout parameter
  - 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.
    
    Returns:
    The message retrieved from the queue, or null, if the timeout expires.
    Parameters:
    duration - how long to wait before giving up, in units of unit
  - 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.
  - 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.
    
    Returns:
    The message retrieved from the queue, or null, if the timeout expires.
    Parameters:
    timeout - how long to wait before giving up, in units of unit
    units - a TimeUnit determining how to interpret the timeout parameter
  - private ActorMessage retrieveNextMessage()
    
    Takes the next message from the outputQueue, decrements the counter and possibly throws control exceptions
    Returns:
    The next message
  - @java.lang.Override @java.lang.SuppressWarnings({"ThrowCaughtLocally", "OverlyLongMethod"}) public void run()
  - @java.lang.Override public final MessageStream send(java.lang.Object message)
  - @java.lang.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
  - @java.lang.Override public Actor silentStart()
  - @java.lang.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
  - @java.lang.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
  - @java.lang.Override protected final ActorMessage sweepNextMessage()
    
    Removes the head of the message queue
    Returns:
    The head message, or null, if the message queue is empty
  - 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
  - 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.
    
    Returns:
    The message
    Parameters:
    timeout - Max time to wait for a message
    timeUnit - The units for the timeout
  - @java.lang.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
  - 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
  - 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)

| Detail:
Field
Constructor
Method

[Java] Class SequentialProcessingActor

Nested Class Summary

Field Summary

Constructor Summary

Methods Summary

Inherited Methods Summary

Field Detail

private static final java.lang.String SHOULD_NOT_REACH_HERE

protected static final int S_ACTIVE_MASK

protected static final int S_FINISHED_MASK

protected static final int S_FINISHING_MASK

protected static final int S_NOT_STARTED

protected static final int S_RUNNING

protected static final int S_STOPPED

protected static final int S_STOPPING

protected static final int S_STOP_TERMINATE_MASK

protected static final int S_TERMINATED

protected static final int S_TERMINATING

@java.lang.SuppressWarnings({"UnusedDeclaration"}) //modified through inputQueryUpdater private SequentialProcessingActor.Node inputQueue

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

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

private SequentialProcessingActor.Node outputQueue

private static final long serialVersionUID

protected int stopFlag

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

private java.lang.Thread waitingThread

Constructor Detail

protected SequentialProcessingActor()

Method Detail

protected void act()

private ActorMessage awaitNextMessage(long endTime)

protected final void checkStopTerminate()

@java.lang.Override protected void handleTermination()

@java.lang.Override protected final boolean hasBeenStopped()

@java.lang.Override public final boolean isActive()

protected final ActorMessage pollMessage()

protected final java.lang.Object receive()

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

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

protected java.lang.Object receiveImpl()

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

private ActorMessage retrieveNextMessage()

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

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

@java.lang.Override public final void setParallelGroup(PGroup group)

@java.lang.Override public Actor silentStart()

@java.lang.Override public final SequentialProcessingActor start()

@java.lang.Override public final Actor stop()

@java.lang.Override protected final ActorMessage sweepNextMessage()

protected final ActorMessage takeMessage()

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

@java.lang.Override public final Actor terminate()

private void throwIfNeeded(ActorMessage toProcess)

private void transferQueues()

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