com.neeve.aep
Class AepEngine

java.lang.Object
  com.neeve.root.RootObject
      com.neeve.aep.AepEngine

All Implemented Interfaces:

IEventHandler, IEventSource, IStoreEventHandler

public final class AepEngine
extends RootObject
implements IStoreEventHandler, IEventHandler, IEventSource

The AEP engine.

Threading:

In general, an AEP engine is not safe for concurrent access by multiple threads i.e. the public methods exposed by the engine cannot be invoked concurrently by multiple threads. However, the public methods can be invoked concurrently with inbound message/event dispatch which executes in the context of AEP runtime threads not necessarily under the control of the application.

Nested Class Summary

static class	AepEngine.AppExceptionHandlingPolicy Enumerates an engine's application exception handling policy.
static class	AepEngine.EventMultiplexerType Enumerates an engine's supported event multiplexer types
static class	AepEngine.HAPolicy Enumerates an engine's supported HA policies
static class	AepEngine.InboundEventAcknowledgementPolicy Enumerates an engine's inbound event acknowledgement policy.
static class	AepEngine.InboundMessageLoggingFailurePolicy Enumerates policies for handling inbound message logging failures.
static class	AepEngine.InboundMessageLoggingPolicy Enumerates an engine's inbound message logging policy.
static class	AepEngine.MessageBusBindingFailPolicy Enumerates an engine's message bus binding fail policy.
static class	AepEngine.MessageHandlingPolicy Enumerates an engine's inbound message handling policy
static class	AepEngine.MessageSendExceptionHandlingPolicy Enumerates an engine's message send exception handling policy.
static class	AepEngine.MessageSendPolicy Deprecated.
static class	AepEngine.MessagingStartFailPolicy Enumerates an engine's messaging start fail policy.
static class	AepEngine.MessagingState Enumerates an AEP engine's messaging state.
static class	AepEngine.OutboundMessageLoggingFailurePolicy Enumerates policies for handling outbound message logging failures.
static class	AepEngine.OutboundMessageLoggingPolicy Enumerates an engine's outbound message logging policy.
static class	AepEngine.PerTransactionStatsLoggingFailurePolicy Enumerates policies for handling transaction stats logging failures.
static class	AepEngine.PerTransactionStatsLoggingPolicy Enumerates an engine's transaction stats logging policy.
static class	AepEngine.ReplicationPolicy Enumerates an engine's supported replication policies
static class	AepEngine.StartupExpectation Enumerates the set of startup expectations that can be requested.
static class	AepEngine.State Enumerates an AEP engine's state.

Field Summary

String	CLIENT_BUS_NAME
static String	DEFAULT_CHANNEL_KEY The default channel key
static String	DEFAULT_CHANNEL_NAME Name of the default channel in each bus
static String	DEFAULT_CHANNEL_ROUTING_KEY Name of the routing key field of the default channel key
static String	STATS_INTERVAL_PROPNAME Property controlling whether or not the engine's stats dump thread is enabled.
static String	SYSTATS_INTERVAL_PROPNAME Property controlling whether or not the engine's sys stats dump thread is enabled.

Method Summary

void	cancelSchedule(AepScheduleEvent event) Cancel the dispatch of a scheduled event.
static AepEngine	create(AepEngineDescriptor engineDescriptor, IAepApplicationStateFactory stateFactory, Set<Object> eventHandlerContainers, IEventHandler defaultEventHandler, Collection<AepMessageSender> messageSenders, IAepWatcher watcher) Create an AEP engine.
static AepEngine	create(AepEngineDescriptor engineDescriptor, IAepApplicationStateFactory stateFactory, Set<Object> eventHandlerContainers, IEventHandler defaultEventHandler, IAepWatcher watcher) Create an AEP engine.
int	createTransactionSavepoint() Creates a savepoint in the current transaction.
AepEngine.AppExceptionHandlingPolicy	getAppExceptionHandlingPolicy() Get an engine's application exception handling policy.
<T> T	getApplicationState(int flowid) Get the application state associated with a flow.
<T> T	getApplicationState(MessageView message) Get the application state associated with an inbound message.
Collection<AepBusManager>	getBusManagers() Get an engine's bus managers.
AepEngineDescriptor	getDescriptor() Get an engine's descriptor.
long	getEngineTime() Get current engine time.
AepEventDispatcher	getEventDispatcher() Get an engine's event dispatcher
IEventMultiplexer	getEventMultiplexer() Get an engine's event multiplexer.
AepEngine.EventMultiplexerType	getEventMultiplexerType() Get an engine's event multiplexer type.
AepEngine.HAPolicy	getHAPolicy() Get an engine's HA policy.
IRogMessageLogger	getInboundMessageLogger() Get an engine's inbound message logger.
AepEngine.InboundMessageLoggingPolicy	getInboundMessageLoggingPolicy() Get an engine's inbound message logging policy.
AepEngine.MessageBusBindingFailPolicy	getMessageBusBindingFailPolicy() Get an engine's message bus binding fail policy.
AepEngine.MessageHandlingPolicy	getMessageHandlingPolicy() Get an engine's message handling policy.
AepEngine.MessageSendExceptionHandlingPolicy	getMessageSendExceptionHandlingPolicy() Get an engine's message send exception handling policy.
AepEngine.MessagingStartFailPolicy	getMessagingStartFailPolicy() Get an engine's messaging start fail policy.
AepEngine.MessagingState	getMessagingState() Get an engine's messaging state.
String	getName() Implementation of IEventSource.getName()
IRogMessageLogger	getOutboundMessageLogger() Get an engine's outbound message logger.
AepEngine.OutboundMessageLoggingPolicy	getOutboundMessageLoggingPolicy() Get an engine's outbound message logging policy.
AepEngine.ReplicationPolicy	getReplicationPolicy() Get an engine's replication policy.
MessageBusBinding	getServerBusBinding(String name) Get an application's 'server' message bus binding.
AepEngine.State	getState() Get an engine's state.
IAepEngineStats	getStats() Get the set of a engine's operational statistics.
IStoreBinding	getStore() Get an engine's operational store.
int	getTransactionSavepoint() Get the current savepoint for the transaction.
void	injectMessage(IRogMessage message) Enqueue a message into this AepEngine's event multiplexer.
void	injectMessage(IRogMessage message, boolean nonBlocking) Enqueue a message into this AepEngine's event multiplexer.
void	injectMessage(IRogMessage message, boolean nonBlocking, int delay) Enqueue a message into this AepEngine's event multiplexer.
boolean	isClustered() Get if an engine is running in clustered or non-clustered mode.
boolean	isDispatchThread() Tests if the current thread is currently dispatching an AepEngine event.
boolean	isPersistent() Get if an engine is running in persistent or non-persistent mode.
boolean	isPrimary() Get if the engine is running as primary or secondary.
void	multiplexMessage(IRogMessage message) Deprecated. This method is deprecated by injectMessage(IRogMessage)
void	multiplexMessage(IRogMessage message, boolean nonBlocking) Deprecated. This method is deprecated by injectMessage(IRogMessage, boolean)
void	onEvent(Event event) Implementation of IEventHandler.onEvent(com.neeve.event.Event) This method is the engine multiplexer's event handler, it should never be invoked directly by the application.
void	onEvent(IStoreEvent event) Implementation of IStoreEventHandler.onEvent(com.neeve.ods.IStoreEvent) This method is the store event handler and should never be invoked directly by the application.
void	registerCounterStat(IStats.Counter counter) Registers a user counter stat with the engine stats.
AepEngine	registerEnvironmentProvider(IAepEnvironmentProvider provider) Register an environment provider.
AepEngine	registerFactory(Object factory) Register a factory.
void	registerGaugeStat(IStats.Gauge gauge) Registers a user Gauge stat with the engine stats.
AepEngine	registerMessageFactory(MessageViewFactory factory) Register a message factory.
AepEngine	registerObjectFactory(IStoreObjectFactory factory) Register an object factory
void	registerSeriesStat(IStats.Series series) Registers a user latency stat with the engine stats.
AepEngine	registerTypeFactory(com.neeve.sto.StoTypeFactory factory) Register a type factory.
void	replyToClient(MessageChannel channel, PktPacket request, PktPacket reply, int replyCount, int replyNum) Send a success reply to a client request.
void	replyToClientWithError(MessageChannel channel, PktPacket request, PktPacket reply, long errorCode) Send an error reply to a client request.
void	resumeCommit(AepTransactionStageEvent.ICommitContext context) Resume a suspended transaction commit.
void	rollbackToTransactionSavepoint(int savepoint) Rolls back state changes and outbound messages to a given savepoint.
AepScheduleEvent	scheduleMessage(IRogMessage message) Schedule a message for immediate dispatch.
AepScheduleEvent	scheduleMessage(IRogMessage message, int delay) Schedule a message for a (one-time) delayed dispatch with the default schedule HA policy.
AepScheduleEvent	scheduleMessage(IRogMessage message, int delay, AepScheduleEvent.HAPolicy haPolicy) Schedule a message for (one-time) delayed dispatch.
AepScheduleEvent	scheduleObject(Object object) Schedule an opaque object for immediate dispatch.
AepScheduleEvent	scheduleObject(Object object, int interval) Schedule an opaque object for periodic dispatch with the default schedule HA policy.
AepScheduleEvent	scheduleObject(Object object, int interval, AepScheduleEvent.HAPolicy haPolicy) Schedule an opaque object for periodic dispatch at a specified interval.
void	sendMessage(MessageChannel channel, IRogMessage message) Send an outbound message.
void	sendMessage(MessageChannel channel, IRogMessage message, MessageChannel.RawKeyResolutionTable keyResolutionTable) Send a message using a XString key resolution table.
void	sendMessage(MessageChannel channel, IRogMessage message, Properties keyResolutionTable) Send a message.
void	sendMessage(MessageChannel channel, IRogMessage message, String key, Properties keyResolutionTable) Send an outbound message.
void	sendMessage(MessageChannel channel, IRogMessage message, XString key, MessageChannel.RawKeyResolutionTable keyResolutionTable) Send an outbound message.
void	setAppExceptionHandlingPolicy(AepEngine.AppExceptionHandlingPolicy policy) Set an engine's application exception handling policy.
void	setAsLastTransaction(Exception cause, boolean ignoreIfStarting) Set current transaction as last transction (only for primary agent) This method invoked setAsLastTransaction(cause, ignoreIfStarting, false)
void	setAsLastTransaction(Exception cause, boolean ignoreIfStarting, boolean lastClusterTransaction) Set current transaction as the last transaction.
void	setAsynchronousEventHandler(IAepAsynchronousEventHandler handler) Sets the handler for asynchronous engine events.
void	setCentralMessageHandler(IAepPredispatchMessageHandler handler) Deprecated. use setPredispatchMessageHandler(IAepPredispatchMessageHandler) instead
void	setMessageFilter(MessageViewFilter messageFilter) Sets a MessageFilter for live messages received by an AepEngine operating in the primary role.
void	setMessageHandlingPolicy(AepEngine.MessageHandlingPolicy policy) Set an engine's message handling policy.
void	setMessageSendExceptionHandlingPolicy(AepEngine.MessageSendExceptionHandlingPolicy policy) Set an engine's message send exception handling policy.
IAepPostdispatchMessageHandler	setPostdispatchMessageHandler(IAepPostdispatchMessageHandler handler) Register the central post dispatch message handler.
IAepPredispatchMessageHandler	setPredispatchMessageHandler(IAepPredispatchMessageHandler handler) Register the pre dispatch message handler.
void	start() Start an AEP engine.
void	start(Map<AepEngine.StartupExpectation,Enum<?>> expectations) Deprecated. This method has been deprecated. Startup expectations should be set in the engine's descriptor rather than supplied to this start method
void	stop() Stop an AEP engine (normal).
void	stop(Exception cause) Stop an AEP engine.
void	waitForMessagingToStart() Wait for messaging to be started.

Methods inherited from class com.neeve.root.RootObject

getChecked, getThreaded, getTracer, setChecked, setTracer

Methods inherited from class java.lang.Object

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

Field Detail

CLIENT_BUS_NAME

public final String CLIENT_BUS_NAME

DEFAULT_CHANNEL_NAME

public static final String DEFAULT_CHANNEL_NAME

Name of the default channel in each bus

See Also:

Constant Field Values

DEFAULT_CHANNEL_ROUTING_KEY

public static final String DEFAULT_CHANNEL_ROUTING_KEY

Name of the routing key field of the default channel key

See Also:

Constant Field Values

DEFAULT_CHANNEL_KEY

public static final String DEFAULT_CHANNEL_KEY

The default channel key

See Also:

Constant Field Values

STATS_INTERVAL_PROPNAME

public static final String STATS_INTERVAL_PROPNAME

Property controlling whether or not the engine's stats dump thread is enabled.

Can be set to a postive integer indicate the period in seconds at which the engine's stats dump thread will dump recorded engine statistics. Setting a value of 0 disables creation of the stats thread.

When enabled engine stats are traced to the logger 'nv.aep.engine.stats' at a level of Tracer.Level.INFO, so to see dumped stats a trace level of 'nv.aep.engine.stats.trace=info' must be enabled.

NOTE:: disabling the engine stats thread only stops stats from being periodically traced. It does not stop the engine from collecting stats ... stats can still be collected by an external thread (such as the Talon Server which reports the stats in server heartbeats). In other words enabling the stats thread is not a prerequisite for collecting stats, and disabling the stats reporting thread does not stop them from being collected.

NOTE: that while collection of engine stats is a zero garbage operation, tracing engine stats is not a zero garbage when performed by this stats thread. For latency sensitive apps it is recommended to run in a Talon server which can collect engine stats and report them in heartbeats in a zero garbage fashion.

See Also:

Constant Field Values

SYSTATS_INTERVAL_PROPNAME

public static final String SYSTATS_INTERVAL_PROPNAME

Property controlling whether or not the engine's sys stats dump thread is enabled.

Can be set to a postive integer indicate the period in seconds at which the system stats are traced.

When enabled engine stats are traced to the logger 'nv.aep.aep.engine.systats', at a level of 'debug', so to see dumped stats a trace level of 'nv.aep.aep.engine.systats.trace=info' must be enabled.

System stats printing is useful when not executing in a Talon server, but in most cases, it makes more sense to use server stats to report system stats, so in practice this property will not generally be used.

NOTE: tracing system stats is not a zero garbage when performed by this stats thread. For latency sensitive apps it is recommended to run in a Talon server which can collect system stats and report them in heartbeats in a near zero garbage fashion.

See Also:

Constant Field Values

Method Detail

create

public static final AepEngine create(AepEngineDescriptor engineDescriptor,
                                     IAepApplicationStateFactory stateFactory,
                                     Set<Object> eventHandlerContainers,
                                     IEventHandler defaultEventHandler,
                                     Collection<AepMessageSender> messageSenders,
                                     IAepWatcher watcher)

Create an AEP engine.

Parameters:

engineDescriptor - A descriptor object containing messaging and persistence information of interest to the application. In particular, the descriptor specifies the SMA messaging buses and channels of interest to the application and the ODS store to join.

stateFactory - The application's state factory interface. This is the interface used by the AEP engine to create application state instances that it attaches to each inbound message flow. A value of null indicates that the application does not want the AEP engine to store application state. If the application specifies null in this parameter but specifies an ODS store in the engine descriptor, the engine will store messaging state in the ODS store to ensure exactly once delivery on failover.

eventHandlerContainers - This parameer specifies the set of objects in the application that implement the message and event handlers. On creation, the engine walks through these objects and identifies handlers that are annotated with EventHandler.

defaultEventHandler - This parameter identifies the application's default event handler. Inbound events that do not match an annotated handler are passed to this handler. This parameter can be null.

messageSenders - The set of message senders to register with the engine. The engine will add these senders into the event handler container list.

watcher - This parameter specifies a watcher object. The AEP engine dispatches significant operational events to the watcher through the course of its operation. This parameter can be null.

create

public static final AepEngine create(AepEngineDescriptor engineDescriptor,
                                     IAepApplicationStateFactory stateFactory,
                                     Set<Object> eventHandlerContainers,
                                     IEventHandler defaultEventHandler,
                                     IAepWatcher watcher)

Create an AEP engine.

This method invokes create(engineDescriptor, stateFactory, eventHandlerContainers, defaultEventHandler, null, watcher)

getEventDispatcher

public final AepEventDispatcher getEventDispatcher()

Get an engine's event dispatcher

getDescriptor

public final AepEngineDescriptor getDescriptor()

Get an engine's descriptor.

This method returns the descriptor using which the engine was created.

isClustered

public final boolean isClustered()

Get if an engine is running in clustered or non-clustered mode.

An engine is considered to operate in clustered mode if and only if it is configured to attach to a store and the store is operating in clustered mode.

isPersistent

public final boolean isPersistent()

Get if an engine is running in persistent or non-persistent mode.

An engine is considered to operate in persistent mode if and only if it is configured to attach to a store and the store is operating in persistent mode.

Returns:

true if the engine is the engine is configured with a persister.

getHAPolicy

public final AepEngine.HAPolicy getHAPolicy()

Get an engine's HA policy.

getReplicationPolicy

public final AepEngine.ReplicationPolicy getReplicationPolicy()

Get an engine's replication policy.

getMessagingStartFailPolicy

public final AepEngine.MessagingStartFailPolicy getMessagingStartFailPolicy()

Get an engine's messaging start fail policy.

getMessageBusBindingFailPolicy

public final AepEngine.MessageBusBindingFailPolicy getMessageBusBindingFailPolicy()

Get an engine's message bus binding fail policy.

getInboundMessageLoggingPolicy

public final AepEngine.InboundMessageLoggingPolicy getInboundMessageLoggingPolicy()

Get an engine's inbound message logging policy.

getOutboundMessageLoggingPolicy

public final AepEngine.OutboundMessageLoggingPolicy getOutboundMessageLoggingPolicy()

Get an engine's outbound message logging policy.

setMessageHandlingPolicy

public final void setMessageHandlingPolicy(AepEngine.MessageHandlingPolicy policy)

Set an engine's message handling policy.

This method can be invoked in any state.

Parameters:

policy - The message handling policy to set.

getMessageHandlingPolicy

public final AepEngine.MessageHandlingPolicy getMessageHandlingPolicy()

Get an engine's message handling policy.

setAppExceptionHandlingPolicy

public final void setAppExceptionHandlingPolicy(AepEngine.AppExceptionHandlingPolicy policy)

Set an engine's application exception handling policy.

getAppExceptionHandlingPolicy

public final AepEngine.AppExceptionHandlingPolicy getAppExceptionHandlingPolicy()

Get an engine's application exception handling policy.

setMessageSendExceptionHandlingPolicy

public final void setMessageSendExceptionHandlingPolicy(AepEngine.MessageSendExceptionHandlingPolicy policy)

Set an engine's message send exception handling policy.

getMessageSendExceptionHandlingPolicy

public final AepEngine.MessageSendExceptionHandlingPolicy getMessageSendExceptionHandlingPolicy()

Get an engine's message send exception handling policy.

getEventMultiplexerType

public final AepEngine.EventMultiplexerType getEventMultiplexerType()

Get an engine's event multiplexer type.

registerMessageFactory

public final AepEngine registerMessageFactory(MessageViewFactory factory)

Register a message factory.

Parameters:

factory - The message factory to register

Returns:

Returns this object for invocation chaining.

Threading:

This method is not safe for concurrent access by multiple threads with itself or any of the other engine methods.

This method is a convenience method to programmatically register a message factory with the SMA runtime. This method is one of the various mechanisms available to register a message factory with the SMA runtime. The other two are to configure the factory in the platform repository or directly invoke MessageViewFactoryRegistry to install the factory (which is what this method does). The SMA runtime uses the collection of all administratively configured and programatically registered factories to instantiate received application messages.

This method is only permitted on an engine that has not been started.

registerObjectFactory

public final AepEngine registerObjectFactory(IStoreObjectFactory factory)

Register an object factory

Parameters:

factory - The object factory to register

Returns:

Returns this object for invocation chaining.

Threading:

This method is not safe for concurrent access by multiple threads with itself or any of the other engine methods.

This method is a convenience method to programmatically register an object factory with the ODS runtime. This method is one of the various mechanisms available to register a object factory with the ODS runtime. The other two are to configure the factory in the platform repository or directly invoke StoreObjectFactoryRegistry to install the factory (which is what this method does). The ODS runtime uses the collection of all administratively configured and programatically registered factories to instantiate replicated data objects.

This method is only permitted on an engine that has not been started.

registerTypeFactory

public final AepEngine registerTypeFactory(com.neeve.sto.StoTypeFactory factory)

Register a type factory.

This method is a convenience method to programmatically register a type factory with the STO runtime. This method is one of the various mechanisms available to register a type factory with the STO runtime. The other two are to configure the factory in the platform repository or directly invoke StoTypeFactoryRegistry to install the factory (which is what this method does). The ODS runtime uses the collection of all administratively configured and programatically registered factories to instantiate transmitted complex types.

This method is only permitted on an engine that has not been started.

Parameters:

factory - The type factory to register.

Returns:

Returns this object for invocation chaining.

Threading:

This method is not safe for concurrent access by multiple threads with itself or any of the other engine methods.

registerFactory

public final AepEngine registerFactory(Object factory)

Register a factory.

Parameters:

factory - The factory to register

Returns:

Returns this object for invocation chaining.

Threading:

This method is not safe for concurrent access by multiple threads with itself or any of the other engine methods.

This method is a convenience methods that detects the type of factory supplied and appropriately calls registerMessageFactory(com.neeve.sma.MessageViewFactory), registerObjectFactory(com.neeve.ods.IStoreObjectFactory) and/or registerTypeFactory(com.neeve.sto.StoTypeFactory)

This method is only permitted on an engine that has not been started.

registerEnvironmentProvider

public final AepEngine registerEnvironmentProvider(IAepEnvironmentProvider provider)

Register an environment provider.

This method registers an environment provider with an AEP engine.

This method may not be called on an engine:

• That has been started (state != AepEngine.State.Init)
• That is configured for parallel replicationion AepEngineDescriptor.getReplicateInParallel() == false

Parameters:

provider - The provider to register.

Returns:

Returns this object for invocation chaining.

Throws:

IllegalArgumentException - if the provider is null

IllegalStateException - if the engine is not in a state that allows environment providers to be registered.

Threading:

This method is not safe for concurrent access by multiple threads with itself or any of the other engine methods.

registerSeriesStat

public final void registerSeriesStat(IStats.Series series)

Registers a user latency stat with the engine stats. A user stat must be registered prior to starting the engine.

Parameters:

series - The stat to add

Throws:

IllegalStateException - if the engine has been started

See Also:

StatsFactory.createSeriesStat(String)

registerCounterStat

public final void registerCounterStat(IStats.Counter counter)

Registers a user counter stat with the engine stats. A user stat must be registered prior to starting the engine.

Parameters:

counter - The stat to add

Throws:

IllegalStateException - if the engine has been started

See Also:

StatsFactory.createCounterStat(String)

registerGaugeStat

public final void registerGaugeStat(IStats.Gauge gauge)

Registers a user Gauge stat with the engine stats. A user stat must be registered prior to starting the engine.

Parameters:

gauge - The stat to add

Throws:

IllegalStateException - if the engine has been started

See Also:

IStats.Gauge

setCentralMessageHandler

@Deprecated
public final void setCentralMessageHandler(IAepPredispatchMessageHandler handler)

Deprecated. use setPredispatchMessageHandler(IAepPredispatchMessageHandler) instead

setPredispatchMessageHandler

public final IAepPredispatchMessageHandler setPredispatchMessageHandler(IAepPredispatchMessageHandler handler)

Register the pre dispatch message handler.

This method registers the pre dispatch message handler. The predispatch handler is invoked prior to message/event annotated handlers and the default handler.

This method can be invoked at any point in an engine's lifecycle.

Parameters:

handler - The handler to register (null to deregister)

Returns:

The previous predispatch message handler if one existed.

setPostdispatchMessageHandler

public final IAepPostdispatchMessageHandler setPostdispatchMessageHandler(IAepPostdispatchMessageHandler handler)

Register the central post dispatch message handler.

A post dispatch message handler is useful for doing any post message processing book keeping.

The post dispatch handler will not be invoked if an exception is thrown from either the IAepPredispatchMessageHandler, or a normal or default message handler, but an application may register an EventHandler for AepApplicationExceptionEvent to be notified that dispatch did not complete.

This method registers the postdispatch message handler. The postdispatch handler is invoked subsequent to message/event annotated handlers and the default handler.

This method can be invoked at any point in an engine's lifecycle.

Parameters:

handler - The handler to register (null to deregister)

Returns:

The previous central message handler if one existed.

setAsynchronousEventHandler

public final void setAsynchronousEventHandler(IAepAsynchronousEventHandler handler)

Sets the handler for asynchronous engine events. This handler is invoked for events that aren't appropriate for dispatch from the engine's default dispatcher thread.

Parameters:

handler - The asynchronous event handler.

Threading:

This call is safe for access by multiple threads and can be accessed while the engine is in any state.

setMessageFilter

public final void setMessageFilter(MessageViewFilter messageFilter)

Sets a MessageFilter for live messages received by an AepEngine operating in the primary role.

A message filter does not apply to messages replicated by a peer or recovered from disk. Filtered messages are acknowledged, and are not logged to a transaction logs.

Effects of a MessageFilter:
Filtered messages:

• Are acknowledged.
• Are not logged to the transaction log, inbound message log or replicated to cluster peers.
• Are not dispatched to application EventHandlers.
• Do not start new transactions or contribute to adaptive batching counts.
• Contribute to event received/processed and message received stats, but are accounted for in numMsgsFiltered. They do not undergo duplicate checking so filtered duplicates will not be accounted for duplicate counts.

This call replaces a currently registered MessageViewFilter.

Exceptions thrown from a MessageFilter:
If an exception is thrown by a MessageFilter, it is handled according to the AepEngine.AppExceptionHandlingPolicy for this engine. If the policy is AepEngine.AppExceptionHandlingPolicy.LogExceptionAndContinue, the message is not dispatched to application handlers and is acknowledged.

Parameters:

messageFilter - The message filter.

Threading:

This call is safe for access by multiple threads.

start

@Deprecated
public final void start(Map<AepEngine.StartupExpectation,Enum<?>> expectations)
                 throws EAepException

Deprecated. This method has been deprecated. Startup expectations should be set in the engine's descriptor rather than supplied to this start method

Start an AEP engine with set of expectations.

This method starts an engine with the specified set of startup expectations. The specified set of expectations overrides the expectations in the engine's descriptor

Throws:

EAepException

start

public final void start()
                 throws EAepException

Start an AEP engine.

This method starts an AEP engine. This method needs to be invoked before the engine can be used. The method first opens the engine's store (if the engine is configured to operate in clustered mode) followed by starting the messaging machinery (opening the configured bus bindings) if elected as the cluster primary or if operating in non-clustered mode. If elected as a cluster backup, the method returns immediately and messaging is started when the engine switches to primary.

Throws:

EAepException - Thrown if an error is encountered while starting the engine.

Threading:

This method is not safe for concurrent access by multiple threads with itself or any of the other engine methods.

waitForMessagingToStart

public final void waitForMessagingToStart()
                                   throws EAepException

Wait for messaging to be started.

This method waits for the AEP engine to start the messaging machinery. For cluster primaries or standalone engines, the messaging is started by the start(java.util.Map>) method. For backup engines, messaging is started when the engine becomes the primary. It is ensured that channel up events for bus bindings that could be successfully established at the time messaging is started are dispatched before this method returns.

This method can only be invoked on a started engine

Throws:

EAepException

getState

public final AepEngine.State getState()

Get an engine's state.

getMessagingState

public final AepEngine.MessagingState getMessagingState()

Get an engine's messaging state.

getStore

public final IStoreBinding getStore()

Get an engine's operational store.

This method returns the engine's binding to the operational store configured for use by the engine. A return value of null indicates that no operational store has been configured for use.

This method is permitted on an engine in any state. However, note that a store is only opened when an engine is started. Therefore, one should always be aware of the store and engine's state when working with a store obtained via this method.

Threading:

This method is safe for concurrent access by multiple threads.

getBusManagers

public final Collection<AepBusManager> getBusManagers()

Get an engine's bus managers.

This method returns the set of bus managers created by an engine as configured for use by the engine. An empty collection indicates that the engine has not been configured to connect to any messaging bus

This method is permitted on an engine in any state. However, note that bus managers are only opened when an engine goes active i.e. becomes the primary in a cluster. Therefore, one should always be aware of the engine and bus manager's state when working with bus managers obtained via this method.

Returns:

Returns an engine's bus managers.

Threading:

This method is safe for concurrent access by multiple threads.

getEventMultiplexer

public final IEventMultiplexer getEventMultiplexer()

Get an engine's event multiplexer.

This method returns the inbound event multiplexer used by an engine.

This method is permitted only on a primary engine in the started state. An exception is thrown if invoked on a backup or in any other state.

Threading:

This method is safe for concurrent access by multiple threads.

isDispatchThread

public final boolean isDispatchThread()

Tests if the current thread is currently dispatching an AepEngine event.

Depending on the EventMultiplexer type and other factors the dispatch thread may change from one event to another.

For a given thread this will only return true when called from that thread if the AepEngine is dispatching an event to the application at the time of the call, which while in most cases will be the case

Returns:

True if the current thread is currently dispatching an event for this AepEngine.

injectMessage

public void injectMessage(IRogMessage message)

Enqueue a message into this AepEngine's event multiplexer.

This method is the same as injectMessage(message, false, 0).

Parameters:

message - The message to enqueue.

Throws:

IllegalStateException - Thrown if the method is invoked in any state other than AepEngine.State.Started or AepEngine.State.Starting.

Threading:

This method is safe for concurrent access by multiple threads.

injectMessage

public void injectMessage(IRogMessage message,
                          boolean nonBlocking)

Enqueue a message into this AepEngine's event multiplexer.

This method is the same as injectMessage(message, nonBlocking, 0).

Parameters:

message - The message to enqueue.

nonBlocking - Indicates whether the multiplexing should be a non-blocking action or not. If blocking, then the calling thread will block if the main multiplexer queue is full and wait until space is available. If non-blocking, then the method will not wait but rather enque the message in an unbounded multiplexer feeder queue and return.

Throws:

IllegalStateException - Thrown if the method is invoked in any state other than AepEngine.State.Started or AepEngine.State.Starting.

Threading:

This method is safe for concurrent access by multiple threads.

injectMessage

public final void injectMessage(IRogMessage message,
                                boolean nonBlocking,
                                int delay)

Enqueue a message into this AepEngine's event multiplexer.

This method enqueues a message for execution on this engine's event multiplexer event loop and returns to the caller. The message is then dispatched subsequently using the normal message dispatch mechanisms like a message received from a message bus binding.

Injection from an Event/Message Handler

Important: Injection of messages from an event handler does not result in the Message being added to the current transaction. Consequently a failure of the engine will prior to it being dispatched will result in the message being discarded. See HA Considerations below.

Delayed or Priority Injection

The value of the delay parameter is interpreted as follows:

• A value of 0 causes the message to be added to the end of the underlying engine's event multiplexer queue for immediate dispatch after already enqueued events.
• A positive value is interpreted in milliseconds and will cause the execution of the event to be scheduled no sooner than the delay period. When 2 events are scheduled by the same thread with a positive delay that results in the same scheduled execution time, then the events are executed in the order they were added. E.g. if E1 is injected at time T with delay of 2ms and and E2 injected at time T + 1ms with a delay of 1ms, E2 will be scheduled after E1 (presuming a clock precision with a resolution of 1ms for System.currentTimeMillis()).
• A negative value is treated as a high priority dispatch and such messages are added to the beginning of the engine's event multiplexer queue, a lower delay represents a higher priority dispatch. When 2 events are scheduled with same negative delay the latter scheduled event will be executed after the former.
  NOTE: Priorities less than -1000 are reserved for platform use.

The differences between this method and scheduleMessage(IRogMessage) are:

• Injecting a message is currently less resilient in the event of a failure while the message is in the injection queue. On the other hand injection is lower overhead and more efficient.
• When injecting a message with a delay of 0 the behavior is the same except a message scheduled with a delay of 0 is scheduled at the head of the multiplexer queue and injecting a message enqueues the message at the tail of the queue. Otherwise, for non 0 delay values the message is placed in the front of the queue and is subject to multiplexer scheduling.
• Injection can be done from any thread while message scheduling is only allowed from an event handler called by an engine thread.

Behavior when not in HA Active Role

The message will only be injected on Started engines operating in the Primary role. Calls to inject are ignored on backup instances as the expectation is that the injected message will be replicated from the primary. Calls made while an engine is replaying from a transaction log (e.g. AepEngine.State.Starting) are similarly ignored as they would interfere with the stream being replayed. An application that injects messages from an external source may call waitForMessagingToStart() to avoid an injected message being discarded while an engine is transitioning to a started, primary role. It is important to note that message injection is thus effectively a BestEffort operation because injections of message that are in the event multiplexer queue at the time of failure will be lost. Reliable injection can be achived via scheduleMessage(IRogMessage, int, com.neeve.aep.AepScheduleEvent.HAPolicy) though fault tolerant scheduling incurs slight higher overhead.

Message Pooling Considerations

This method transfers ownership of the message to the platform, the method caller must not modify the message subsequent to this call. The platform will call IRogNode.dispose() on the message once it has been dispatched, so an application must call MessageView.acquire() if it will hold on to a (read-only) reference to the message.

Blocking vs. Non Blocking Considerations

Care must be taken when considering whether to use using blocking or non blocking injection. If the injecting thread is injecting to a engine multiplexer that may itself block on a resource held by the thread trying to inject, is can cause a deadlock. Conversely, using non blocking dispatch can result in excessive memory growth, increased latency and fairness issues, so if the injecting thread is drawing events from an external source, blocking dispatch is generally the right choice.

Parameters:

message - The IRogMessage to enqueue.

nonBlocking - Indicates whether the multiplexing should be a non-blocking action or not. If blocking, then the calling thread will block if the main multiplexer queue is full and wait until space is available. If non-blocking, then the method will not wait but rather enque the message in a multiplexer feeder queue and return. Note that if this method is called from the engine's dispatch thread, this parameter is ignored and non blocking enqueue is used implicitly.

delay - The delay in milliseconds at which the message should be injected.

Throws:

IllegalStateException - Thrown if the method is invoked in any state other than AepEngine.State.Started or AepEngine.State.Starting.

Threading:

This method is safe for concurrent access by multiple threads.

multiplexMessage

@Deprecated
public final void multiplexMessage(IRogMessage message,
                                              boolean nonBlocking)

Deprecated. This method is deprecated by injectMessage(IRogMessage, boolean)

multiplexMessage

@Deprecated
public final void multiplexMessage(IRogMessage message)

Deprecated. This method is deprecated by injectMessage(IRogMessage)

scheduleMessage

public final AepScheduleEvent scheduleMessage(IRogMessage message)

Schedule a message for immediate dispatch.

This method is equivalent to invoking scheduleMessage(message, 0, HAPolicy.Cancel).

Parameters:

message - The message to dispatch.

Throws:

IllegalArgumentException - Thrown if the supplied message is null.

IllegalStateException - Thrown if the method is invoked in any state other than AepEngine.State.Starting or AepEngine.State.Started, from outside of a message, 'messaging started' or schedule event handler or by a thread other than the one that that invoked the message/event handler.

Threading:

This method can only be invoked from a message handler, from the 'messaging started' event handler or a schedule event handler. Also, it must be incoked by the thread that performed the handler dispatch.

scheduleMessage

public final AepScheduleEvent scheduleMessage(IRogMessage message,
                                              int delay)

Schedule a message for a (one-time) delayed dispatch with the default schedule HA policy.

This method is equivalent to invoking scheduleMessage(message, delay, HAPolicy.Cancel).

Parameters:

message - The message to dispatch.

delay - The delay, in milliseconds, after which to dispatch the message. A time of <= 0 is interpreted as a request for immediate dispatch. In this case, this parameter is interpreted as an event priority in a manner that treats lower numbers as higher priorities.

Throws:

IllegalArgumentException - Thrown if the supplied message is null.

IllegalStateException - Thrown if the method is invoked in any state other than AepEngine.State.Starting or AepEngine.State.Started, from outside of a message, 'messaging started' or schedule event handler or by a thread other than the one that that invoked the message/event handler.

Threading:

This method can only be invoked from a message handler, from the 'messaging started' event handler or a schedule event handler. Also, it must be incoked by the thread that performed the handler dispatch.

scheduleMessage

public final AepScheduleEvent scheduleMessage(IRogMessage message,
                                              int delay,
                                              AepScheduleEvent.HAPolicy haPolicy)

Schedule a message for (one-time) delayed dispatch.

This method schedules a message with the engine's event multiplexer for dispatch after a specified delay. The delay is expressed in milliseconds. A delay of > 0 is interpreted as a request for delayed dispatch while a value of <= 0 is interpeted as immediate dispatch. The method instantiates a new schedule, registers the schedule with the engine's event multiplexer and returns a handle to the schedule in the form of AepScheduleEvent. The scheduled message is dispatched to the application when the schedule fires. When dispatched, the schedule event is tagged to the message and can be retrieved from the message as follows:
message.getTag(MessageViewTags.TAG_AEP_SCHEDULE_EVENT)

For EventSourced engines, the act of scheduling of and dispatch of the scheduled message is replicated to the backup. Therefore, the during steady state the scheduled message will be dispatched to the application in the same order relative to other messages. Also, activated scheduled persist failover and restart. On failover or restart, schedules are automatically reactivated.

For StateReplication engines, the act of scheduling and dispatch of schedule messages is not persisted or replicated. Therefore, the application needs to reschedule the messages on failover/restart.

This method takes ownership of the scheduled message. This means that the caller should relinquish control of the message until the message is dispatched to the application. Once dispatched, the ownership is transferred to the application.

Note: Upon completion of the message dispatch to the application or on cancellation of the schedule, the engine disposes the schedule handle i.e. AepScheduleEvent. When event pooling is enabled, the act of disposing the event will clear out its contents. Therefore, if the application intends to refer to the contents of the event post dispatch/cancel, the application must acquire a reference to the event via Event.acquire(). The acquired reference must be disposed via Event.dispose() when the application is done referencing the event.

Parameters:

message - The message to dispatch.

delay - The delay, in milliseconds, after which to dispatch the message. A time of <= 0 is interpreted as a request for immediate dispatch. In this case, this parameter is interpreted as an event priority in a manner that treats lower numbers as higher priorities.

haPolicy - Specifies the schedule's HA policy.

Returns:

A schedule event object. The returned object serves as a handle to the schedule managing the delayed dispatch.

Throws:

IllegalArgumentException - Thrown if the supplied message is null.

IllegalStateException - Thrown if the method is invoked in any state other than AepEngine.State.Starting or AepEngine.State.Started, from outside of a message, 'messaging started' or schedule event handler or by a thread other than the one that that invoked the message/event handler.

Threading:

This method can only be invoked from a message handler, from the 'messaging started' event handler or a schedule event handler. Also, it must be incoked by the thread that performed the handler dispatch.

scheduleObject

public final AepScheduleEvent scheduleObject(Object object)

Schedule an opaque object for immediate dispatch.

This method is equivalent to invoking scheduleObject(message, 0, HAPolicy.Cancel).

Parameters:

object - Specifies the schedule's HA policy.

Returns:

A schedule event object. The returned object serves as a handle to the schedule managing the delayed dispatch.

Throws:

IllegalArgumentException - Thrown if the supplied message is null.

IllegalStateException - Thrown if the method is invoked in any state other than AepEngine.State.Starting or AepEngine.State.Started, from outside of a message, 'messaging started' or schedule event handler or by a thread other than the one that that invoked the message/event handler.

Threading:

This method can only be invoked from a message handler, from the 'messaging started' event handler or a schedule event handler. Also, it must be incoked by the thread that performed the handler dispatch.

scheduleObject

public final AepScheduleEvent scheduleObject(Object object,
                                             int interval)

Schedule an opaque object for periodic dispatch with the default schedule HA policy.

This method is equivalent to invoking scheduleObject(message, delay, HAPolicy.Cancel).

Parameters:

object - Specifies the schedule's HA policy.

interval - The interval (in milliseconds) at which to dispatch the object. A value <= 0 is interpreted as a request for one-time immediate dispatch. In this case, this parameter is interpreted as an event priority in a manner that treats lower numbers as higher priorities.

Returns:

A schedule event object. The returned object serves as a handle to the schedule managing the delayed dispatch.

Throws:

IllegalArgumentException - Thrown if the supplied message is null.

IllegalStateException - Thrown if the method is invoked in any state other than AepEngine.State.Starting or AepEngine.State.Started, from outside of a message, 'messaging started' or schedule event handler or by a thread other than the one that that invoked the message/event handler.

Threading:

This method can only be invoked from a message handler, from the 'messaging started' event handler or a schedule event handler. Also, it must be incoked by the thread that performed the handler dispatch.

scheduleObject

public final AepScheduleEvent scheduleObject(Object object,
                                             int interval,
                                             AepScheduleEvent.HAPolicy haPolicy)

Schedule an opaque object for periodic dispatch at a specified interval.

This method schedules an object with the engine's event multiplexer for periodic dispatch at the specified interval. The interval is expressed in milliseconds. A delay of > 0 is interpreted as a request for delayed dispatch while a value of <= 0 is interpeted as immediate dispatch. The method instantiates a new schedule, registers the schedule with the engine's event multiplexer and returns a handle to the schedule in the form of AepScheduleEvent. The returned event is dispatched to the application each time the schedule fires. The provided object is encapsulated within the dispatched event.

For EventSourced engines, the act of scheduling of and dispatch of the scheduled object is replicated to the backup. Therefore, during steady state the scheduled object will be dispatched to the application in the same order relative to other messages and events. Also, schedules persist failover and restart i.e. on failover or restart, the schedules are automatically reactivated.

For StateReplication engines, the act of scheduling and dispatch of schedule objects is not persisted or replicated. Therefore, the application needs to reschedule the objects on failover/restart.

Note: Upon completion of the object dispatch to the application for an imediate dispatch schedule or on cancellation of a periodic dispatch schedule, the engine disposes the schedule handle i.e. AepScheduleEvent. When event pooling is enabled, the act of disposing the event will clear out its contents. Therefore, if the application intends to refer to the contents of the event post dispatch/cancel, the application must acquire a reference to the event via Event.acquire(). The acquired reference must be disposed via Event.dispose() when the application is done referencing the event.

Parameters:

interval - The interval (in milliseconds) at which to dispatch the object. A value <= 0 is interpreted as a request for one-time immediate dispatch. In this case, this parameter is interpreted as an event priority in a manner that treats lower numbers as higher priorities.

object - The object to dispatch.

haPolicy - Specifies the schedule's HA policy.

Returns:

A schedule event object. The returned object serves as a handle to the schedule managing the delayed dispatch.

Throws:

IllegalStateException - Thrown if the method is invoked in any state other than AepEngine.State.Starting or AepEngine.State.Started, from outside of a message, 'messaging started' or schedule event handler or by a thread other than the one that that invoked the message/event handler.

Threading:

This method can only be invoked from a message handler, from the 'messaging started' event handler or a schedule event handler. Also, it must be incoked by the thread that performed the handler dispatch.

cancelSchedule

public final void cancelSchedule(AepScheduleEvent event)

Cancel the dispatch of a scheduled event.

This method cancels the dispatch of a schedule event. If the event is scheduled for perdiodic dispatch, then this method cancels all future dispatch. If scheduled for a single dispatch, then the dispatch is cancelled. The caller is guaranteed that the schedule event will not be dispatched to the caller after successful return from this method.

This method can be safely invoked multiple times.

Parameters:

event - The schedule to cancel.

Throws:

IllegalStateException - Thrown if the method is invoked in any state other than AepEngine.State.Starting or AepEngine.State.Started, from outside of a message, 'messaging started' or schedule event handler or by a thread other than the one that that invoked the message/event handler.

Threading:

This method can only be invoked from a message handler, from the 'messaging started' event handler or a schedule event handler. Also, it must be incoked by the thread that performed the handler dispatch.

getInboundMessageLogger

public final IRogMessageLogger getInboundMessageLogger()

Get an engine's inbound message logger.

This method returns the inbound message logger used by an engine.

An engine opens it's message loggers when engine is started. Therefore, this method will return null prior to a successful engine start.

The contents of an engine's inbound message log are managed very carefully by the engine. Logging messages using this method is very advanced use and should only be used if absolutely necessary and the caller has deep knowledge of how the engine manages entries in the log.

Threading:

This method is safe for concurrent access by multiple threads.

getOutboundMessageLogger

public final IRogMessageLogger getOutboundMessageLogger()

Get an engine's outbound message logger.

This method returns the outbound message logger used by an engine.

An engine opens it's message loggers when engine is started. Therefore, this method will return null prior to a successful engine start.

The contents of an engine's outbound message log are managed very carefully by the engine particularly when the engine is configured to use the inbound message logger for outbound messages too. Logging messages using this method is very advanced use and should only be used if absolutely necessary and the caller has deep knowledge of how the engine manages entries in the log.

Threading:

This method is safe for concurrent access by multiple threads.

isPrimary

public final boolean isPrimary()

Get if the engine is running as primary or secondary.

The method will return true if operating in a non-clustered (standalone) mode. To differentiate between a primary in a cluster or running in standalone mode, use isClustered()

This method is permitted only in any state except init or stopped state. An exception is thrown if invoked in any other state.

Returns:

true if the engine is the current primary.

Throws:

IllegalStateException - if the the engine is not in Starting, Started or Stopping state.

getServerBusBinding

public final MessageBusBinding getServerBusBinding(String name)

Get an application's 'server' message bus binding.

An engine starts its messaging when it becomes the primary. Therefore, this method will return null prior to being started and being elected primary.

Threading:

This method is safe for concurrent access by multiple threads.

getStats

public final IAepEngineStats getStats()

Get the set of a engine's operational statistics.

Threading:

This method is safe for concurrent access by multiple threads.

getApplicationState

public final <T> T getApplicationState(int flowid)

Get the application state associated with a flow.

This method returns the application state object associated with a flow. It can only be invoked on a started engine. Given this state constraint, the threading contraints and the how application state is created/managed by the engine, this method should only be invoked from the AepMessagingPrestartEvent or the AepMessagingStartedEvent event handler. These are the only point in the application lifecycle when the engine is considered to be started, the message/event dispatch machinery is halted and the state created implcitly by this method will not conflict with state created implicitly by replication.

Parameters:

flowid - The flow id.

Throws:

IllegalStateException - Thrown if invoked after messaging has been started.

Threading:

This method is notsafe for concurrent access by multiple threads with itself, with any of the other engine methods or with the event/message dispatch machinery.

getApplicationState

public final <T> T getApplicationState(MessageView message)

Get the application state associated with an inbound message.

This method returns the application state object associated with the flow on which an inbound message was received.

Parameters:

message - The inbound message.

Threading:

This method is safe for concurrent access by multiple threads with itself, with any of the other engine methods and with the event dispatch machinery.

getEngineTime

public final long getEngineTime()

Get current engine time.

This method is intended for use by applications using event sourcing that do time dependent message processing. This method provides the current wall time (in millisecond resolution) as perceived by the engine. If invoked from within a message processor handler and the HA policy is set to event sourcing, this method returns the time stamped on the message event (stamped just before the method is dispatched to the application for processing). Since, for event sourced applications, the message is also replicated for parallel processing on the backup, the backup will receive the same time when invoking this method thus ensuring identical processing. If this method is called on an engine operating in state replication mode or called from outside a message processor, then this method will return the value returned by the System.currentTimeMillis() method

Threading:

This method is safe for concurrent access by multiple threads with itself, with any of the other engine methods and with the event dispatch machinery.

resumeCommit

public final void resumeCommit(AepTransactionStageEvent.ICommitContext context)

Resume a suspended transaction commit.

This method should be invoked by the application to resume a transaction commit suspended via the AepTransactionStageEvent. When the transaction needs to be resumed, the application should call this method supplying the transaction commit context present in the event used to suspend the transaction.

Parameters:

context - The context of the suspended commit.

sendMessage

public final void sendMessage(MessageChannel channel,
                              IRogMessage message,
                              Properties keyResolutionTable)

Send a message.

This method can only be invoked on a started engine.

NOTE: It is not safe for both message handlers and unsolicited sender threads to call this method, the only case where a thread that is not the dispatch thread may call this method is if the application is a purely sending application that will never send a message from a message or event handler.

Parameters:

channel - The channel on which to send the message.

message - The message to send.

keyResolutionTable - The message's key resolution table

Threading:

This method is not safe for concurrent access by multiple threads with itself or any of the other engine methods.

sendMessage

public final void sendMessage(MessageChannel channel,
                              IRogMessage message,
                              MessageChannel.RawKeyResolutionTable keyResolutionTable)

Send a message using a XString key resolution table. This method is the same sendMessage(MessageChannel, IRogMessage, Properties), but performs key resolution using XString logic that doesn't produce garbage.

This method can only be invoked on a started engine.

A MessageChannel.RawKeyResolutionTable instance can be obtained by calling MessageBusBindingFactory.createRawKeyResolutionTable().

Parameters:

channel - The channel on which to send the message.

message - The message to send.

keyResolutionTable - The message's key resolution table.

Threading:

This method is not safe for concurrent access by multiple threads with itself or any of the other engine methods.

sendMessage

public final void sendMessage(MessageChannel channel,
                              IRogMessage message,
                              String key,
                              Properties keyResolutionTable)

Send an outbound message.

This method sets the key in the message and invokes the following:
sendMessage(MessageChannel, IRogMessage, Properties)

Parameters:

channel - The channel on which to send the message.

message - The message to send.

key - The message key.

keyResolutionTable - The message's key resolution table (ignored if the supplied key is non-null)

Threading:

This method is not safe for concurrent access by multiple threads with itself or any of the other engine methods.

sendMessage

public final void sendMessage(MessageChannel channel,
                              IRogMessage message,
                              XString key,
                              MessageChannel.RawKeyResolutionTable keyResolutionTable)

Send an outbound message. This method has the same semantics as sendMessage(MessageChannel, IRogMessage, String, Properties), but uses a XString to avoid creation of garbage.

Parameters:

channel - The channel on which to send the message.

message - The message to send.

key - The message key.

keyResolutionTable - The message's key resolution table (ignored if the supplied key is non-null)

Threading:

This method is not safe for concurrent access by multiple threads with itself or any of the other engine methods.

sendMessage

public final void sendMessage(MessageChannel channel,
                              IRogMessage message)

Send an outbound message.

This method invokes sendMessage(MessageChannel, IRogMessage, Properties)

Parameters:

channel - The channel on which to send the message.

message - The message to send.

Threading:

This method is not safe for concurrent access by multiple threads with itself or any of the other engine methods.

replyToClient

public final void replyToClient(MessageChannel channel,
                                PktPacket request,
                                PktPacket reply,
                                int replyCount,
                                int replyNum)
                         throws Exception

Send a success reply to a client request.

This method is a convenience method to send a reply to a request packet received from an AEP application client. The reply sent by this method indicates successful request processing.

Parameters:

channel - The client's messaging channel.

request - The request for which to the send the reply.

reply - The reply packet to send. A null value will cause this method to create and send a generic reply packet.

replyCount - The number of replies for this request.

replyNum - The reply number.

Throws:

Exception

Threading:

This method is not safe for concurrent access by multiple threads with itself or any of the other engine methods.

replyToClientWithError

public final void replyToClientWithError(MessageChannel channel,
                                         PktPacket request,
                                         PktPacket reply,
                                         long errorCode)
                                  throws Exception

Send an error reply to a client request.

This method is a convenience method to send a reply to a request packet received from an AEP application client. The reply sent by this method indicates a failed request processing.

Parameters:

channel - The client's messaging channel.

request - The request for which to the send the reply.

reply - The reply packet to send. A null value will cause this method to create and send a generic reply packet.

errorCode - The error code.

Throws:

Exception

Threading:

This method is not safe for concurrent access by multiple threads with itself or any of the other engine methods.

createTransactionSavepoint

public final int createTransactionSavepoint()

Creates a savepoint in the current transaction.

A savepoint serves as a delimiting marker that can used to partially rollback changes within the current transaction. A subsequent call to rollbackToTransactionSavepoint(int) with the returned savepoint will roll back state changes and outbound messages done subsequent to the savepoint.

The Restrictions outlined in rollbackToTransactionSavepoint(int) apply to this method. If any of the criteria listed in the those restrictions is not met, this method with throw an IllegalStateException

Usage of savepoints when operating with an AepEngine.HAPolicy of EventSourcing is currently classified as an expiremental feature.

Returns:

The new savepoint.

Throws:

IllegalStateException - Thrown if the the engine is not in a state that supports save point operations.

See Also:

rollbackToTransactionSavepoint(int)

getTransactionSavepoint

public final int getTransactionSavepoint()

Get the current savepoint for the transaction.

The method returns the current savepoint for a transaction in progress.

The Restrictions outlined in rollbackToTransactionSavepoint(int) apply to this method. If any of the criteria listed in the those restrictions is not met, this method with throw an IllegalStateException

Returns:

the current savepoint.

Throws:

IllegalStateException - Thrown if the the engine is not in a state that supports savepoint operations.

See Also:

rollbackToTransactionSavepoint(int)

rollbackToTransactionSavepoint

public final void rollbackToTransactionSavepoint(int savepoint)

Rolls back state changes and outbound messages to a given savepoint.

This method rolls back changes made to application state since the given savepoint. Additionally it rolls back any outbound messages sent as part of the transaction since the save point.

The rollback operation leaves the provided savepoint marker in place. For example, if the application calls rollback with a savepoint value of 1, a subsequent call to getTransactionSavepoint() will return 1. Additional work done after the rollback can thus be rolled back to the same point.

Any savepoints after the provided savepoint are discarded. If rollback is called with a savepoint value of 1 when getTransactionSavepoint() is at 2, savepoint 2 is discarded.

Note that rolling back to the initial savepoint (0) does not rollback the entire transaction. Instead, the processed message remains in the transaction stream as a successfully event with no effects.

Restrictions

Transaction savepoints are only supported when:

• the engine is configured with savepoints enabled AepEngineDescriptor#getEnableTransactionSavepoints() == true
• the engine is backed by a store (e.g. getStore() != null as the AepEngine relies on the store's transaction machinery to perform rollback. This restriction may be relaxed in a future release.
• Only application message handlers called from the engine's message processing thread may work with savepoints. isDispatchThread() && #
• the engine is not configured for parallel cluster replication AepEngineDescriptor.getReplicateInParallel() == false

If any of the above criteria is not met an IllegalStateException is thrown. In addition to the above restrictions an application must not use the following savepoint operations in the underlying store:

• IStoreBinding.createSavepoint()
• IStoreBinding.rollback(int)
• IStoreBinding.rollback()
• IStoreBinding.getLastSavepoint()

Adaptive Batching Considerations

When the engine is configured for adapative batching, multiple inbound events (messages) are grouped into a single store transaction. Savepoints don't span multiple events. Instead each event is currently processed as an atomic unit of work, and applications can be assured that the rollback won't affect work that was done for prior events. For each message processed in the batch the savepoint is reset.

Multiple event handlers

If there are multiple event handlers for a given event, savepoints do span those handlers meaning that a subsequent handler for an event can rollback work done by a previous handler. Applications may create a savepoint via createTransactionSavepoint() at the beginning of message processing to avoid rolling back work done by another handler.

HA considerations

Savepoint operations can only be performed in a message handler. When using StateReplication this means that savepoint creation and rollbacks can only be done on a Primary instance.

Usage of savepoints when operating with an AepEngine.HAPolicy of EventSourcing is currently classified as an expiremental feature. Because application state is not managed by the platform for EventSourcing, only an applications outbound messages are rolled back by this method.

When using EventSourcing, message handlers are invoked on a Backup instance or an instance recovering from a transaction log. It is crucial that a backup or recovering instance's behavior or it will lead to divergence in the application's outbound messages. This means that application logic on a backup must create the same savepoints as a primary and rollback based on the same criteria. For this reason it is often preferable for an EventSourcing app that encounters an error to simply throw an exception from its event handler and let the inbound message's fate be governed by the AepEngine.AppExceptionHandlingPolicy.

Rollback Errors

An EAepRollbackError thrown from this method indicates that therd was an internal or runtime error performing the rollback. In this case, application event handlers must allow the error to be thrown back for the platform to handle as the application state may be in a corrupt state. If the AepEngine can recover by rolling back the entire transaction, the error will be handled according to the AepEngine.AppExceptionHandlingPolicy, otherwise the engine will stop with the EAepRollbackError.

Outbound Message Considerations

Outbound messages that are rolled back cannot be reused - the transfer of ownership to the AepEngine is preserved. Additionally, it should be noted that rollback does not rollback changes made to outbound messages' fields. When the engine is configured to dispose on send the engine may dispose of such messages during rollback, so applications should not rely on messages tranfered to the engine being valid post rollback.

State and Embedded Entity Considerations

Objects that were created since the savepoint that is rolled back may be discarded and cleared by the engine during rollback. Therefore applications should not attempt to reuse any state objects created since the savepoint that was rolled back.

Parameters:

savepoint - The savepoint to roll back to.

Throws:

IllegalStateException - Thrown if the the engine is not in a state that supports savepoint operations.

IllegalArgumentException - if the savepoint is greater than the current savepoint or less than 0.

EAepRollbackError - if the rollback failed - an application should not trap this exception.

See Also:

createTransactionSavepoint()

Threading:

This method may only be called from an Event handler processing a message.

setAsLastTransaction

public final void setAsLastTransaction(Exception cause,
                                       boolean ignoreIfStarting,
                                       boolean lastClusterTransaction)

Set current transaction as the last transaction.

The method (1) marks the current transaction as the last transaction and (2) marks the message from whose handler the method is being invoked as the last message in the transaction. Once the current transaction is marked as the last transaction, all subsequent inbound messages are ignored (but ensured that they are not acknowledged). Once the last transaction is complete, the engine will shutdown (stop(Exception)) using same cause object supplied to this method.

This method can only be invoked from within a message handler.

This method has no effect on backup agents unless lastClusterTransaction is set in which case backup agents shutdown too.

Parameters:

cause - The cause of why last transaction is being triggered. A value of null indicates 'normal' shutdown

ignoreIfStarting - Normally, this method would throw an illegal state exception if invoked while the engine is starting. However, there are cases where special messages may be used to trigger a graceful shutdown. Since those messages may be logged and can be replayed during initialization with event sourcing, the same behavior may not be desirable during engine initialization. With this flag set, the engine will just return from this method if the engine is starting rather than throw an exception.

lastClusterTransaction - This parameter indicates whether the current transaction should be treated as the last transaction for not only the running agent but for the entire cluster. When operating in Event Sourcing configuration, a value of true for this parameter will cause this method call to be processed on all members of the cluster (note that, with Event Sourcing, all cluster members process inbound messages). A value of false, on the other hand, will cause this method invocation to be treated as a no-op on backup members. Last cluster transaction functionality is not currently supported for State Replication. A value of true specified in this parameter with State Replication will cause an IllegalArgumentException to be thrown.

Throws:

IllegalArgumentException - Thrown if a value of true is specified on an engine operating in the StateReplication HA mode.

IllegalStateException - Thrown if this method is invoked on an engine in any state other than the started state (unless the ignoreIfStarting flag is specified and in starting state) or if invoked from outside of a message handler.

setAsLastTransaction

public final void setAsLastTransaction(Exception cause,
                                       boolean ignoreIfStarting)

Set current transaction as last transction (only for primary agent)

This method invoked setAsLastTransaction(cause, ignoreIfStarting, false)

stop

public final void stop(Exception cause)

Stop an AEP engine.

This method stops and AEP engine. It first flags the engine as 'stopping' to prevent messages/events beyond the current one from being processed. It then schedules a 'stop' event inline with its inbound event/message stream to stop the engine. If the method is called by the thread actively dispatching the current engine event, then the call immediately returns. Otherwise, the call will block and wait for the stop event to be processed. Note that this implies that for stops performed by the engine dispatching thread, the caller cannot assume that the engine has been stopped upon return from the method. However, this can be assumed if this method is invoked by a thread different from he current dispatching thread.

Parameters:

cause - The reason why the engine is being stopped. A null value indicates a 'normal' stop. A 'normal' stop results in the engine 'leaving' all joined channels. An 'abnormal' stop will not explicitly 'leave' its joined channels. When used with transports that support durable subscription, this means that a 'normal' stop will remove the durable subscriptions while an 'abnormal' stop will leave the durable subscriptions intact. To preserve joined channels on a 'normal' stop, an event handler for AepEngineStoppingEvent can change the value of AepEngineStoppingEvent.setPreserveChannelJoins(boolean).

Threading:

This method is not safe for concurrent access by multiple threads with itself or any of the other engine methods.

stop

public final void stop()

Stop an AEP engine (normal).

This method invokes stop(null)

getName

public final String getName()

Implementation of IEventSource.getName()

Specified by:

getName in interface IEventSource

onEvent

public final void onEvent(Event event)

Implementation of IEventHandler.onEvent(com.neeve.event.Event)

This method is the engine multiplexer's event handler, it should never be invoked directly by the application.

Specified by:

onEvent in interface IEventHandler

Parameters:

event - The event to be handled.

onEvent

public final void onEvent(IStoreEvent event)

Implementation of IStoreEventHandler.onEvent(com.neeve.ods.IStoreEvent)

This method is the store event handler and should never be invoked directly by the application.

Specified by:

onEvent in interface IStoreEventHandler

Parameters:

event - The event to be handled.