org.nuxeo.ecm.platform.importer.mqueues.workmanager

Class WorkManagerComputation

java.lang.Object

org.nuxeo.runtime.model.DefaultComponent

org.nuxeo.ecm.core.work.WorkManagerImpl

org.nuxeo.ecm.platform.importer.mqueues.workmanager.WorkManagerComputation

All Implemented Interfaces:

WorkManager, Adaptable, Component, Extensible, TimestampedService

Direct Known Subclasses:

WorkManagerComputationChronicle, WorkManagerComputationKafka

public abstract class WorkManagerComputation
extends WorkManagerImpl

Since:

9.2

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
class	WorkManagerComputation.WorkScheduling

Nested classes/interfaces inherited from class org.nuxeo.ecm.core.work.WorkManagerImpl

WorkManagerImpl.WorkCompletionSynchronizer, WorkManagerImpl.WorkThreadPoolExecutor

Nested classes/interfaces inherited from interface org.nuxeo.ecm.core.work.api.WorkManager

WorkManager.Scheduling

Field Summary

Fields

Modifier and Type	Field and Description
protected static int	DEFAULT_CONCURRENCY
protected static org.apache.commons.logging.Log	log
protected MQComputationManager	manager
protected MQManager<Record>	mqManager
protected Settings	settings
protected Set<String>	streamIds
protected Topology	topology

Fields inherited from class org.nuxeo.ecm.core.work.WorkManagerImpl

completionSynchronizer, DEFAULT_CATEGORY, DEFAULT_QUEUE_ID, executors, IMPL_EP, NAME, QUEUES_EP, queuing, registry, shutdownInProgress, started, THREAD_PREFIX, workQueueConfig, workQueuingConfig

Fields inherited from class org.nuxeo.runtime.model.DefaultComponent

lastModified

Constructor Summary

Constructors

Constructor and Description
WorkManagerComputation()

Method Summary

Modifier and Type	Method and Description
boolean	awaitCompletion(String queueId, long duration, TimeUnit unit) Waits for completion of work in a given queue.
Work	find(String s, Work.State state) Finds a work instance.
int	getApplicationStartedOrder() The component notification order for Component.applicationStarted(org.nuxeo.runtime.model.ComponentContext).
WorkQueueMetrics	getMetrics(String queueId) Gets the metrics for the queueId
protected abstract int	getOverProvisioningFactor()
protected int	getPartitions(int maxThreads)
int	getQueueSize(String queueId, Work.State state) Gets the number of work instances in a given queue in a defined state.
String	getStreamForCategory(String category)
Work.State	getWorkState(String s) Gets the state in which a work instance is.
void	init() Starts up this WorkManager and attempts to resume work previously suspended and saved at WorkManager.shutdown(long, java.util.concurrent.TimeUnit) time.
protected abstract MQManager<Record>	initStream()
protected void	initTopology()
List<Work>	listWork(String s, Work.State state) Lists the work instances in a given queue in a defined state.
List<String>	listWorkIds(String s, Work.State state) Lists the work ids in a given queue in a defined state.
void	schedule(Work work, WorkManager.Scheduling scheduling, boolean afterCommit) Schedules work for execution at a later time, with a specific scheduling policy.
protected boolean	scheduleAfterCommit(Work work, WorkManager.Scheduling scheduling) Schedule after commit.
boolean	shutdown(long timeout, TimeUnit timeUnit) Shuts down this WorkManager and attempts to suspend and save the running and scheduled work instances.
boolean	shutdownQueue(String queueId, long timeout, TimeUnit unit) Shuts down a work queue and attempts to suspend and save the running and scheduled work instances.
void	start(ComponentContext context) Start the component.
protected void	startComputation()
protected void	supplantWorkManagerImpl() Hack to steal the WorkManagerImpl queue contributions.

Methods inherited from class org.nuxeo.ecm.core.work.WorkManagerImpl

awaitCompletion, enableProcessing, enableProcessing, getCategoryQueueId, getExecutor, getTimestampAfter, getWorkQueueDescriptor, getWorkQueueIds, hasWorkInState, isProcessingEnabled, isProcessingEnabled, isQueuingEnabled, isStarted, newWorkQueuing, noScheduledOrRunningWork, registerContribution, remainingMillis, removeExecutor, schedule, schedule, schedule, shutdownExecutors, stop, unregisterContribution

Methods inherited from class org.nuxeo.runtime.model.DefaultComponent

activate, deactivate, getAdapter, getLastModified, registerExtension, setLastModified, setModifiedNow, unregisterExtension

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.nuxeo.runtime.model.Component

applicationStarted

Field Detail

log

protected static final org.apache.commons.logging.Log log

DEFAULT_CONCURRENCY

protected static final int DEFAULT_CONCURRENCY

See Also:

Constant Field Values

topology

protected Topology topology

settings

protected Settings settings

manager

protected MQComputationManager manager

mqManager

protected MQManager<Record> mqManager

streamIds

protected final Set<String> streamIds

Constructor Detail

WorkManagerComputation

public WorkManagerComputation()

Method Detail

initStream

protected abstract MQManager<Record> initStream()

getOverProvisioningFactor

protected abstract int getOverProvisioningFactor()

schedule

public void schedule(Work work,
                     WorkManager.Scheduling scheduling,
                     boolean afterCommit)

Description copied from interface: WorkManager

Schedules work for execution at a later time, with a specific scheduling policy.

Specified by:

schedule in interface WorkManager

Overrides:

schedule in class WorkManagerImpl

Parameters:

work - the work to execute

scheduling - the scheduling policy

afterCommit - if true and the work is scheduled, it will only be run after the current transaction (if any) has committed

See Also:

WorkManager.schedule(Work)

getStreamForCategory

public String getStreamForCategory(String category)

getApplicationStartedOrder

public int getApplicationStartedOrder()

Description copied from interface: Component

The component notification order for Component.applicationStarted(org.nuxeo.runtime.model.ComponentContext).

Components are notified in increasing order. Order 1000 is the default order for components that don't care. Order 100 is the repository initialization.

Specified by:

getApplicationStartedOrder in interface Component

Overrides:

getApplicationStartedOrder in class WorkManagerImpl

Returns:

the order, 1000 by default

start

public void start(ComponentContext context)

Description copied from interface: Component

Start the component. This method is called after all the components were resolved and activated

Specified by:

start in interface Component

Overrides:

start in class WorkManagerImpl

init

public void init()

Description copied from interface: WorkManager

Starts up this WorkManager and attempts to resume work previously suspended and saved at WorkManager.shutdown(long, java.util.concurrent.TimeUnit) time.

Specified by:

init in interface WorkManager

Overrides:

init in class WorkManagerImpl

supplantWorkManagerImpl

protected void supplantWorkManagerImpl()

Hack to steal the WorkManagerImpl queue contributions.

startComputation

protected void startComputation()

initTopology

protected void initTopology()

getPartitions

protected int getPartitions(int maxThreads)

shutdownQueue

public boolean shutdownQueue(String queueId,
                             long timeout,
                             TimeUnit unit)
                      throws InterruptedException

Description copied from interface: WorkManager

Shuts down a work queue and attempts to suspend and save the running and scheduled work instances.

Specified by:

shutdownQueue in interface WorkManager

Overrides:

shutdownQueue in class WorkManagerImpl

Parameters:

queueId - the queue id

timeout - the time to wait

unit - the timeout unit

Returns:

true if shutdown is done, false if there are still some threads executing after the timeout

Throws:

InterruptedException

shutdown

public boolean shutdown(long timeout,
                        TimeUnit timeUnit)
                 throws InterruptedException

Description copied from interface: WorkManager

Shuts down this WorkManager and attempts to suspend and save the running and scheduled work instances.

Specified by:

shutdown in interface WorkManager

Overrides:

shutdown in class WorkManagerImpl

Parameters:

timeout - the time to wait

timeUnit - the timeout unit

Returns:

true if shutdown is done, false if there are still some threads executing after the timeout

Throws:

InterruptedException

getQueueSize

public int getQueueSize(String queueId,
                        Work.State state)

Description copied from interface: WorkManager

Gets the number of work instances in a given queue in a defined state.

Specified by:

getQueueSize in interface WorkManager

Overrides:

getQueueSize in class WorkManagerImpl

Parameters:

queueId - the queue id

state - the state defining the state to look into, SCHEDULED, RUNNING, COMPLETED, or null for non-completed ( SCHEDULED or RUNNING)

Returns:

the number of work instances in the given state

getMetrics

public WorkQueueMetrics getMetrics(String queueId)

Description copied from interface: WorkManager

Gets the metrics for the queueId

Specified by:

getMetrics in interface WorkManager

Overrides:

getMetrics in class WorkManagerImpl

awaitCompletion

public boolean awaitCompletion(String queueId,
                               long duration,
                               TimeUnit unit)
                        throws InterruptedException

Description copied from interface: WorkManager

Waits for completion of work in a given queue.

Specified by:

awaitCompletion in interface WorkManager

Overrides:

awaitCompletion in class WorkManagerImpl

Parameters:

queueId - the queue id

duration - the time to wait

unit - the timeout unit

Returns:

true if all work completed in the queue, or false if there is still some non-completed work after the timeout

Throws:

InterruptedException

getWorkState

public Work.State getWorkState(String s)

Description copied from interface: WorkManager

Gets the state in which a work instance is.

This can be SCHEDULED, RUNNING, or null.

Specified by:

getWorkState in interface WorkManager

Overrides:

getWorkState in class WorkManagerImpl

Parameters:

s - the id of the work to find

Returns:

the work state, or null if not found

find

public Work find(String s,
                 Work.State state)

Description copied from interface: WorkManager

Finds a work instance.

Specified by:

find in interface WorkManager

Overrides:

find in class WorkManagerImpl

Parameters:

s - the id of the work to find

state - the state defining the state to look into, SCHEDULED, RUNNING, or null for non-completed

Returns:

the found work instance, or null if not found

listWork

public List<Work> listWork(String s,
                           Work.State state)

Description copied from interface: WorkManager

Lists the work instances in a given queue in a defined state.

Specified by:

listWork in interface WorkManager

Overrides:

listWork in class WorkManagerImpl

Parameters:

s - the queue id

state - the state defining the state to look into, SCHEDULED, RUNNING, or null for non-completed

Returns:

the list of work instances in the given state

listWorkIds

public List<String> listWorkIds(String s,
                                Work.State state)

Description copied from interface: WorkManager

Lists the work ids in a given queue in a defined state.

Specified by:

listWorkIds in interface WorkManager

Overrides:

listWorkIds in class WorkManagerImpl

Parameters:

s - the queue id

state - the state defining the state to look into, SCHEDULED, RUNNING, or null for non-completed

Returns:

the list of work ids in the given state

scheduleAfterCommit

protected boolean scheduleAfterCommit(Work work,
                                      WorkManager.Scheduling scheduling)

Description copied from class: WorkManagerImpl

Schedule after commit. Returns false if impossible (no transaction or transaction manager).

Overrides:

scheduleAfterCommit in class WorkManagerImpl