alma.acs.nc

Class AcsEventSubscriberImplBase<T>

java.lang.Object

alma.acs.nc.AcsEventSubscriberImplBase<T>

Type Parameters:

T - The event (base) type. If all events are of the same type then that type should be used; otherwise a common base type for all events that may be sent on the given "channel" should be used, such as Object or IDLEntity.

All Implemented Interfaces:

AcsEventSubscriber<T>, AcsScxmlActionExecutor<EventSubscriberAction>

Direct Known Subclasses:

NCSubscriber

public abstract class AcsEventSubscriberImplBase<T>
extends java.lang.Object
implements AcsEventSubscriber<T>, AcsScxmlActionExecutor<EventSubscriberAction>

Base class for an event subscriber, that can be used for both Corba NC and for in-memory test NC.

We will have to see if it can also be used for DDS-based events, or if the required modifications will be too heavy.

Nested Class Summary

Nested classes/interfaces inherited from interface alma.acs.nc.AcsEventSubscriber

AcsEventSubscriber.Callback<U>, AcsEventSubscriber.GenericCallback

Field Summary

Fields

Modifier and Type	Field and Description
protected java.lang.String	clientName A name that identifies the client of this NCSubscriber, to be used both for logging and also (if applicable) as the supplier proxy name so that looking at the proxy objects of the NC we can figure out who the clients are.
static int	EVENT_QUEUE_CAPACITY Event queue should hold at least two events to avoid unnecessary scary logs about slow receivers, but must be short enough to get receivers to actually implement their own queue and discard mechanism instead of relying on this ACS queue which may buffer events for a limited time and thus obscure the problem.
protected java.lang.Class<T>	eventType Runtime access to the type parameter <T>.
protected AcsEventSubscriber.GenericCallback	genericReceiver Contains a generic receiver to be used by the #addGenericSubscription() method.
protected java.util.logging.Logger	logger Provides access to the ACS logging system.
protected MultipleRepeatGuard	processTimeLogRepeatGuard Contains a list of repeat guards for each different type of event.
protected java.util.Map<java.lang.Class<? extends T>,AcsEventSubscriber.Callback<? extends T>>	receivers Contains a list of receiver functions to be invoked when an event of a particular type is received.
protected ContainerServicesBase	services
protected AcsScxmlEngine<EventSubscriberSignal,EventSubscriberAction>	stateMachine State machine for the subscriber lifecycle.
protected EventSubscriberSignalDispatcher	stateMachineSignalDispatcher

Constructor Summary

Constructors

Constructor and Description
AcsEventSubscriberImplBase(ContainerServicesBase services, java.lang.String clientName, java.lang.Class<T> eventType) Base class constructor, to be called from subclass ctor.

Method Summary

Modifier and Type	Method and Description
void	addGenericSubscription(AcsEventSubscriber.GenericCallback receiver) Subscribes to all events.
<U extends T> void	addSubscription(AcsEventSubscriber.Callback<U> receiver) Adds a handler that will receive events of a specific type.
protected void	createConnectionAction(EventDispatcher evtDispatcher, ErrorReporter errRep, SCInstance scInstance, java.util.Collection<TriggerEvent> derivedEvents) Subclass may override, but must call super.createConnectionAction().
protected void	createEnvironmentAction(EventDispatcher evtDispatcher, ErrorReporter errRep, SCInstance scInstance, java.util.Collection<TriggerEvent> derivedEvents) Subclass may override, but must call super.createEnvironmentAction().
protected void	destroyConnectionAction(EventDispatcher evtDispatcher, ErrorReporter errRep, SCInstance scInstance, java.util.Collection<TriggerEvent> derivedEvents) Handler for "destroyConnection" state machine action.
protected void	destroyEnvironmentAction(EventDispatcher evtDispatcher, ErrorReporter errRep, SCInstance scInstance, java.util.Collection<TriggerEvent> derivedEvents) Subclass may override, but must call super.destroyEnvironmentAction().
void	disconnect() Disconnects this subscriber from the Notification Channel, and releases all the resources associated with it.
boolean	execute(EventSubscriberAction action, EventDispatcher evtDispatcher, ErrorReporter errRep, SCInstance scInstance, java.util.Collection<TriggerEvent> derivedEvents) Dispatches the action enum to one of the action handler methods.
java.lang.String	getLifecycleState() Returns the lifecycle state as obtained from an internal state machine.
protected abstract double	getMaxProcessTimeSeconds(java.lang.String eventName) Gets the configured (or default) max time that a receiver may take to process an event, regardless of the actual event rate.
int	getNumberOfReceivers() Use only for unit testing!
boolean	hasGenericReceiver() Use only for unit testing!
boolean	isDisconnected()
boolean	isSuspended() Returns true if this subscriber has been suspended.
protected abstract boolean	isTraceEventsEnabled()
protected abstract void	logEventProcessingTimeExceeded(java.lang.String eventName, long logOcurrencesNumber) Logs the error that event processing time was exceeded.
protected abstract void	logEventProcessingTooSlowForEventRate(long numEventsDiscarded, java.lang.String eventName) Logs the error that the receiver cannot keep up with the actual event rate, in spite of the small event buffering done.
protected abstract void	logEventReceiveHandlerException(java.lang.String eventName, java.lang.String receiverClassName, java.lang.Throwable thr) Logs an exception thrown by an event handler (user code).
protected abstract void	logNoEventReceiver(java.lang.String eventName) Logs or ignores the fact that an event was received for which no receiver could be found.
protected abstract void	logQueueShutdownError(int timeoutMillis, int remainingEvents) Logs the error that the local event buffer could not be emptied before shutting down the subscriber.
protected abstract void	notifyFirstSubscription(java.lang.Class<?> structClass)
protected abstract void	notifyNoSubscription()
protected abstract void	notifySubscriptionRemoved(java.lang.Class<?> structClass)
protected void	processEvent(java.lang.Object eventData, EventDescription eventDesc) This method should be called from the subclass-specific method that receives the event, for example push_structured_event in case of Corba NC, or preferably via processEventAsync(Object, EventDescription).
protected void	processEventAsync(java.lang.Object eventData, EventDescription eventDesc) Asynchronously calls processEvent(Object, EventDescription), using eventHandlingExecutor.
void	removeGenericSubscription() Removes the generic receiver handler.
<U extends T> void	removeSubscription(java.lang.Class<U> structClass) Removes the subscription for a specified event type or for all events types, so that the handler previously registered for that event type will no longer receive events.
void	resume() Used to reenable the Subscriber after a call to the suspend() method.
protected void	resumeAction(EventDispatcher evtDispatcher, ErrorReporter errRep, SCInstance scInstance, java.util.Collection<TriggerEvent> derivedEvents) Subclass may override, but must call super.resumeAction().
void	startReceivingEvents() This method must be called to actually start receiving events.
void	suspend() Used to temporarily halt receiving events of all types.
protected void	suspendAction(EventDispatcher evtDispatcher, ErrorReporter errRep, SCInstance scInstance, java.util.Collection<TriggerEvent> derivedEvents) Subclass may override, but must call super.suspendAction().

Methods inherited from class java.lang.Object

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

Field Detail

clientName

protected final java.lang.String clientName

A name that identifies the client of this NCSubscriber, to be used both for logging and also (if applicable) as the supplier proxy name so that looking at the proxy objects of the NC we can figure out who the clients are.

services

protected final ContainerServicesBase services

logger

protected final java.util.logging.Logger logger

Provides access to the ACS logging system.

stateMachine

protected final AcsScxmlEngine<EventSubscriberSignal,EventSubscriberAction> stateMachine

State machine for the subscriber lifecycle. It is not used for data handling itself, even though the measured overhead of less than 0.2 ms per signal might allow this.

stateMachineSignalDispatcher

protected final EventSubscriberSignalDispatcher stateMachineSignalDispatcher

EVENT_QUEUE_CAPACITY

public static final int EVENT_QUEUE_CAPACITY

Event queue should hold at least two events to avoid unnecessary scary logs about slow receivers, but must be short enough to get receivers to actually implement their own queue and discard mechanism instead of relying on this ACS queue which may buffer events for a limited time and thus obscure the problem.

See Also:

eventHandlingExecutor, Constant Field Values

genericReceiver

protected AcsEventSubscriber.GenericCallback genericReceiver

Contains a generic receiver to be used by the #addGenericSubscription() method.

receivers

protected final java.util.Map<java.lang.Class<? extends T>,AcsEventSubscriber.Callback<? extends T>> receivers

Contains a list of receiver functions to be invoked when an event of a particular type is received.

key = the event type (Java class derived from IDL struct).
value = the matching event handler.

processTimeLogRepeatGuard

protected final MultipleRepeatGuard processTimeLogRepeatGuard

Contains a list of repeat guards for each different type of event.

eventType

protected final java.lang.Class<T> eventType

Runtime access to the type parameter <T>.

Constructor Detail

AcsEventSubscriberImplBase

public AcsEventSubscriberImplBase(ContainerServicesBase services,
                                  java.lang.String clientName,
                                  java.lang.Class<T> eventType)
                           throws AcsJException

Base class constructor, to be called from subclass ctor. IMPORTANT: Subclasses MUST call "stateMachine.setUpEnvironment()" at the end of their constructors. This is because Java does not support template method design for constructors (see for example http://stackoverflow.com/questions/2906958/running-a-method-after-the-constructor-of-any-derived-class). Since the subclasses are all meant to be produced within ACS I did not want to go for dirty tricks to work around this. We also don't want to call a public init() method after constructing the subscriber.

Normally an ACS class such as container services will act as the factory for event subscriber objects, but for exceptional cases it is also possible to create one stand-alone, as long as the required parameters can be provided.

Parameters:

services - To get ACS logger, access to the CDB, etc.

clientName - A name that identifies the client of this NCSubscriber. TODO: Check if we still need this name to be specified separately from services#getName().

Throws:

AcsJException - Thrown on any really bad error conditions encountered.

Method Detail

isTraceEventsEnabled

protected abstract boolean isTraceEventsEnabled()

Returns:

true if events should be logged when they are received.

execute

public boolean execute(EventSubscriberAction action,
                       EventDispatcher evtDispatcher,
                       ErrorReporter errRep,
                       SCInstance scInstance,
                       java.util.Collection<TriggerEvent> derivedEvents)
                throws AcsJStateMachineActionEx

Dispatches the action enum to one of the action handler methods.

The overhead of implementing this method is the price we pay for avoiding reflection and allowing flexible action implementation in the SM design.

Specified by:

execute in interface AcsScxmlActionExecutor<EventSubscriberAction>

Returns:

true if the action was recognized.

Throws:

AcsJStateMachineActionEx

See Also:

alma.acs.nc.sm.generic.AcsScxmlActionExecutor#execute(java.lang.Enum, org.apache.commons.scxml.EventDispatcher, org.apache.commons.scxml.ErrorReporter, org.apache.commons.scxml.SCInstance, java.util.Collection)

createEnvironmentAction

protected void createEnvironmentAction(EventDispatcher evtDispatcher,
                                       ErrorReporter errRep,
                                       SCInstance scInstance,
                                       java.util.Collection<TriggerEvent> derivedEvents)
                                throws AcsJStateMachineActionEx

Subclass may override, but must call super.createEnvironmentAction().

Throws:

AcsJStateMachineActionEx

destroyEnvironmentAction

protected void destroyEnvironmentAction(EventDispatcher evtDispatcher,
                                        ErrorReporter errRep,
                                        SCInstance scInstance,
                                        java.util.Collection<TriggerEvent> derivedEvents)
                                 throws AcsJStateMachineActionEx

Subclass may override, but must call super.destroyEnvironmentAction().

Throws:

AcsJStateMachineActionEx

createConnectionAction

protected void createConnectionAction(EventDispatcher evtDispatcher,
                                      ErrorReporter errRep,
                                      SCInstance scInstance,
                                      java.util.Collection<TriggerEvent> derivedEvents)
                               throws AcsJStateMachineActionEx

Subclass may override, but must call super.createConnectionAction().

Throws:

AcsJStateMachineActionEx

destroyConnectionAction

protected void destroyConnectionAction(EventDispatcher evtDispatcher,
                                       ErrorReporter errRep,
                                       SCInstance scInstance,
                                       java.util.Collection<TriggerEvent> derivedEvents)
                                throws AcsJStateMachineActionEx

Handler for "destroyConnection" state machine action.

Shuts down the event queue. Queued events may still be processed by the receivers afterwards, but here we wait for up to 500 ms to log it if it is the case.

Further events delivered to processEventAsync(Object, EventDescription) will cause an exception there.

Subclass may override, but must call super.destroyConnectionAction().

Throws:

AcsJStateMachineActionEx

suspendAction

protected void suspendAction(EventDispatcher evtDispatcher,
                             ErrorReporter errRep,
                             SCInstance scInstance,
                             java.util.Collection<TriggerEvent> derivedEvents)
                      throws AcsJStateMachineActionEx

Subclass may override, but must call super.suspendAction().

Throws:

AcsJStateMachineActionEx

resumeAction

protected void resumeAction(EventDispatcher evtDispatcher,
                            ErrorReporter errRep,
                            SCInstance scInstance,
                            java.util.Collection<TriggerEvent> derivedEvents)
                     throws AcsJStateMachineActionEx

Subclass may override, but must call super.resumeAction().

Throws:

AcsJStateMachineActionEx

getMaxProcessTimeSeconds

protected abstract double getMaxProcessTimeSeconds(java.lang.String eventName)

Gets the configured (or default) max time that a receiver may take to process an event, regardless of the actual event rate.

logEventReceiveHandlerException

protected abstract void logEventReceiveHandlerException(java.lang.String eventName,
                                                        java.lang.String receiverClassName,
                                                        java.lang.Throwable thr)

Logs an exception thrown by an event handler (user code).

logEventProcessingTimeExceeded

protected abstract void logEventProcessingTimeExceeded(java.lang.String eventName,
                                                       long logOcurrencesNumber)

Logs the error that event processing time was exceeded.

logEventProcessingTooSlowForEventRate

protected abstract void logEventProcessingTooSlowForEventRate(long numEventsDiscarded,
                                                              java.lang.String eventName)

Logs the error that the receiver cannot keep up with the actual event rate, in spite of the small event buffering done.

Parameters:

numEventsDiscarded - The number of events that have actually been discarded since the last log. Will often be 0 when we just warn about the queue filling up.

eventName -

logNoEventReceiver

protected abstract void logNoEventReceiver(java.lang.String eventName)

Logs or ignores the fact that an event was received for which no receiver could be found.

The subclass must know whether such a condition is expected or not, e.g. because event filtering is set up outside of the subscriber and only subscribed event types are expected to arrive.

logQueueShutdownError

protected abstract void logQueueShutdownError(int timeoutMillis,
                                              int remainingEvents)

Logs the error that the local event buffer could not be emptied before shutting down the subscriber.

processEventAsync

protected void processEventAsync(java.lang.Object eventData,
                                 EventDescription eventDesc)

Asynchronously calls processEvent(Object, EventDescription), using eventHandlingExecutor.

This method should be called from the subclass-specific method that receives the event, for example push_structured_event in case of Corba NC.

This method is thread-safe.

Parameters:

eventData - (defined as Object instead of T to include data for generic subscription).

eventDesc - event meta data

processEvent

protected void processEvent(java.lang.Object eventData,
                            EventDescription eventDesc)

This method should be called from the subclass-specific method that receives the event, for example push_structured_event in case of Corba NC, or preferably via processEventAsync(Object, EventDescription).

No exception is allowed to be thrown by this method, even if the receiver implementation throws a RuntimeExecption

Parameters:

eventData - (defined as Object instead of T to include data for generic subscription).

eventDesc -

getLifecycleState

public java.lang.String getLifecycleState()

Description copied from interface: AcsEventSubscriber

Returns the lifecycle state as obtained from an internal state machine.

This state can be used only for debug output. The state names may change over time and should not be used to control execution.

Specified by:

getLifecycleState in interface AcsEventSubscriber<T>

hasGenericReceiver

public boolean hasGenericReceiver()

Use only for unit testing!

getNumberOfReceivers

public int getNumberOfReceivers()

Use only for unit testing!

addGenericSubscription

public final void addGenericSubscription(AcsEventSubscriber.GenericCallback receiver)
                                  throws AcsJEventSubscriptionEx

Subscribes to all events. The latest generic receiver displaces the previous one.

If in addition to this generic subscription we also have specific subscriptions via #addSubscription(Class, Callback), then those more specific subscriptions will take precedence in receiving an event.

Notice though that any server-side filters previously created for the event type specific subscriptions get deleted when calling this method, so that even after removing a generic subscription the network performance gain of server-side filtering is lost.

Specified by:

addGenericSubscription in interface AcsEventSubscriber<T>

Parameters:

receiver - The callback to use when receiving events

Throws:

AcsJEventSubscriptionEx - If there is a problem and the generic receiver cannot be added

removeGenericSubscription

public final void removeGenericSubscription()
                                     throws AcsJEventSubscriptionEx

Removes the generic receiver handler.

Specified by:

removeGenericSubscription in interface AcsEventSubscriber<T>

Throws:

AcsJCORBAProblemEx

AcsJEventSubscriptionEx - If there is any problem while unsubscribing the generic receiver

addSubscription

public final <U extends T> void addSubscription(AcsEventSubscriber.Callback<U> receiver)
                                         throws AcsJEventSubscriptionEx

Description copied from interface: AcsEventSubscriber

Adds a handler that will receive events of a specific type. The event type is determined by the value returned by calling AcsEventSubscriber.Callback.getEventType() on the receiver parameter.

Note that the same event type can only be subscribed to with one handler, which means that another handler added for the same type will replace the previous handler.

The event type must be that of actual events, not of base classes. For example, the AcsEventSubscriber could be parameterized with base type IDLEntity and we could then add two subscriptions for events defined as IDL structs, say AntennaStatus and TemperatureData. If you want to subscribe to events without knowing their exact type, use AcsEventSubscriber.addGenericSubscription(GenericCallback) instead.

Specified by:

addSubscription in interface AcsEventSubscriber<T>

Parameters:

receiver - The callback to use when receiving events for the specified type.

Throws:

AcsJEventSubscriptionEx - If there is a problem and the receiver cannot be added

removeSubscription

public final <U extends T> void removeSubscription(java.lang.Class<U> structClass)
                                            throws AcsJEventSubscriptionEx

Description copied from interface: AcsEventSubscriber

Removes the subscription for a specified event type or for all events types, so that the handler previously registered for that event type will no longer receive events.

Specified by:

removeSubscription in interface AcsEventSubscriber<T>

Parameters:

structClass - the event type to be unsubscribed. If null, then all subscriptions but the generic subscription are removed.

Throws:

AcsJEventSubscriptionEx - if the specified event type has not been previously subscribed or if the removal fails with a technical problem.

startReceivingEvents

public final void startReceivingEvents()
                                throws AcsJIllegalStateEventEx,
                                       AcsJCouldntPerformActionEx

Description copied from interface: AcsEventSubscriber

This method must be called to actually start receiving events. Typically it is called after the subscriptions are set up. User may still add and remove subscriptions at any given time, though. Also, the connection can be suspended and resumed.

No further invocations should be attempted on this method after one has been already successful. Otherwise, an AcsJIllegalStateEventEx will be thrown.

If this method is not called, no event will ever be received

Specified by:

startReceivingEvents in interface AcsEventSubscriber<T>

Throws:

AcsJIllegalStateEventEx - If the user calls this method on an object that is already receiving events

AcsJCouldntPerformActionEx - If any error happens while trying to start receiving events

disconnect

public final void disconnect()
                      throws AcsJIllegalStateEventEx,
                             AcsJCouldntPerformActionEx

Description copied from interface: AcsEventSubscriber

Disconnects this subscriber from the Notification Channel, and releases all the resources associated with it. After this call, all registered handlers will stop receiving events, and this subscriber becomes unusable.

Calling this method over a subscriber object that has been already disconnected will throw an AcsJIllegalStateEventEx.

Specified by:

disconnect in interface AcsEventSubscriber<T>

Throws:

AcsJIllegalStateEventEx - If this method is called on an AcsEventSubscriber object that has been already disconnected

AcsJCouldntPerformActionEx

See Also:

AcsEventSubscriber.disconnect()

suspend

public final void suspend()
                   throws AcsJIllegalStateEventEx,
                          AcsJCouldntPerformActionEx

Description copied from interface: AcsEventSubscriber

Used to temporarily halt receiving events of all types.

If the Subscriber has been connected already (method AcsEventSubscriber.startReceivingEvents(), then after calling this method, incoming events will be buffered instead of being discarded; unexpired events will be received later, after a call to AcsEventSubscriber.resume().

// * This call has no effect if the Subscriber is not connected, or if it is // * connected but already suspended.

Specified by:

suspend in interface AcsEventSubscriber<T>

Throws:

AcsJIllegalStateEventEx - if the subscriber is not connected to an NC.

AcsJCouldntPerformActionEx

resume

public final void resume()
                  throws AcsJIllegalStateEventEx,
                         AcsJCouldntPerformActionEx

Description copied from interface: AcsEventSubscriber

Used to reenable the Subscriber after a call to the suspend() method. Queued events will be received after this call, see AcsEventSubscriber.suspend(). This call has no effect if the Subscriber is not connected, or if it is connected and already processing events.

Specified by:

resume in interface AcsEventSubscriber<T>

Throws:

AcsJIllegalStateEventEx

AcsJCouldntPerformActionEx

notifyFirstSubscription

protected abstract void notifyFirstSubscription(java.lang.Class<?> structClass)
                                         throws AcsJEventSubscriptionEx

Parameters:

structClass - Can be null in case of generic subscription.

Throws:

AcsJEventSubscriptionEx

notifySubscriptionRemoved

protected abstract void notifySubscriptionRemoved(java.lang.Class<?> structClass)
                                           throws AcsJEventSubscriptionEx

Parameters:

structClass -

Throws:

AcsJEventSubscriptionEx

notifyNoSubscription

protected abstract void notifyNoSubscription()

isSuspended

public boolean isSuspended()

Description copied from interface: AcsEventSubscriber

Returns true if this subscriber has been suspended.

Specified by:

isSuspended in interface AcsEventSubscriber<T>

isDisconnected

public final boolean isDisconnected()