com.broadleafcommerce.jobsevents.service

Class DatabaseSystemEventNodeImpl

java.lang.Object

com.broadleafcommerce.jobsevents.service.DatabaseSystemEventNodeImpl

All Implemented Interfaces:

DatabaseSystemEventNode, EventListener, org.springframework.beans.factory.Aware, org.springframework.context.ApplicationContextAware, org.springframework.context.ApplicationListener<org.springframework.context.event.ContextRefreshedEvent>

@Service(value="blDatabaseSystemEventNode")
 @ManagedResource(objectName="org.broadleafcommerce:name=DatabaseSystemEventNode",
                 description="Database System Event Node",
                 currencyTimeLimit=15)
public class DatabaseSystemEventNodeImpl
extends Object
implements DatabaseSystemEventNode, org.springframework.context.ApplicationContextAware, org.springframework.context.ApplicationListener<org.springframework.context.event.ContextRefreshedEvent>

Author:

Jeff Fischer

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	DatabaseSystemEventNodeImpl.ConsumeJob
static class	DatabaseSystemEventNodeImpl.DurableConsumeJob
static class	DatabaseSystemEventNodeImpl.KeepAliveJob

Field Summary

Fields

Modifier and Type	Field and Description
protected CacheInvalidationHelper	cacheInvalidationHelper
protected org.springframework.context.ApplicationContext	context
protected int	durableEventsPerPoll
protected org.quartz.Scheduler	durableScheduler
protected Map<String,List<SystemEventConsumer>>	eventConsumers
protected boolean	eventConsumptionEnabled
protected EventLockUtil	eventLockUtil
protected int	eventsPerPoll
protected EventWorkerType	eventWorkerType
protected org.broadleafcommerce.profile.core.service.IdGenerationService	idGenerationService
protected long	jobThreadPoolCount
protected org.quartz.Scheduler	keepAliveScheduler
protected long	missedDurableEventsPollingThresholdSeconds The maximum age of a durable events (in seconds) that a node will consume.
protected boolean	pause
protected long	pollingIntervalSeconds
protected ProcessManager	processManager
protected org.quartz.Scheduler	scheduler
protected DatabaseSystemEventDao	systemEventDao
protected SystemEventSender	systemEventSender
protected org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor	taskExecutor
protected List<ScheduledJobThreadInitializer>	threadInitializers
protected org.springframework.transaction.PlatformTransactionManager	transactionManager
protected org.broadleafcommerce.common.util.StreamingTransactionCapableUtil	transUtil

Constructor Summary

Constructors

Constructor and Description
DatabaseSystemEventNodeImpl()

Method Summary

Modifier and Type	Method and Description
protected void	buildEventScheduler()
void	consumeLocalEvent(SystemEvent event) Cause asynchronous consumption of this event to occur in the same JVM, rather than provide to event to the cluster.
EventWorkerType	determineHostWorkerType() Determine if this is an ADMIN or SITE node
protected void	executeEvent(SystemEventConsumer consumer, SystemEvent lockedEvent)
void	executeEvents(List<SystemEvent> events, boolean durable) Iterate through a list of events and provide those events to the appropriate SystemEventConsumer instances.
int	getDurableEventsPerPoll() The number of events of type DURABLE_GLOBAL to retrieve in each poll.
Map<String,List<SystemEventConsumer>>	getEventConsumers() All the SystemEventConsumer instances registered in Spring.
int	getEventsPerPoll() The number of events to retrieve in each poll.
long	getJobThreadPoolCount() The number of threads actively polling for available events.
long	getMissedDurableEventsPollingThresholdSeconds() The maximum amount of time into the past that durable events should be retrieved.
Long	getNodeId()
NodeRegistration	getNodeRegistration() Get the NodeRegistration associated.
long	getPollingIntervalSeconds() The wait time between each polling attempt for available events.
void	invalidateCacheElement(String entityClass, String id)
void	invalidateCacheElement(String entityClass, String property, String id) Utility method for initiating a cache invalidation event.
void	invalidateCacheRegion(String region)
void	invalidateCategory(String id)
void	invalidateCustomer(String id)
void	invalidateOffer(String id)
void	invalidateProduct(String id)
boolean	isCreated()
boolean	isPause() Whether or not this node is paused.
void	onApplicationEvent(org.springframework.context.event.ContextRefreshedEvent event)
protected void	removeUniversalDuplicates(List<SystemEvent> events)
void	setApplicationContext(org.springframework.context.ApplicationContext applicationContext)
void	setDurableEventsPerPoll(int durableEventsPerPoll)
void	setEventConsumers(Map<String,List<SystemEventConsumer>> eventConsumers)
void	setEventsPerPoll(int eventsPerPoll)
void	setJobThreadPoolCount(long jobThreadPoolCount)
void	setMissedDurableEventsPollingThresholdSeconds(long missedDurableEventsPollingThresholdSeconds)
void	setPause(boolean pause)
void	setPollingIntervalSeconds(long pollingIntervalSeconds)
void	stop()

Methods inherited from class java.lang.Object

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

Field Detail

transUtil

protected org.broadleafcommerce.common.util.StreamingTransactionCapableUtil transUtil

transactionManager

protected org.springframework.transaction.PlatformTransactionManager transactionManager

idGenerationService

protected org.broadleafcommerce.profile.core.service.IdGenerationService idGenerationService

systemEventDao

protected DatabaseSystemEventDao systemEventDao

processManager

protected ProcessManager processManager

pollingIntervalSeconds

@Value(value="${database.event.polling.interval.seconds}")
protected long pollingIntervalSeconds

eventsPerPoll

@Value(value="${database.event.quantity.per.poll:5}")
protected int eventsPerPoll

durableEventsPerPoll

@Value(value="${database.durable.event.quantity.per.poll:250}")
protected int durableEventsPerPoll

missedDurableEventsPollingThresholdSeconds

@Value(value="${database.missed.durable.event.polling.threshold.seconds:604800}")
protected long missedDurableEventsPollingThresholdSeconds

The maximum age of a durable events (in seconds) that a node will consume. A value of -1 means there is no max age and all available missed durable events will be processed. 604800 is the default (7 days).

eventConsumptionEnabled

@Value(value="${database.event.consumption.enabled:true}")
protected boolean eventConsumptionEnabled

systemEventSender

protected SystemEventSender systemEventSender

cacheInvalidationHelper

protected CacheInvalidationHelper cacheInvalidationHelper

taskExecutor

protected org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor taskExecutor

eventLockUtil

protected EventLockUtil eventLockUtil

threadInitializers

protected List<ScheduledJobThreadInitializer> threadInitializers

eventConsumers

protected Map<String,List<SystemEventConsumer>> eventConsumers

scheduler

protected org.quartz.Scheduler scheduler

durableScheduler

protected org.quartz.Scheduler durableScheduler

keepAliveScheduler

protected org.quartz.Scheduler keepAliveScheduler

pause

protected boolean pause

jobThreadPoolCount

protected long jobThreadPoolCount

eventWorkerType

protected EventWorkerType eventWorkerType

context

protected org.springframework.context.ApplicationContext context

Constructor Detail

DatabaseSystemEventNodeImpl

public DatabaseSystemEventNodeImpl()

Method Detail

onApplicationEvent

public void onApplicationEvent(org.springframework.context.event.ContextRefreshedEvent event)

Specified by:

onApplicationEvent in interface org.springframework.context.ApplicationListener<org.springframework.context.event.ContextRefreshedEvent>

buildEventScheduler

protected void buildEventScheduler()
                            throws IOException,
                                   org.quartz.SchedulerException

Throws:

IOException

org.quartz.SchedulerException

consumeLocalEvent

public void consumeLocalEvent(SystemEvent event)

Description copied from interface: DatabaseSystemEventNode

Cause asynchronous consumption of this event to occur in the same JVM, rather than provide to event to the cluster. This avenue also avoids persistence of the event in the database.

Specified by:

consumeLocalEvent in interface DatabaseSystemEventNode

stop

@PreDestroy
public void stop()

setApplicationContext

public void setApplicationContext(org.springframework.context.ApplicationContext applicationContext)
                           throws org.springframework.beans.BeansException

Specified by:

setApplicationContext in interface org.springframework.context.ApplicationContextAware

Throws:

org.springframework.beans.BeansException

determineHostWorkerType

public EventWorkerType determineHostWorkerType()

Description copied from interface: DatabaseSystemEventNode

Determine if this is an ADMIN or SITE node

Specified by:

determineHostWorkerType in interface DatabaseSystemEventNode

Returns:

executeEvents

public void executeEvents(List<SystemEvent> events,
                          boolean durable)

Description copied from interface: DatabaseSystemEventNode

Iterate through a list of events and provide those events to the appropriate SystemEventConsumer instances.

Specified by:

executeEvents in interface DatabaseSystemEventNode

durable - Whether or not these events are durable (i.e. GLOBAL_DURABLE)

getNodeRegistration

public NodeRegistration getNodeRegistration()

Description copied from interface: DatabaseSystemEventNode

Get the NodeRegistration associated. Every instance of DatabaseSystemEventNode creates a registration in the database.

Specified by:

getNodeRegistration in interface DatabaseSystemEventNode

Returns:

isPause

@ManagedAttribute(description="Whether or not event processing is currently paused.",
                  currencyTimeLimit=15)
public boolean isPause()

Description copied from interface: DatabaseSystemEventNode

Whether or not this node is paused. Paused nodes still participate in heartbeat checks, but do not actively consume events.

Specified by:

isPause in interface DatabaseSystemEventNode

Returns:

setPause

@ManagedAttribute(description="Whether or not event processing is currently paused.",
                  currencyTimeLimit=15)
public void setPause(boolean pause)

Specified by:

setPause in interface DatabaseSystemEventNode

invalidateCacheElement

@ManagedOperation(description="Invalidate the cache for an entity collection property",
                  currencyTimeLimit=15)
public void invalidateCacheElement(String entityClass,
                                                                                                                                                                   String property,
                                                                                                                                                                   String id)

Description copied from interface: DatabaseSystemEventNode

Utility method for initiating a cache invalidation event.

Specified by:

invalidateCacheElement in interface DatabaseSystemEventNode

invalidateCacheElement

@ManagedOperation(description="Invalidate the cache for an entity",
                  currencyTimeLimit=15)
public void invalidateCacheElement(String entityClass,
                                                                                                                                               String id)

Specified by:

invalidateCacheElement in interface DatabaseSystemEventNode

invalidateCacheRegion

@ManagedOperation(description="Invalidate an entire cache region",
                  currencyTimeLimit=15)
public void invalidateCacheRegion(String region)

Specified by:

invalidateCacheRegion in interface DatabaseSystemEventNode

invalidateProduct

@ManagedOperation(description="Convenience method for invalidating a Product",
                  currencyTimeLimit=15)
public void invalidateProduct(String id)

Specified by:

invalidateProduct in interface DatabaseSystemEventNode

invalidateCategory

@ManagedOperation(description="Convenience method for invalidating a Category",
                  currencyTimeLimit=15)
public void invalidateCategory(String id)

Specified by:

invalidateCategory in interface DatabaseSystemEventNode

invalidateOffer

@ManagedOperation(description="Convenience method for invalidating a Offer",
                  currencyTimeLimit=15)
public void invalidateOffer(String id)

Specified by:

invalidateOffer in interface DatabaseSystemEventNode

invalidateCustomer

@ManagedOperation(description="Convenience method for invalidating a Customer",
                  currencyTimeLimit=15)
public void invalidateCustomer(String id)

Specified by:

invalidateCustomer in interface DatabaseSystemEventNode

getPollingIntervalSeconds

public long getPollingIntervalSeconds()

Description copied from interface: DatabaseSystemEventNode

The wait time between each polling attempt for available events. This value can be decreased for quicker event consumption response times at the cost of increased database traffic. Defaults to 5.

Specified by:

getPollingIntervalSeconds in interface DatabaseSystemEventNode

Returns:

setPollingIntervalSeconds

public void setPollingIntervalSeconds(long pollingIntervalSeconds)

Specified by:

setPollingIntervalSeconds in interface DatabaseSystemEventNode

getEventsPerPoll

public int getEventsPerPoll()

Description copied from interface: DatabaseSystemEventNode

The number of events to retrieve in each poll. This can be increased to process more events at any given time at the cost of prolonged CPU consumption. It is likely advantageous to throttle event consumption to avoid overly taxing the system and penalizing other operations. Also, smaller event processing groups allows the system to respond more quickly to events that utilize the priority feature (see SystemEvent.getPriority(). The default value is 5. This does not include events of type GLOBAL_DURABLE.

Specified by:

getEventsPerPoll in interface DatabaseSystemEventNode

Returns:

setEventsPerPoll

public void setEventsPerPoll(int eventsPerPoll)

Specified by:

setEventsPerPoll in interface DatabaseSystemEventNode

getJobThreadPoolCount

public long getJobThreadPoolCount()

Description copied from interface: DatabaseSystemEventNode

The number of threads actively polling for available events. Defaults to 1.

Specified by:

getJobThreadPoolCount in interface DatabaseSystemEventNode

Returns:

setJobThreadPoolCount

public void setJobThreadPoolCount(long jobThreadPoolCount)

Specified by:

setJobThreadPoolCount in interface DatabaseSystemEventNode

getEventConsumers

public Map<String,List<SystemEventConsumer>> getEventConsumers()

Description copied from interface: DatabaseSystemEventNode

All the SystemEventConsumer instances registered in Spring.

Specified by:

getEventConsumers in interface DatabaseSystemEventNode

Returns:

setEventConsumers

public void setEventConsumers(Map<String,List<SystemEventConsumer>> eventConsumers)

Specified by:

setEventConsumers in interface DatabaseSystemEventNode

getMissedDurableEventsPollingThresholdSeconds

public long getMissedDurableEventsPollingThresholdSeconds()

Description copied from interface: DatabaseSystemEventNode

The maximum amount of time into the past that durable events should be retrieved. Durable events older than the threshold will be ignored. The default value is 604800 (7 days).

Specified by:

getMissedDurableEventsPollingThresholdSeconds in interface DatabaseSystemEventNode

Returns:

getDurableEventsPerPoll

public int getDurableEventsPerPoll()

Description copied from interface: DatabaseSystemEventNode

The number of events of type DURABLE_GLOBAL to retrieve in each poll. The default value is 250. Durable events are always processed in order and do not take advantage of SystemEvent.getPriority().

Specified by:

getDurableEventsPerPoll in interface DatabaseSystemEventNode

Returns:

setDurableEventsPerPoll

public void setDurableEventsPerPoll(int durableEventsPerPoll)

Specified by:

setDurableEventsPerPoll in interface DatabaseSystemEventNode

setMissedDurableEventsPollingThresholdSeconds

public void setMissedDurableEventsPollingThresholdSeconds(long missedDurableEventsPollingThresholdSeconds)

Specified by:

setMissedDurableEventsPollingThresholdSeconds in interface DatabaseSystemEventNode

isCreated

public boolean isCreated()

Specified by:

isCreated in interface DatabaseSystemEventNode

getNodeId

public Long getNodeId()

Specified by:

getNodeId in interface DatabaseSystemEventNode

executeEvent

protected void executeEvent(SystemEventConsumer consumer,
                            SystemEvent lockedEvent)

removeUniversalDuplicates

protected void removeUniversalDuplicates(List<SystemEvent> events)