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 aWorkManager
. This delegates the queuing implementation to aWorkQueuing
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
AThreadPoolExecutor
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 Map<String,String>
categoryToQueueId
protected WorkManagerImpl.WorkCompletionSynchronizer
completionSynchronizer
static Name
DEAD_LETTER_QUEUE
The dead letter queue stream name.static Codec<Record>
DEAD_LETTER_QUEUE_CODEC
static String
DEFAULT_CATEGORY
static String
DEFAULT_QUEUE_ID
protected Map<String,WorkManagerImpl.WorkThreadPoolExecutor>
executors
protected static String
GLOBAL_METRIC_PREFIX
protected static String
IMPL_EP
static String
NAME
protected static String
QUEUES_EP
protected WorkQueuing
queuing
protected io.dropwizard.metrics5.MetricRegistry
registry
static String
SHUTDOWN_DELAY_MS_KEY
protected boolean
shutdownInProgress
protected boolean
started
protected static String
THREAD_PREFIX
static String
WORKMANAGER_PROCESSING_DISABLE
Deprecated.UseWORKMANAGER_PROCESSING_ENABLED
insteadstatic String
WORKMANAGER_PROCESSING_ENABLED
-
Fields inherited from class org.nuxeo.runtime.model.DefaultComponent
lastModified, name
-
-
Constructor Summary
Constructors Constructor Description WorkManagerImpl()
-
Method Summary
All Methods Instance Methods Concrete Methods Deprecated Methods 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 queuesvoid
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 forComponent.applicationStarted(org.nuxeo.runtime.model.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 thequeueId
int
getQueueSize(String queueId, Work.State state)
Gets the number of work instances in a given queue in a defined state.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 thisWorkManager
and attempts to resume work previously suspended and saved atWorkManager.shutdown(long, java.util.concurrent.TimeUnit)
time.protected void
initDeadLetterQueueStream()
protected boolean
isProcessingDisabled()
boolean
isProcessingEnabled()
Is processing enabled for at least one queueboolean
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 unusedprotected void
removeExecutor(String queueId)
Deprecated.since 10.2 because unusedvoid
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 thisWorkManager
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
-
Methods inherited from interface org.nuxeo.runtime.model.Component
applicationStarted
-
-
-
-
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
-
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.UseWORKMANAGER_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
-
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
-
queuing
protected WorkQueuing queuing
-
active
protected boolean active
-
completionSynchronizer
protected WorkManagerImpl.WorkCompletionSynchronizer completionSynchronizer
-
started
protected volatile boolean started
-
shutdownInProgress
protected volatile boolean shutdownInProgress
-
-
Method Detail
-
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 interfaceComponent
- Overrides:
setName
in classDefaultComponent
- Parameters:
name
- the name
-
registerContribution
public void registerContribution(Object contribution, String xp, ComponentInstance component)
- Overrides:
registerContribution
in classDefaultComponent
-
isQueuingEnabled
public boolean isQueuingEnabled(String queueId)
Description copied from interface:WorkManager
Is queuing enabled for a given queue id.- Specified by:
isQueuingEnabled
in interfaceWorkManager
-
enableProcessing
public void enableProcessing(boolean value)
Description copied from interface:WorkManager
Set processing for all queues- Specified by:
enableProcessing
in interfaceWorkManager
-
enableProcessing
public void enableProcessing(String queueId, boolean value)
Description copied from interface:WorkManager
Set processing for a given queue id.- Specified by:
enableProcessing
in interfaceWorkManager
-
isProcessingEnabled
public boolean isProcessingEnabled()
Description copied from interface:WorkManager
Is processing enabled for at least one queue- Specified by:
isProcessingEnabled
in interfaceWorkManager
-
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 interfaceWorkManager
-
getWorkQueueIds
public List<String> getWorkQueueIds()
Description copied from interface:WorkManager
Lists the ids of the existing work queues.- Specified by:
getWorkQueueIds
in interfaceWorkManager
- 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 interfaceWorkManager
- 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 interfaceWorkManager
- Parameters:
category
- the category- Returns:
- the queue id
-
getApplicationStartedOrder
public int getApplicationStartedOrder()
Description copied from interface:Component
The component notification order forComponent.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 interfaceComponent
- 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 interfaceComponent
- Overrides:
start
in classDefaultComponent
-
initDeadLetterQueueStream
protected void initDeadLetterQueueStream()
-
init
public void init()
Description copied from interface:WorkManager
Starts up thisWorkManager
and attempts to resume work previously suspended and saved atWorkManager.shutdown(long, java.util.concurrent.TimeUnit)
time.- Specified by:
init
in interfaceWorkManager
-
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 interfaceWorkManager
- Parameters:
queueId
- the queue idtimeout
- the time to waitunit
- 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 thisWorkManager
and attempts to suspend and save the running and scheduled work instances.- Specified by:
shutdown
in interfaceWorkManager
- Parameters:
timeout
- the time to waitunit
- 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)
withafterCommit = false
.- Specified by:
schedule
in interfaceWorkManager
- 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 interfaceWorkManager
- Parameters:
work
- the work to executeafterCommit
- iftrue
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)
withafterCommit = false
.- Specified by:
schedule
in interfaceWorkManager
- Parameters:
work
- the work to executescheduling
- 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 interfaceWorkManager
- Parameters:
work
- the work to executescheduling
- the scheduling policyafterCommit
- iftrue
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. Returnsfalse
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 interfaceWorkManager
- Parameters:
workId
- the id of the work to findstate
- the state defining the state to look into,SCHEDULED
,RUNNING
, ornull
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
- Specified by:
getWorkState
in interfaceWorkManager
- 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 interfaceWorkManager
- Parameters:
queueId
- the queue idstate
- the state defining the state to look into,SCHEDULED
,RUNNING
, ornull
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 interfaceWorkManager
- Parameters:
queueId
- the queue idstate
- the state defining the state to look into,SCHEDULED
,RUNNING
, ornull
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 thequeueId
- Specified by:
getMetrics
in interfaceWorkManager
-
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 interfaceWorkManager
- Parameters:
queueId
- the queue idstate
- the state defining the state to look into,SCHEDULED
,RUNNING
, ornull
for non-completed (SCHEDULED
orRUNNING
)- Returns:
- the number of work instances in the given state
-
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 interfaceWorkManager
- Parameters:
duration
- the time to waitunit
- the timeout unit- Returns:
true
if all work completed, orfalse
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 interfaceWorkManager
- Parameters:
queueId
- the queue idduration
- the time to waitunit
- the timeout unit- Returns:
true
if all work completed in the queue, orfalse
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 interfaceWorkManager
- Returns:
true
if active- See Also:
Component.applicationStarted(org.nuxeo.runtime.model.ComponentContext)
,WorkManager.init()
,WorkManager.shutdown(long, TimeUnit)
-
supportsProcessingDisabling
public boolean supportsProcessingDisabling()
- Specified by:
supportsProcessingDisabling
in interfaceWorkManager
- Returns:
- true if the implementation supports processing disabling
-
-