GigaSpaces XAP 7.0 API

org.openspaces.events.polling
Class SimplePollingEventListenerContainer

java.lang.Object
  extended by org.openspaces.events.AbstractSpaceListeningContainer
      extended by org.openspaces.events.AbstractEventListenerContainer
          extended by org.openspaces.events.AbstractTemplateEventListenerContainer
              extended by org.openspaces.events.AbstractTransactionalEventListenerContainer
                  extended by org.openspaces.events.polling.AbstractPollingEventListenerContainer
                      extended by 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.


GigaSpaces XAP 7.0 API

Copyright © GigaSpaces.