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

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 and Description
static class	WorkManagerImpl.CancelingPolicy A handler for rejected tasks that discards them.
protected class	WorkManagerImpl.ShutdownListener
class	WorkManagerImpl.WorkCompletionSynchronizer
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 and Description
protected WorkManagerImpl.WorkCompletionSynchronizer	completionSynchronizer
static String	DEFAULT_CATEGORY
static String	DEFAULT_QUEUE_ID
protected Map<String,WorkManagerImpl.WorkThreadPoolExecutor>	executors
protected static String	IMPL_EP
static String	NAME
protected static String	QUEUES_EP
protected WorkQueuing	queuing
protected com.codahale.metrics.MetricRegistry	registry
protected boolean	started
protected static String	THREAD_PREFIX
protected WorkQueueDescriptorRegistry	workQueueDescriptors

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

lastModified

Constructor Summary

Constructors

Constructor and Description
WorkManagerImpl()

Method Summary

Modifier and Type	Method and Description
void	activate(ComponentContext context) Activates the component.
protected void	activateQueue(WorkQueueDescriptor workQueueDescriptor)
void	applicationStarted(ComponentContext context) Notify the component that Nuxeo Framework finished starting all Nuxeo bundles.
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	cleanup() Clears the list of completed work instances older than what's configured for each queue.
void	clearCompletedWork(long completionTime) Clears the list of completed work instances older than the given time.
void	clearCompletedWork(String queueId) Clears the list of completed work instances for a given queue.
protected void	closeQueuing()
void	deactivateQueue(WorkQueueDescriptor workQueueDescriptor)
Work	find(String workId, Work.State state) Finds a work instance.
Work	find(Work work, Work.State state, boolean useEquals, int[] pos) Deprecated.
String	findResult(String workId) Finds a work result.
int	getApplicationStartedOrder() The component notification order for Component.applicationStarted(org.nuxeo.runtime.model.ComponentContext).
String	getCategoryQueueId(String category) Gets the queue id used for a given work category.
protected int	getCompletedSize(String queueId)
protected WorkManagerImpl.WorkThreadPoolExecutor	getExecutor(String queueId)
int	getNonCompletedWorkSize(String queueId) Deprecated.
int	getQueueSize(String queueId, Work.State state) Gets the number of work instances in a given queue in a defined state.
protected int	getRunningSize(String queueId)
protected int	getScheduledOrRunningSize(String queueId)
protected int	getScheduledSize(String queueId)
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)
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 boolean	isProcessingEnabled(String queueId)
protected boolean	isQueuingEnabled(String queueId)
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 WorkQueuing	newWorkQueuing(Class<? extends WorkQueuing> klass)
void	registerContribution(Object contribution, String extensionPoint, ComponentInstance contributor)
void	registerWorkQueueDescriptor(WorkQueueDescriptor workQueueDescriptor)
void	registerWorkQueuing(WorkQueuing q)
void	registerWorkQueuingDescriptor(WorkQueuingImplDescriptor descr)
protected long	remainingMillis(long t0, long delay)
protected void	removeExecutor(String queueId)
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.
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	unregisterContribution(Object contribution, String extensionPoint, ComponentInstance contributor)
void	unregisterWorkQueing()
void	unregisterWorkQueueDescriptor(WorkQueueDescriptor workQueueDescriptor)
void	unregisterWorkQueuingDescriptor(WorkQueuingImplDescriptor descr)

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

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

Field Detail

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

registry

protected final com.codahale.metrics.MetricRegistry registry

workQueueDescriptors

protected final WorkQueueDescriptorRegistry workQueueDescriptors

executors

protected final Map<String,WorkManagerImpl.WorkThreadPoolExecutor> executors

completionSynchronizer

protected final WorkManagerImpl.WorkCompletionSynchronizer completionSynchronizer

queuing

protected WorkQueuing queuing

started

protected volatile boolean started

Constructor Detail

WorkManagerImpl

public WorkManagerImpl()

Method Detail

activate

public void activate(ComponentContext context)

Description copied from interface: Component

Activates the component.

This method is called by the runtime when a component is activated.

Specified by:

activate in interface Component

Overrides:

activate in class DefaultComponent

Parameters:

context - the runtime context

registerContribution

public void registerContribution(Object contribution,
                                 String extensionPoint,
                                 ComponentInstance contributor)

Overrides:

registerContribution in class DefaultComponent

unregisterContribution

public void unregisterContribution(Object contribution,
                                   String extensionPoint,
                                   ComponentInstance contributor)

Overrides:

unregisterContribution in class DefaultComponent

registerWorkQueueDescriptor

public void registerWorkQueueDescriptor(WorkQueueDescriptor workQueueDescriptor)

unregisterWorkQueueDescriptor

public void unregisterWorkQueueDescriptor(WorkQueueDescriptor workQueueDescriptor)

activateQueue

protected void activateQueue(WorkQueueDescriptor workQueueDescriptor)

deactivateQueue

public void deactivateQueue(WorkQueueDescriptor workQueueDescriptor)

registerWorkQueuingDescriptor

public void registerWorkQueuingDescriptor(WorkQueuingImplDescriptor descr)

registerWorkQueuing

public void registerWorkQueuing(WorkQueuing q)

unregisterWorkQueuingDescriptor

public void unregisterWorkQueuingDescriptor(WorkQueuingImplDescriptor descr)

unregisterWorkQueing

public void unregisterWorkQueing()

newWorkQueuing

protected WorkQueuing newWorkQueuing(Class<? extends WorkQueuing> klass)

closeQueuing

protected void closeQueuing()

isQueuingEnabled

protected boolean isQueuingEnabled(String queueId)

isProcessingEnabled

protected boolean isProcessingEnabled(String queueId)

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.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 DefaultComponent

Returns:

the order, 1000 by default

applicationStarted

public void applicationStarted(ComponentContext context)

Description copied from interface: Component

Notify the component that Nuxeo Framework finished starting all Nuxeo bundles.

Specified by:

applicationStarted in interface Component

Overrides:

applicationStarted in class DefaultComponent

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

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

protected long remainingMillis(long t0,
                               long delay)

removeExecutor

protected void removeExecutor(String queueId)

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

@Deprecated
public Work find(Work work,
                             Work.State state,
                             boolean useEquals,
                             int[] pos)

Deprecated.

Description copied from interface: WorkManager

Finds a work instance.

Specified by:

find in interface WorkManager

Parameters:

work - the work to find

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

useEquals - ignored, always uses work id equality

pos - ignored, pass null

Returns:

the found work instance, or null if not found

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, COMPLETED, or null for non-completed

Returns:

the found work instance, or null if not found

findResult

public String findResult(String workId)

Description copied from interface: WorkManager

Finds a work result.

Specified by:

findResult in interface WorkManager

Parameters:

workId - the id of the work to find the result

Returns:

the found work result, or null if there is no result or if work is not COMPLETED

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, COMPLETED, Work.State.CANCELED or Work.State.FAILED.

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, COMPLETED, 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, COMPLETED, or null for non-completed

Returns:

the list of work ids in the given state

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

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

getNonCompletedWorkSize

@Deprecated
public int getNonCompletedWorkSize(String queueId)

Deprecated.

Description copied from interface: WorkManager

Gets the size of the non-completed work (scheduled + running) for a give queue.

Specified by:

getNonCompletedWorkSize in interface WorkManager

Parameters:

queueId - the queue id

Returns:

the number of non-completed work instances

getScheduledSize

protected int getScheduledSize(String queueId)

getRunningSize

protected int getRunningSize(String queueId)

getScheduledOrRunningSize

protected int getScheduledOrRunningSize(String queueId)

getCompletedSize

protected int getCompletedSize(String queueId)

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

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

clearCompletedWork

public void clearCompletedWork(String queueId)

Description copied from interface: WorkManager

Clears the list of completed work instances for a given queue.

Specified by:

clearCompletedWork in interface WorkManager

Parameters:

queueId - the queue id

clearCompletedWork

public void clearCompletedWork(long completionTime)

Description copied from interface: WorkManager

Clears the list of completed work instances older than the given time.

Specified by:

clearCompletedWork in interface WorkManager

Parameters:

completionTime - the completion time (milliseconds since epoch) before which completed work instances are cleared, or 0 for all

cleanup

public void cleanup()

Description copied from interface: WorkManager

Clears the list of completed work instances older than what's configured for each queue.

Specified by:

cleanup in interface WorkManager

isStarted

public boolean isStarted()

Specified by:

isStarted in interface WorkManager

Returns:

true if active

See Also:

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