com.neeve.aep
Class AepEngine

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

All Implemented Interfaces:

com.neeve.event.IEventHandler, IStoreEventHandler

public final class AepEngine
extends RootObject
implements IStoreEventHandler, com.neeve.event.IEventHandler

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.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 Enumerates an engine's outbound message send policy
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.OutboundMessageLoggingPolicy Enumerates an engine's outbound message 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

Method Summary

void	cancelSchedule(AepScheduleEvent event) Cancel the dispatch of a scheduled event.
static AepEngine	create(AepEngineDescriptor engineDescriptor, IAepApplicationStateFactory stateFactory, Set<Object> eventHandlerContainers, com.neeve.event.IEventHandler defaultEventHandler, IAepWatcher watcher) Create an AEP engine.
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
com.neeve.event.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.MessageSendPolicy	getMessageSendPolicy() Get an engine's message send policy
AepEngine.MessagingStartFailPolicy	getMessagingStartFailPolicy() Get an engine's messaging start fail policy
AepEngine.MessagingState	getMessagingState() Get an engine's messaging state.
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
boolean	isClustered() Get if an engine is running in clustered or non-clustered mode.
boolean	isPrimary() Get if the engine is running as primary or secondary.
void	multiplexMessage(IRogMessage message) Enque a message into an engine's event multiplexer.
void	multiplexMessage(IRogMessage message, boolean nonBlocking) Enque a message into an engine's event multiplexer.
void	onEvent(com.neeve.event.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(com.neeve.stats.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(com.neeve.stats.IStats.Gauge gauge) Registers a user Gauge stat with the engine stats.
void	registerLatencyStat(com.neeve.stats.IStats.Latencies latencies) Registers a user latency stat with the engine stats.
AepEngine	registerMessageFactory(MessageViewFactory factory) Register a message factory.
AepEngine	registerObjectFactory(IStoreObjectFactory factory) Register an object factory
AepEngine	registerTypeFactory(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.
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, Properties keyResolutionTable) Send a message.
void	sendMessage(MessageChannel channel, IRogMessage message, String key, Properties keyResolutionTable) Send an outbound message.
void	sendMessageWithRawKey(MessageChannel channel, IRogMessage message, com.neeve.raw.RawString key, MessageChannel.RawKeyResolutionTable rawKeyResolutionTable) Send an outbound message.
void	sendMessageWithRawKRT(MessageChannel channel, IRogMessage message, MessageChannel.RawKeyResolutionTable rawKeyResolutionTable) Send a message using a RawString key resolution table.
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(IAepCentralMessageHandler handler) Register the central message handler
void	setInboundMessageLogger(IRogMessageLogger logger) Set an engine's inbound message logger.
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
void	setMessageSendPolicy(AepEngine.MessageSendPolicy policy) Set an engine's message send policy
void	setOutboundMessageLogger(IRogMessageLogger logger) Set an engine's outbound message logger.
void	start() Start an engine without any expectations.
void	start(Map<AepEngine.StartupExpectation,Enum<?>> expectations) Start an AEP engine.
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

Method Detail

create

public static final AepEngine create(AepEngineDescriptor engineDescriptor,
                                     IAepApplicationStateFactory stateFactory,
                                     Set<Object> eventHandlerContainers,
                                     com.neeve.event.IEventHandler defaultEventHandler,
                                     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.

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.

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.

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

Parameters:

policy - The message handling policy to set.

This method can be invoked in any state.

getMessageHandlingPolicy

public final AepEngine.MessageHandlingPolicy getMessageHandlingPolicy()

Get an engine's message handling policy

setMessageSendPolicy

public final void setMessageSendPolicy(AepEngine.MessageSendPolicy policy)

Set an engine's message send policy

Parameters:

policy - The message send policy to set.

This method can be invoked in any state.

getMessageSendPolicy

public final AepEngine.MessageSendPolicy getMessageSendPolicy()

Get an engine's message send 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(StoTypeFactory factory)

Register a type factory.

Parameters:

factory - The type factory to register.

Returns:

Returns this object for invocation chaining.

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.

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.

Parameters:

provider - The provider 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 registers an environment provider with an AEP engine.

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

registerLatencyStat

public final void registerLatencyStat(com.neeve.stats.IStats.Latencies latencies)

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

Parameters:

latencies - The stat to add

Throws:

IllegalStateException - if the engine has been started

See Also:

StatsFactory.createLatencyStat(String)

registerCounterStat

public void registerCounterStat(com.neeve.stats.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 void registerGaugeStat(com.neeve.stats.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

public final void setCentralMessageHandler(IAepCentralMessageHandler handler)

Register the central message handler

Parameters:

handler - The handler to register (null to deregister)

This method registers the central message handler. The central 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.

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.

start

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

Start an AEP engine.

Parameters:

expectations - The set of expectations to start the engine with. This method will throw an exception if any of the expectations specified are not met.

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.

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.

start

public final void start()
                 throws EAepException

Start an engine without any expectations.

This method invoked start(null)

Throws:

EAepException

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

Threading:

This method is safe for concurrent access by multiple threads.

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.

getBusManagers

public final Collection<AepBusManager> getBusManagers()

Get an engine's bus managers

Returns:

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

Threading:

This method is safe for concurrent access by multiple threads.

getEventMultiplexer

public final com.neeve.event.IEventMultiplexer getEventMultiplexer()

Get an engine's event multiplexer.

Threading:

This method is safe for concurrent access by multiple threads.

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.

multiplexMessage

public final void multiplexMessage(IRogMessage message,
                                   boolean nonBlocking)

Enque a message into an engine's event multiplexer.

Parameters:

message - The message to enque.

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.

Throws:

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

Threading:

This method is safe for concurrent access by multiple threads.

This method enqueues a message into an engine's event multiplexer event loop and returns to the caller. The message is then dispatched subsequently using the normal message dispatch mechanisms. The difference between this method and scheduleMessage(IRogMessage) is that this method enques the message into end of the multiplexer queue while scheduleMessage(IRogMessage) enqueues the message to the head of the queue with 0 priority (enqueueing using different priorities at the head of the queue can also be accomplished via the scheduleMessage(IRogMessage, int) method.

multiplexMessage

public final void multiplexMessage(IRogMessage message)

Enque a message into an engine's event multiplexer.

This method invokes the following:
multiplexMessage(message, false)

scheduleMessage

public final AepScheduleEvent scheduleMessage(IRogMessage message)

Schedule a message for immediate dispatch.

This method is equivalent to invoking the following:
scheduleMessage(message, 0, AepScheduleEvent.HAPolicy.Cancel)

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 the following:
scheduleMessage(message, delay, AepScheduleEvent.HAPolicy.Cancel)

scheduleMessage

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

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

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.

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.

scheduleObject

public final AepScheduleEvent scheduleObject(Object object)

Schedule an opaque object for immediate dispatch.

This method is equivalent to invoking the following:
scheduleObject(object, 0, AepScheduleEvent.HAPolicy.Cancel)

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 the following:
scheduleObject(object, interval, AepScheduleEvent.HAPolicy.Cancel)

scheduleObject

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

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

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.

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.

cancelSchedule

public final void cancelSchedule(AepScheduleEvent event)

Cancel the dispatch of a scheduled event.

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.

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.

getInboundMessageLogger

public final IRogMessageLogger getInboundMessageLogger()

Get an engine's inbound message logger.

Threading:

This method is safe for concurrent access by multiple threads.

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.

setInboundMessageLogger

public final void setInboundMessageLogger(IRogMessageLogger logger)

Set an engine's inbound message logger.

Threading:

This method is safe for concurrent access by multiple threads.

This method sets the inbound message logger to use. This method is only permitted when used with AepEngine.InboundMessageLoggingPolicy.UserSupplied and before messaging has been started.

getOutboundMessageLogger

public final IRogMessageLogger getOutboundMessageLogger()

Get an engine's outbound message logger.

Threading:

This method is safe for concurrent access by multiple threads.

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.

setOutboundMessageLogger

public final void setOutboundMessageLogger(IRogMessageLogger logger)

Set an engine's outbound message logger.

Threading:

This method is safe for concurrent access by multiple threads.

This method sets the outbound message logger to use. This method is only permitted when used with AepEngine.OutboundMessageLoggingPolicy.UserSupplied and before messaging has been started.

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.

getServerBusBinding

public final MessageBusBinding getServerBusBinding(String name)

Get an application's 'server' message bus binding

Threading:

This method is safe for concurrent access by multiple threads.

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

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.

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.

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.

getApplicationState

public final <T> T getApplicationState(MessageView message)

Get the application state associated with an inbound message.

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.

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

getEngineTime

public final long getEngineTime()

Get current engine time.

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.

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

resumeCommit

public final void resumeCommit(AepTransactionStageEvent.ICommitContext context)

Resume a suspended transaction commit.

Parameters:

context - The context of the suspended 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.

sendMessage

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

Send a message.

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.

This method can only be invoked on a started engine.

sendMessageWithRawKRT

public final void sendMessageWithRawKRT(MessageChannel channel,
                                        IRogMessage message,
                                        MessageChannel.RawKeyResolutionTable rawKeyResolutionTable)

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

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.

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

This method can only be invoked on a started engine.

sendMessage

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

Send an outbound message.

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.

This method sets the key in the message and invokes the following:
sendMessage(channel, message, keyResolutionTable)

sendMessageWithRawKey

public final void sendMessageWithRawKey(MessageChannel channel,
                                        IRogMessage message,
                                        com.neeve.raw.RawString key,
                                        MessageChannel.RawKeyResolutionTable rawKeyResolutionTable)

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

Parameters:

channel - The channel on which to send the message.

message - The message to send.

key - The message key.

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

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.

This method invokes sendMessage(channel, message, null)

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.

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.

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.

replyToClientWithError

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

Send an error reply to a client request.

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.

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.

setAsLastTransaction

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

Set current transaction as the last transaction.

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.

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.

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.

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.

Threading:

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

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.

stop

public final void stop()

Stop an AEP engine (normal).

This method invokes stop(null)

onEvent

public final void onEvent(com.neeve.event.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 com.neeve.event.IEventHandler

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.