groovyx.gpars.dataflow.operator.component

Class GracefulShutdownListener

java.lang.Object

groovyx.gpars.dataflow.operator.DataflowEventAdapter

groovyx.gpars.dataflow.operator.component.GracefulShutdownListener

All Implemented Interfaces:

DataflowEventListener

public class GracefulShutdownListener
extends DataflowEventAdapter

Listens to an operator/selector and reports its state and activity to a GracefulShutdownMonitor, shared with other listeners.

Author:

Vaclav Pech

Field Summary

Fields

Modifier and Type	Field and Description
private java.util.concurrent.atomic.AtomicInteger	activeForks
private boolean	collectingMessages
private java.util.concurrent.atomic.AtomicLong	messagesInChannels
private OperatorStateMonitor	monitor
private DataflowProcessor	processor
private boolean	shutdownFlag

Constructor Summary

Constructors

Constructor and Description
GracefulShutdownListener(OperatorStateMonitor monitor) Hooks hooks the shared monitor

Method Summary

Methods

Modifier and Type	Method and Description
void	afterRun(DataflowProcessor processor, java.util.List<java.lang.Object> messages) Enters an idle state, so a notification needs to be sent to the monitor.
java.util.List<java.lang.Object>	beforeRun(DataflowProcessor processor, java.util.List<java.lang.Object> messages) Entering a different non-idle state, so a notification needs to be sent to the monitor.
java.lang.Object	controlMessageArrived(DataflowProcessor processor, DataflowReadChannel<java.lang.Object> channel, int index, java.lang.Object message) Entering a non-idle state, so a notification needs to be sent to the monitor.
private void	fireEvent() If shutdown is in progress, we'll notify the monitor
(package private) void	initiateShutdown() Starts the shutdown phase by turning shutdownFlag on
boolean	isIdle() A quick check on, whether the operator/selector is in the Idle state
boolean	isIdleAndNoIncomingMessages() A more sophisticated test for being Idle
java.lang.Object	messageArrived(DataflowProcessor processor, DataflowReadChannel<java.lang.Object> channel, int index, java.lang.Object message) Entering a non-idle state, so a notification needs to be sent to the monitor.
void	registered(DataflowProcessor processor) When hooked into an operator/selectors, the listener will tap to its input channels to get notified about all incoming messages.
(package private) void	terminateProcessor() Used by the monitor to terminate the underlying operator/selector

Methods inherited from class groovyx.gpars.dataflow.operator.DataflowEventAdapter

afterStart, afterStop, customEvent, messageSentOut, onException

Methods inherited from class java.lang.Object

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

Field Detail

collectingMessages

private boolean collectingMessages

activeForks

private final java.util.concurrent.atomic.AtomicInteger activeForks

monitor

private final OperatorStateMonitor monitor

processor

private DataflowProcessor processor

shutdownFlag

private volatile boolean shutdownFlag

messagesInChannels

private final java.util.concurrent.atomic.AtomicLong messagesInChannels

Constructor Detail

GracefulShutdownListener

public GracefulShutdownListener(OperatorStateMonitor monitor)

Hooks hooks the shared monitor

Parameters:

monitor - The monitor that will orchestrate the shutdown

Method Detail

registered

public void registered(DataflowProcessor processor)

When hooked into an operator/selectors, the listener will tap to its input channels to get notified about all incoming messages. These incoming messages will be stored temporarily so as the listener can keep track of it until the message is passed to the operator/selector. The messageArrived() event handler will then remove the message from the temporary storage. However, it is not guaranteed that the channel reports an incoming message before the corresponding messageArrived() handler gets invoked. These cases are fine with respect to shutdown, however, we still need to take care of such situation in order to remove the message from teh temporary cache.

Specified by:

registered in interface DataflowEventListener

Overrides:

registered in class DataflowEventAdapter

Parameters:

processor - The reporting dataflow operator/selector

messageArrived

public java.lang.Object messageArrived(DataflowProcessor processor,
                              DataflowReadChannel<java.lang.Object> channel,
                              int index,
                              java.lang.Object message)

Entering a non-idle state, so a notification needs to be sent to the monitor. Also, the received message must be removed from the temporary message cache (or added to it so that so that the not-yet-arrived notification from the channel can remove it).

Specified by:

messageArrived in interface DataflowEventListener

Overrides:

messageArrived in class DataflowEventAdapter

Parameters:

processor - The reporting dataflow operator/selector

channel - The input channel holding the message

index - The index of the input channel within the operator

message - The incoming message

Returns:

The same object that was passed in as the last argument

controlMessageArrived

public java.lang.Object controlMessageArrived(DataflowProcessor processor,
                                     DataflowReadChannel<java.lang.Object> channel,
                                     int index,
                                     java.lang.Object message)

Entering a non-idle state, so a notification needs to be sent to the monitor. Also, the received message must be removed from the temporary message cache (or added to it so that so that the not-yet-arrived notification from the channel can remove it).

Specified by:

controlMessageArrived in interface DataflowEventListener

Overrides:

controlMessageArrived in class DataflowEventAdapter

Parameters:

processor - The reporting dataflow operator/selector

channel - The input channel holding the message

index - The index of the input channel within the operator

message - The incoming message

Returns:

The same object that was passed in as the last argument

beforeRun

public java.util.List<java.lang.Object> beforeRun(DataflowProcessor processor,
                                         java.util.List<java.lang.Object> messages)

Entering a different non-idle state, so a notification needs to be sent to the monitor.

Specified by:

beforeRun in interface DataflowEventListener

Overrides:

beforeRun in class DataflowEventAdapter

Parameters:

processor - The reporting dataflow operator/selector

messages - The incoming messages

Returns:

The same set of messages that was passed in

afterRun

public void afterRun(DataflowProcessor processor,
            java.util.List<java.lang.Object> messages)

Enters an idle state, so a notification needs to be sent to the monitor.

Specified by:

afterRun in interface DataflowEventListener

Overrides:

afterRun in class DataflowEventAdapter

Parameters:

processor - The reporting dataflow operator/selector

messages - The incoming messages that have been processed

fireEvent

private void fireEvent()

If shutdown is in progress, we'll notify the monitor

initiateShutdown

final void initiateShutdown()

Starts the shutdown phase by turning shutdownFlag on

isIdle

public final boolean isIdle()

A quick check on, whether the operator/selector is in the Idle state

Returns:

True, if the current state is Idle

isIdleAndNoIncomingMessages

public final boolean isIdleAndNoIncomingMessages()

A more sophisticated test for being Idle

Returns:

True, if the operator/selector state is Idle, there are no messages in the input channels and there are no messages in the intermediate state between having been removed from the channel and being accepted by the operator

terminateProcessor

final void terminateProcessor()

Used by the monitor to terminate the underlying operator/selector