org.openspaces.events.polling
Class SimplePollingEventListenerContainer

java.lang.Object
  org.openspaces.events.AbstractSpaceListeningContainer
      org.openspaces.events.AbstractEventListenerContainer
          org.openspaces.events.AbstractTemplateEventListenerContainer
              org.openspaces.events.AbstractTransactionalEventListenerContainer
                  org.openspaces.events.polling.AbstractPollingEventListenerContainer
                      org.openspaces.events.polling.SimplePollingEventListenerContainer

All Implemented Interfaces:

EventListener, ServiceDetailsProvider, ServiceMonitorsProvider, BeanNameAware, DisposableBean, InitializingBean, ApplicationContextAware, ApplicationListener, Lifecycle

public class SimplePollingEventListenerContainer
extends AbstractPollingEventListenerContainer

Event listener container variant that uses plain Space take API, specifically a loop of GigaSpace.take(Object,long) calls that also allow for transactional reception of messages.

Actual event listener execution happens in asynchronous work units which are created through Spring's TaskExecutor abstraction. By default, the specified number of invoker tasks will be created on startup, according to the "concurrentConsumers" setting. Specify an alternative TaskExecutor to integrate with an existing thread pool facility (such as a J2EE server's), for example using a CommonJ WorkManager.

Event reception and listener execution can automatically be wrapped in transactions through passing a Spring PlatformTransactionManager into the "transactionManager" property. This will usually be a LocalJiniTransactionManager.

Dynamic scaling of the number of concurrent invokers can be activated through specifying a "maxConcurrentConsumers" value that is higher than the "concurrentConsumers" value. Since the latter's default is 1, you can also simply specify a "maxConcurrentConsumers" of e.g. 5, which will lead to dynamic scaling up to 5 concurrent consumers in case of increasing event load, as well as dynamic shrinking back to the standard number of consumers once the load decreases. Consider adapting the "idleTaskExecutionLimit" setting to control the lifespan of each new task, to avoid frequent scaling up and down. Note that using more than one consumer might break fifo behavior if fifo is configured by the space or the specific class type.

Author:

kimchy

Field Summary

static long	DEFAULT_RECOVERY_INTERVAL The default recovery interval: 5000 ms = 5 seconds.
static String	DEFAULT_THREAD_NAME_PREFIX Default thread name prefix: "DefaultPollingEventListenerContainer-".

Fields inherited from class org.openspaces.events.polling.AbstractPollingEventListenerContainer

DEFAULT_RECEIVE_TIMEOUT

Fields inherited from class org.openspaces.events.AbstractEventListenerContainer

failedEvents, processedEvents

Fields inherited from class org.openspaces.events.AbstractSpaceListeningContainer

beanName, logger

Constructor Summary

SimplePollingEventListenerContainer()

Method Summary

protected TaskExecutor	createDefaultTaskExecutor() Create a default TaskExecutor.
protected void	doAfterStart()
protected void	doBeforeStop()
protected void	doInitialize() A callback to perform custom initialization steps.
protected void	doRescheduleTask(Object task) Re-executes the given task via this listener container's TaskExecutor.
protected void	doShutdown() Destroy the container by waiting for all the current event listeners to shutdown.
protected void	eventReceived(Object event) Template method that gets called right when a new message has been received, before attempting to process it.
int	getActiveConsumerCount() Return the number of currently active consumers.
int	getConcurrentConsumers() Return the "concurrentConsumer" setting.
int	getIdleTaskExecutionLimit() Return the limit for idle executions of a receive task.
int	getMaxConcurrentConsumers() Return the "maxConcurrentConsumer" setting.
int	getMaxEventsPerTask() Return the maximum number of messages to process in one task.
int	getScheduledConsumerCount() Return the number of currently scheduled consumers.
ServiceDetails[]	getServicesDetails() Retruns one or more service details that the service exposes.
ServiceMonitors[]	getServicesMonitors() Retruns one or more service details that the service exposes.
protected void	handleListenerSetupFailure(Throwable ex, boolean alreadyRecovered) Handle the given exception that arose during setup of a listener.
void	initialize() Initialize this container.
protected void	recoverAfterListenerSetupFailure() Recover this listener container after a listener failed to set itself up, for example reestablishing the underlying Connection.
protected void	refreshConnectionUntilSuccessful() Refresh the underlying Connection, not returning before an attempt has been successful.
protected void	scheduleNewInvokerIfAppropriate() Schedule a new invoker, increasing the total number of scheduled invokers for this listener container, but only if the specified "maxConcurrentConsumers" limit has not been reached yet, and only if this listener container does not currently have idle invokers that are waiting for new messages already.
void	setConcurrentConsumers(int concurrentConsumers) Specify the number of concurrent consumers to create.
void	setIdleTaskExecutionLimit(int idleTaskExecutionLimit) Specify the limit for idle executions of a receive task, not having received any event within its execution.
void	setMaxConcurrentConsumers(int maxConcurrentConsumers) Specify the maximum number of concurrent consumers to create.
void	setMaxEventsPerTask(int maxEventsPerTask) Specify the maximum number of events to process in one task.
void	setRecoveryInterval(long recoveryInterval) Specify the interval between recovery attempts, in milliseconds.
void	setTaskExecutor(TaskExecutor taskExecutor) Set the Spring TaskExecutor to use for running the listener threads.
protected void	sleepInbetweenRecoveryAttempts() Sleep according to the specified recovery interval.

Methods inherited from class org.openspaces.events.polling.AbstractPollingEventListenerContainer

doReceiveAndExecute, getReceiveOperationHandler, getReceiveTimeout, getTriggerOperationHandler, isPassArrayAsIs, receiveAndExecute, receiveEvent, setPassArrayAsIs, setReceiveOperationHandler, setReceiveTimeout, setTriggerOperationHandler

Methods inherited from class org.openspaces.events.AbstractTransactionalEventListenerContainer

getTransactionDefinition, getTransactionManager, getTransactionManagerName, isTransactional, setDisableTransactionValidation, setTransactionIsolationLevel, setTransactionIsolationLevelName, setTransactionManager, setTransactionName, setTransactionTimeout, validateConfiguration

Methods inherited from class org.openspaces.events.AbstractTemplateEventListenerContainer

afterPropertiesSet, getReceiveTemplate, getTemplate, isPerformSnapshot, setPerformSnapshot, setTemplate

Methods inherited from class org.openspaces.events.AbstractEventListenerContainer

doStart, executeListener, getActualEventListener, getApplicationContext, getEventListener, getEventListenerClass, getFailedEvents, getProcessedEvents, handleListenerException, invokeExceptionListener, invokeListener, setApplicationContext, setEventListener, setEventListenerRef

Methods inherited from class org.openspaces.events.AbstractSpaceListeningContainer

destroy, doStop, getBeanName, getGigaSpace, getStatus, isActive, isRunning, message, onApplicationEvent, rescheduleTaskIfNecessary, setActiveWhenPrimary, setAutoStart, setBeanName, setGigaSpace, setRegisterSpaceModeListener, shutdown, start, stop, waitWhileNotRunning

Methods inherited from class java.lang.Object

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

Field Detail

DEFAULT_THREAD_NAME_PREFIX

public static final String DEFAULT_THREAD_NAME_PREFIX

Default thread name prefix: "DefaultPollingEventListenerContainer-".

DEFAULT_RECOVERY_INTERVAL

public static final long DEFAULT_RECOVERY_INTERVAL

The default recovery interval: 5000 ms = 5 seconds.

See Also:

Constant Field Values

Constructor Detail

SimplePollingEventListenerContainer

public SimplePollingEventListenerContainer()

Method Detail

setTaskExecutor

public void setTaskExecutor(TaskExecutor taskExecutor)

Set the Spring TaskExecutor to use for running the listener threads. Default is SimpleAsyncTaskExecutor, starting up a number of new threads, according to the specified number of concurrent consumers.

Specify an alternative TaskExecutor for integration with an existing thread pool. Note that this really only adds value if the threads are managed in a specific fashion, for example within a J2EE environment. A plain thread pool does not add much value, as this listener container will occupy a number of threads for its entire lifetime.

See Also:

setConcurrentConsumers(int), SimpleAsyncTaskExecutor

setRecoveryInterval

public void setRecoveryInterval(long recoveryInterval)

Specify the interval between recovery attempts, in milliseconds. The default is 5000 ms, that is, 5 seconds.

See Also:

handleListenerSetupFailure(java.lang.Throwable, boolean)

setConcurrentConsumers

public void setConcurrentConsumers(int concurrentConsumers)

Specify the number of concurrent consumers to create. Default is 1.

Specifying a higher value for this setting will increase the standard level of scheduled concurrent consumers at runtime: This is effectively the minimum number of concurrent consumers which will be scheduled at any given time. This is a static setting; for dynamic scaling, consider specifying the "maxConcurrentConsumers" setting instead.

Raising the number of concurrent consumers is recommendable in order to scale the consumption of events. However, note that any ordering guarantees are lost once multiple consumers are registered. In general, stick with 1 consumer for low-volume events.

This setting can be modified at runtime, for example through JMX.

See Also:

setMaxConcurrentConsumers(int)

getConcurrentConsumers

public final int getConcurrentConsumers()

Return the "concurrentConsumer" setting.

This returns the currently configured "concurrentConsumers" value; the number of currently scheduled/active consumers might differ.

See Also:

getScheduledConsumerCount(), getActiveConsumerCount()

setMaxConcurrentConsumers

public void setMaxConcurrentConsumers(int maxConcurrentConsumers)

Specify the maximum number of concurrent consumers to create. Default is 1.

If this setting is higher than "concurrentConsumers", the listener container will dynamically schedule new consumers at runtime, provided that enough incoming messages are encountered. Once the load goes down again, the number of consumers will be reduced to the standard level ("concurrentConsumers") again.

Raising the number of concurrent consumers is recommendable in order to scale the consumption of events. However, note that any ordering guarantees are lost once multiple consumers are registered. In general, stick with 1 consumer for low-volume events.

This setting can be modified at runtime, for example through JMX.

See Also:

setConcurrentConsumers(int)

getMaxConcurrentConsumers

public final int getMaxConcurrentConsumers()

Return the "maxConcurrentConsumer" setting.

This returns the currently configured "maxConcurrentConsumers" value; the number of currently scheduled/active consumers might differ.

See Also:

getScheduledConsumerCount(), getActiveConsumerCount()

setMaxEventsPerTask

public void setMaxEventsPerTask(int maxEventsPerTask)

Specify the maximum number of events to process in one task. More concretely, this limits the number of event reception attempts per task, which includes receive iterations that did not actually pick up a event until they hit their timeout (see "receiveTimeout" property).

Default is unlimited (-1) in case of a standard TaskExecutor, and 1 in case of a SchedulingTaskExecutor that indicates a preference for short-lived tasks. Specify a number of 10 to 100 messages to balance between extremely long-lived and extremely short-lived tasks here.

Long-lived tasks avoid frequent thread context switches through sticking with the same thread all the way through, while short-lived tasks allow thread pools to control the scheduling. Hence, thread pools will usually prefer short-lived tasks.

This setting can be modified at runtime, for example through JMX.

See Also:

setTaskExecutor(org.springframework.core.task.TaskExecutor), AbstractPollingEventListenerContainer.setReceiveTimeout(long), SchedulingTaskExecutor.prefersShortLivedTasks()

getMaxEventsPerTask

public int getMaxEventsPerTask()

Return the maximum number of messages to process in one task.

setIdleTaskExecutionLimit

public void setIdleTaskExecutionLimit(int idleTaskExecutionLimit)

Specify the limit for idle executions of a receive task, not having received any event within its execution. If this limit is reached, the task will shut down and leave receiving to other executing tasks (in case of dynamic scheduling; see the "maxConcurrentConsumers" setting). Default is 1.

Within each task execution, a number of event reception attempts (according to the "maxEventsPerTask" setting) will each wait for an incoming event (according to the "receiveTimeout" setting). If all of those receive attempts in a given task return without an event, the task is considered idle with respect to received events. Such a task may still be rescheduled; however, once it reached the specified "idleTaskExecutionLimit", it will shut down (in case of dynamic scaling).

Raise this limit if you encounter too frequent scaling up and down. With this limit being higher, an idle consumer will be kept around longer, avoiding the restart of a consumer once a new load of messages comes in. Alternatively, specify a higher "maxMessagePerTask" and/or "receiveTimeout" value, which will also lead to idle consumers being kept around for a longer time (while also increasing the average execution time of each scheduled task).

This setting can be modified at runtime, for example through JMX.

See Also:

setMaxEventsPerTask(int), AbstractPollingEventListenerContainer.setReceiveTimeout(long)

getIdleTaskExecutionLimit

public int getIdleTaskExecutionLimit()

Return the limit for idle executions of a receive task.

initialize

public void initialize()

Description copied from class: AbstractSpaceListeningContainer

Initialize this container. If this container is not configured with "activeWhenPrimary" flag set to true will call AbstractSpaceListeningContainer.doStart() (if it is set to true, lifecycle of the container will be controlled by the current space mode). AbstractSpaceListeningContainer.doInitialize() will be called for additional initialization after the possible AbstractSpaceListeningContainer.doStart() call.

Overrides:

initialize in class AbstractPollingEventListenerContainer

See Also:

AbstractSpaceListeningContainer.onApplicationEvent(org.springframework.context.ApplicationEvent)

doAfterStart

protected void doAfterStart()
                     throws DataAccessException

Overrides:

doAfterStart in class AbstractSpaceListeningContainer

Throws:

DataAccessException

doBeforeStop

protected void doBeforeStop()
                     throws DataAccessException

Overrides:

doBeforeStop in class AbstractSpaceListeningContainer

Throws:

DataAccessException

createDefaultTaskExecutor

protected TaskExecutor createDefaultTaskExecutor()

Create a default TaskExecutor. Called if no explicit TaskExecutor has been specified.

The default implementation builds a SimpleAsyncTaskExecutor with the specified bean name (or the class name, if no bean name specified) as thread name prefix.

See Also:

SimpleAsyncTaskExecutor.SimpleAsyncTaskExecutor(String)

doRescheduleTask

protected void doRescheduleTask(Object task)

Re-executes the given task via this listener container's TaskExecutor.

Overrides:

doRescheduleTask in class AbstractSpaceListeningContainer

Parameters:

task - the task object to reschedule

See Also:

setTaskExecutor(org.springframework.core.task.TaskExecutor)

eventReceived

protected void eventReceived(Object event)

Description copied from class: AbstractPollingEventListenerContainer

Template method that gets called right when a new message has been received, before attempting to process it. Allows subclasses to react to the event of an actual incoming message, for example adapting their consumer count.

Overrides:

eventReceived in class AbstractPollingEventListenerContainer

scheduleNewInvokerIfAppropriate

protected void scheduleNewInvokerIfAppropriate()

Schedule a new invoker, increasing the total number of scheduled invokers for this listener container, but only if the specified "maxConcurrentConsumers" limit has not been reached yet, and only if this listener container does not currently have idle invokers that are waiting for new messages already.

Called once an event has been received, to scale up while processing the event in the invoker that originally received it.

See Also:

setTaskExecutor(org.springframework.core.task.TaskExecutor), getMaxConcurrentConsumers()

getScheduledConsumerCount

public final int getScheduledConsumerCount()

Return the number of currently scheduled consumers.

This number will always be inbetween "concurrentConsumers" and "maxConcurrentConsumers", but might be higher than "activeConsumerCount" (in case of some consumers being scheduled but not executed at the moment).

See Also:

getConcurrentConsumers(), getMaxConcurrentConsumers(), getActiveConsumerCount()

getActiveConsumerCount

public final int getActiveConsumerCount()

Return the number of currently active consumers.

This number will always be inbetween "concurrentConsumers" and "maxConcurrentConsumers", but might be lower than "scheduledConsumerCount". (in case of some consumers being scheduled but not executed at the moment).

See Also:

getConcurrentConsumers(), getMaxConcurrentConsumers(), getActiveConsumerCount()

handleListenerSetupFailure

protected void handleListenerSetupFailure(Throwable ex,
                                          boolean alreadyRecovered)

Handle the given exception that arose during setup of a listener. Called for every such exception in every concurrent listener.

The default implementation logs the exception at error level if not recovered yet, and at debug level if already recovered. Can be overridden in subclasses.

Parameters:

ex - the exception to handle

alreadyRecovered - whether a previously executing listener already recovered from the present listener setup failure (this usually indicates a follow-up failure than be ignored other than for debug log purposes)

See Also:

recoverAfterListenerSetupFailure()

recoverAfterListenerSetupFailure

protected void recoverAfterListenerSetupFailure()

Recover this listener container after a listener failed to set itself up, for example reestablishing the underlying Connection.

The default implementation delegates to refreshConnectionUntilSuccessful which pings the space until it is available.

See Also:

refreshConnectionUntilSuccessful()

refreshConnectionUntilSuccessful

protected void refreshConnectionUntilSuccessful()

Refresh the underlying Connection, not returning before an attempt has been successful.

The default implementation pings the space until a successful ping has been established.

See Also:

setRecoveryInterval(long)

sleepInbetweenRecoveryAttempts

protected void sleepInbetweenRecoveryAttempts()

Sleep according to the specified recovery interval. Called inbetween recovery attempts.

doInitialize

protected void doInitialize()
                     throws DataAccessException

Description copied from class: AbstractSpaceListeningContainer

A callback to perform custom initialization steps.

Specified by:

doInitialize in class AbstractSpaceListeningContainer

Throws:

DataAccessException

doShutdown

protected void doShutdown()
                   throws DataAccessException

Destroy the container by waiting for all the current event listeners to shutdown.

Specified by:

doShutdown in class AbstractSpaceListeningContainer

Throws:

DataAccessException

See Also:

AbstractSpaceListeningContainer.shutdown()

getServicesDetails

public ServiceDetails[] getServicesDetails()

Description copied from interface: ServiceDetailsProvider

Retruns one or more service details that the service exposes.

getServicesMonitors

public ServiceMonitors[] getServicesMonitors()

Description copied from interface: ServiceMonitorsProvider

Retruns one or more service details that the service exposes.