Package com.broadleafcommerce.search.index.core.worker

Class GenericSearchIndexWorkerProcessLauncher<T>

java.lang.Object

com.broadleafcommerce.search.index.core.worker.GenericSearchIndexWorkerProcessLauncher<T>

Type Parameters:

T - The type of the objects retrieved from the Queue

All Implemented Interfaces:

IndexableTypeAware, SearchIndexWorkerProcessLauncher, Runnable

public abstract class GenericSearchIndexWorkerProcessLauncher<T>
extends Object
implements SearchIndexWorkerProcessLauncher, Runnable

Abstract implementation of SearchIndexWorkerProcessLauncher. This should generally be invoked after receiving a ProcessStartedEvent or a QueueCreatedEvent. This is the entry point for executing worker threads to process batches of IDs from a Queue.

Author:

Kelly Tisdell (ktisdell)

Constructor Summary

Constructors

Modifier	Constructor	Description
protected	GenericSearchIndexWorkerProcessLauncher(com.broadleafcommerce.search.api.type.IndexableType indexableType, ProcessStateService processStateService, QueueProvider queueProvider, int threads, long waitTime, int maxTries)

Method Summary

Modifier and Type	Method	Description
protected Executor	createExecutor(int threads, com.broadleafcommerce.search.api.type.IndexableType type)	This creates a ThreadPoolTaskExecutor by default.
protected abstract AbstractWorker<T>	createNewTask(ProcessStateService processStateService, Semaphore semaphore, com.broadleafcommerce.search.api.type.IndexableType type, ReindexQueue<T> queue)	Launch a worker process.
protected void	destroyExecutor(Executor executor)	This is always called to close the Executor when it's finished.
protected void	executeInternally()
com.broadleafcommerce.search.api.type.IndexableType	getIndexableType()	Components that implement this method should return the same, non-null value on each invocation.
protected int	getMaxTries()	Defines the number of times that we'll try to retrieve something from a queue with the given wait time before giving up.
protected ProcessStateService	getProcessStateService()
protected QueueProvider	getQueueProvider()
protected long	getWaitTime()	Time in milliseconds to wait for something to arrive in the Queue.
protected int	getWorkerThreads()	The number of worker threads which should be started on a reindex process.
static SearchIndexWorkerProcessLauncher	instance(com.broadleafcommerce.search.api.type.IndexableType type)	Convenient way to access the SearchIndexWorkerProcessLauncher associated with a particular IndexableType.
boolean	isActive()	Indicates if this process is currently executing.
protected boolean	isExecuting()	Whether or not this is currently executing.
void	launchWorkers()	Method to begin work by worker or slave processes.
static void	registerSearchIndexWorkerProcessLauncher(SearchIndexWorkerProcessLauncher launcher)	Allows registration of a SearchIndexWorkerProcessLauncher that does not extend GenericSearchIndexWorkerProcessLauncher.
void	run()

Methods inherited from class java.lang.Object

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

Constructor Detail

GenericSearchIndexWorkerProcessLauncher

protected GenericSearchIndexWorkerProcessLauncher(com.broadleafcommerce.search.api.type.IndexableType indexableType,
                                                  ProcessStateService processStateService,
                                                  QueueProvider queueProvider,
                                                  int threads,
                                                  long waitTime,
                                                  int maxTries)

Method Detail

launchWorkers

public final void launchWorkers()

Description copied from interface: SearchIndexWorkerProcessLauncher

Method to begin work by worker or slave processes. This could take place on the same machine / JVM as the master, or it could begin on a different node.

Specified by:

launchWorkers in interface SearchIndexWorkerProcessLauncher

executeInternally

protected void executeInternally()

run

public final void run()

Specified by:

run in interface Runnable

isActive

public final boolean isActive()

Description copied from interface: SearchIndexWorkerProcessLauncher

Indicates if this process is currently executing.

Specified by:

isActive in interface SearchIndexWorkerProcessLauncher

Returns:

whether the worker launcher is executing

createExecutor

protected Executor createExecutor(int threads,
                                  com.broadleafcommerce.search.api.type.IndexableType type)

This creates a ThreadPoolTaskExecutor by default. If you override this method, consider overriding the destroyExecutor method as well.

Parameters:

threads - the number of threads for the executor pool

type - the indexable type for the workers

Returns:

new ThreadPoolTaskExecutor for reindex workers

destroyExecutor

protected void destroyExecutor(Executor executor)

This is always called to close the Executor when it's finished. By default, this expects a ThreadPoolTaskExecutor.

Parameters:

executor - the executor to shut down.

createNewTask

protected abstract AbstractWorker<T> createNewTask(ProcessStateService processStateService,
                                                   Semaphore semaphore,
                                                   com.broadleafcommerce.search.api.type.IndexableType type,
                                                   ReindexQueue<T> queue)

Launch a worker process.

Parameters:

processStateService - the service managing process state

semaphore - the semaphore tracking worker completion

type - the indexableType the worker will process for

queue - the queue the worker will read from

Returns:

the new worker

instance

public static SearchIndexWorkerProcessLauncher instance(com.broadleafcommerce.search.api.type.IndexableType type)

Convenient way to access the SearchIndexWorkerProcessLauncher associated with a particular IndexableType. Returns null if no SearchIndexWorkerProcessLauncher was registered for the specified type.

Parameters:

type - the type to get the worker launcher for

Returns:

the worker launcher for the type

registerSearchIndexWorkerProcessLauncher

public static void registerSearchIndexWorkerProcessLauncher(SearchIndexWorkerProcessLauncher launcher)

Allows registration of a SearchIndexWorkerProcessLauncher that does not extend GenericSearchIndexWorkerProcessLauncher. All instances that extend SearchIndexWorkerProcessLauncher get automatically registered.

This throws an IllegalStateException if another SearchIndexWorkerProcessLauncher is already registered with the same IndexableType.

Parameters:

launcher - the worker launcher to register to its type

getIndexableType

public com.broadleafcommerce.search.api.type.IndexableType getIndexableType()

Description copied from interface: IndexableTypeAware

Components that implement this method should return the same, non-null value on each invocation. This should basically return a constant value.

Specified by:

getIndexableType in interface IndexableTypeAware

Returns:

the associated indexable type

getProcessStateService

protected ProcessStateService getProcessStateService()

getQueueProvider

protected QueueProvider getQueueProvider()

getWorkerThreads

protected int getWorkerThreads()

The number of worker threads which should be started on a reindex process.

Returns:

number of worker threads on a reindex process

getWaitTime

protected long getWaitTime()

Time in milliseconds to wait for something to arrive in the Queue. Keep this rather small.

Returns:

time to wait for something to arrive in the queue

getMaxTries

protected int getMaxTries()

Defines the number of times that we'll try to retrieve something from a queue with the given wait time before giving up. Either nothing is putting data on the queue, or the cluster size is too big and we're just not receiving data because other threads are consuming the data fast enough that we never receive any.

Returns:

number of times to try to retrieve something from the queue

isExecuting

protected boolean isExecuting()

Whether or not this is currently executing.

Returns:

whether or not this is currently executing