Package org.nuxeo.ecm.core.work

Class WorkManagerImpl

java.lang.Object

org.nuxeo.runtime.model.DefaultComponent

org.nuxeo.ecm.core.work.WorkManagerImpl

All Implemented Interfaces:

WorkManager, Adaptable, Component, Extensible, TimestampedService

Direct Known Subclasses:

StreamWorkManager

public class WorkManagerImpl
extends DefaultComponent
implements WorkManager

The implementation of a WorkManager. This delegates the queuing implementation to a WorkQueuing implementation.

Since:

5.6

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
protected static class	WorkManagerImpl.WorkCompletionSynchronizer	Simple synchronizer to wake up when an in-JVM work is completed.
class	WorkManagerImpl.WorkScheduling	A work instance and how to schedule it, for schedule-after-commit.
protected class	WorkManagerImpl.WorkThreadPoolExecutor	A ThreadPoolExecutor that keeps available the list of running tasks.

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

WorkManager.Scheduling

Field Summary

Fields

Modifier and Type	Field	Description
protected boolean	active
protected final Map<String,String>	categoryToQueueId
protected WorkManagerImpl.WorkCompletionSynchronizer	completionSynchronizer
static final Name	DEAD_LETTER_QUEUE	The dead letter queue stream name.
static final Codec<Record>	DEAD_LETTER_QUEUE_CODEC
static final String	DEFAULT_CATEGORY
static final String	DEFAULT_QUEUE_ID
protected final Map<String,WorkManagerImpl.WorkThreadPoolExecutor>	executors
protected static final String	GLOBAL_METRIC_PREFIX
protected static final String	IMPL_EP
static final String	NAME
protected static final String	QUEUES_EP
protected WorkQueuing	queuing
protected final io.dropwizard.metrics5.MetricRegistry	registry
static final String	SHUTDOWN_DELAY_MS_KEY
protected boolean	shutdownInProgress
protected boolean	started
protected static final String	THREAD_PREFIX
static final String	WORKMANAGER_PROCESSING_DISABLE	Deprecated. Use WORKMANAGER_PROCESSING_ENABLED instead
static final String	WORKMANAGER_PROCESSING_ENABLED

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

lastModified, name

Constructor Summary

Constructors

Constructor	Description
WorkManagerImpl()

Method Summary

Modifier and Type	Method	Description
boolean	awaitCompletion(long duration, TimeUnit unit)	Waits for completion of all work.
boolean	awaitCompletion(String queueId, long duration, TimeUnit unit)	Waits for completion of work in a given queue.
void	enableProcessing(boolean value)	Set processing for all queues
void	enableProcessing(String queueId, boolean value)	Set processing for a given queue id.
Work	find(String workId, Work.State state)	Finds a work instance.
int	getApplicationStartedOrder()	The component notification order for Component.start(ComponentContext).
String	getCategoryQueueId(String category)	Gets the queue id used for a given work category.
protected WorkManagerImpl.WorkThreadPoolExecutor	getExecutor(String queueId)
WorkQueueMetrics	getMetrics(String queueId)	Gets the metrics for the queueId
protected long	getTimestampAfter(long durationInMs)
WorkQueueDescriptor	getWorkQueueDescriptor(String queueId)	Gets the work queue descriptor for a given queue id.
List<String>	getWorkQueueIds()	Lists the ids of the existing work queues.
Work.State	getWorkState(String workId)	Gets the state in which a work instance is.
protected boolean	hasWorkInState(String workId, Work.State state)
protected void	index()
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 void	initDeadLetterQueueStream()
protected boolean	isProcessingDisabled()
boolean	isProcessingEnabled()	Is processing enabled for at least one queue
boolean	isProcessingEnabled(String queueId)	Is processing enabled for a given queue id.
boolean	isQueuingEnabled(String queueId)	Is queuing enabled for a given queue id.
boolean	isStarted()
List<Work>	listWork(String queueId, Work.State state)	Lists the work instances in a given queue in a defined state.
List<String>	listWorkIds(String queueId, Work.State state)	Lists the work ids in a given queue in a defined state.
protected boolean	noScheduledOrRunningWork(String queueId)
void	registerContribution(Object contribution, String xp, ComponentInstance component)
protected long	remainingMillis(long t0, long delay)	Deprecated. since 10.2 because unused
protected void	removeExecutor(String queueId)	Deprecated. since 10.2 because unused
void	schedule(Work work)	Schedules work for execution at a later time.
void	schedule(Work work, boolean afterCommit)	Schedules work for execution at a later time, after the current transaction (if any) has committed.
void	schedule(Work work, WorkManager.Scheduling scheduling)	Schedules work for execution at a later time, with a specific scheduling policy.
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.
void	setName(String name)	Sets the name for this component, as it was defined in its XML.
boolean	shutdown(long timeout, TimeUnit unit)	Shuts down this WorkManager and attempts to suspend and save the running and scheduled work instances.
protected boolean	shutdownExecutors(Collection<WorkManagerImpl.WorkThreadPoolExecutor> list, long timeout, TimeUnit unit)
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.
boolean	supportsProcessingDisabling()

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

activate, addRuntimeMessage, addRuntimeMessage, deactivate, getAdapter, getDescriptor, getDescriptors, getLastModified, getRegistry, register, registerExtension, setLastModified, setModifiedNow, stop, unregister, unregisterContribution, unregisterExtension

Methods inherited from class java.lang.Object

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

Field Details

NAME

public static final String NAME

See Also:

• Constant Field Values

QUEUES_EP

protected static final String QUEUES_EP

See Also:

• Constant Field Values

IMPL_EP

protected static final String IMPL_EP

See Also:

• Constant Field Values

DEFAULT_QUEUE_ID

public static final String DEFAULT_QUEUE_ID

See Also:

• Constant Field Values

DEFAULT_CATEGORY

public static final String DEFAULT_CATEGORY

See Also:

• Constant Field Values

THREAD_PREFIX

protected static final String THREAD_PREFIX

See Also:

• Constant Field Values

SHUTDOWN_DELAY_MS_KEY

public static final String SHUTDOWN_DELAY_MS_KEY

Since:

10.2

See Also:

• Constant Field Values

WORKMANAGER_PROCESSING_DISABLE

@Deprecated(since="11.2")
public static final String WORKMANAGER_PROCESSING_DISABLE

Deprecated.

Use WORKMANAGER_PROCESSING_ENABLED instead

Since:

11.1

See Also:

• Constant Field Values

WORKMANAGER_PROCESSING_ENABLED

public static final String WORKMANAGER_PROCESSING_ENABLED

Since:

11.2

See Also:

• Constant Field Values

DEAD_LETTER_QUEUE

public static final Name DEAD_LETTER_QUEUE

The dead letter queue stream name.

Since:

11.1

DEAD_LETTER_QUEUE_CODEC

public static final Codec<Record> DEAD_LETTER_QUEUE_CODEC

GLOBAL_METRIC_PREFIX

protected static final String GLOBAL_METRIC_PREFIX

See Also:

• Constant Field Values

registry

protected final io.dropwizard.metrics5.MetricRegistry registry

executors

protected final Map<String,WorkManagerImpl.WorkThreadPoolExecutor> executors

categoryToQueueId

protected final Map<String,String> categoryToQueueId

queuing

protected WorkQueuing queuing

active

protected boolean active

completionSynchronizer

protected WorkManagerImpl.WorkCompletionSynchronizer completionSynchronizer

started

protected volatile boolean started

shutdownInProgress

protected volatile boolean shutdownInProgress

Constructor Details

WorkManagerImpl

public WorkManagerImpl()

Method Details

setName

public void setName(String name)

Description copied from interface: Component

Sets the name for this component, as it was defined in its XML.

This is called once after construction by the runtime framework.

Specified by:

setName in interface Component

Overrides:

setName in class DefaultComponent

Parameters:

name - the name

registerContribution

public void registerContribution(Object contribution,
 String xp,
 ComponentInstance component)

Overrides:

registerContribution in class DefaultComponent

isQueuingEnabled

public boolean isQueuingEnabled(String queueId)

Description copied from interface: WorkManager

Is queuing enabled for a given queue id.

Specified by:

isQueuingEnabled in interface WorkManager

enableProcessing

public void enableProcessing(boolean value)

Description copied from interface: WorkManager

Set processing for all queues

Specified by:

enableProcessing in interface WorkManager

enableProcessing

public void enableProcessing(String queueId,
 boolean value)

Description copied from interface: WorkManager

Set processing for a given queue id.

Specified by:

enableProcessing in interface WorkManager

isProcessingEnabled

public boolean isProcessingEnabled()

Description copied from interface: WorkManager

Is processing enabled for at least one queue

Specified by:

isProcessingEnabled in interface WorkManager

isProcessingDisabled

protected boolean isProcessingDisabled()

isProcessingEnabled

public boolean isProcessingEnabled(String queueId)

Description copied from interface: WorkManager

Is processing enabled for a given queue id.

Specified by:

isProcessingEnabled in interface WorkManager

getWorkQueueIds

public List<String> getWorkQueueIds()

Description copied from interface: WorkManager

Lists the ids of the existing work queues.

Specified by:

getWorkQueueIds in interface WorkManager

Returns:

the list of queue ids

getWorkQueueDescriptor

public WorkQueueDescriptor getWorkQueueDescriptor(String queueId)

Description copied from interface: WorkManager

Gets the work queue descriptor for a given queue id.

Specified by:

getWorkQueueDescriptor in interface WorkManager

Parameters:

queueId - the queue id

Returns:

the work queue descriptor, or null

getCategoryQueueId

public String getCategoryQueueId(String category)

Description copied from interface: WorkManager

Gets the queue id used for a given work category.

Specified by:

getCategoryQueueId in interface WorkManager

Parameters:

category - the category

Returns:

the queue id

getApplicationStartedOrder

public int getApplicationStartedOrder()

Description copied from interface: Component

The component notification order for Component.start(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

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 DefaultComponent

initDeadLetterQueueStream

protected void initDeadLetterQueueStream()

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

index

protected void index()

getExecutor

protected WorkManagerImpl.WorkThreadPoolExecutor getExecutor(String queueId)

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

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

shutdownExecutors

protected boolean shutdownExecutors(Collection<WorkManagerImpl.WorkThreadPoolExecutor> list,
 long timeout,
 TimeUnit unit)
                             throws InterruptedException

Throws:

InterruptedException

remainingMillis

@Deprecated
protected long remainingMillis(long t0,
 long delay)

Deprecated.

since 10.2 because unused

removeExecutor

@Deprecated
protected void removeExecutor(String queueId)

Deprecated.

since 10.2 because unused

shutdown

public boolean shutdown(long timeout,
 TimeUnit unit)
                 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

Parameters:

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

schedule

public void schedule(Work work)

Description copied from interface: WorkManager

Schedules work for execution at a later time.

This method is identical to WorkManager.schedule(Work, boolean) with afterCommit = false.

Specified by:

schedule in interface WorkManager

Parameters:

work - the work to execute

schedule

public void schedule(Work work,
 boolean afterCommit)

Description copied from interface: WorkManager

Schedules work for execution at a later time, after the current transaction (if any) has committed.

Specified by:

schedule in interface WorkManager

Parameters:

work - the work to execute

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

schedule

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

Description copied from interface: WorkManager

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

This method is identical to WorkManager.schedule(Work, Scheduling, boolean) with afterCommit = false.

Specified by:

schedule in interface WorkManager

Parameters:

work - the work to execute

scheduling - the scheduling policy

See Also:

• WorkManager.schedule(Work)

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

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)

scheduleAfterCommit

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

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

Since:

5.8

find

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

Description copied from interface: WorkManager

Finds a work instance.

Specified by:

find in interface WorkManager

Parameters:

workId - 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

hasWorkInState

protected boolean hasWorkInState(String workId,
 Work.State state)

Parameters:

state - SCHEDULED, RUNNING or null for both

getWorkState

public Work.State getWorkState(String workId)

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

Parameters:

workId - the id of the work to find

Returns:

the work state, or null if not found

listWork

public List<Work> listWork(String queueId,
 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

Parameters:

queueId - 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 queueId,
 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

Parameters:

queueId - 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

getMetrics

public WorkQueueMetrics getMetrics(String queueId)

Description copied from interface: WorkManager

Gets the metrics for the queueId

Specified by:

getMetrics in interface WorkManager

awaitCompletion

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

Description copied from interface: WorkManager

Waits for completion of all work.

Specified by:

awaitCompletion in interface WorkManager

Parameters:

duration - the time to wait

unit - the timeout unit

Returns:

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

Throws:

InterruptedException

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

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

getTimestampAfter

protected long getTimestampAfter(long durationInMs)

noScheduledOrRunningWork

protected boolean noScheduledOrRunningWork(String queueId)

isStarted

public boolean isStarted()

Specified by:

isStarted in interface WorkManager

Returns:

true if active

See Also:

• DefaultComponent.start(org.nuxeo.runtime.model.ComponentContext)
• WorkManager.init()
• WorkManager.shutdown(long, TimeUnit)

supportsProcessingDisabling

public boolean supportsProcessingDisabling()

Specified by:

supportsProcessingDisabling in interface WorkManager

Returns:

true if the implementation supports processing disabling