java.lang.Object
- org.nuxeo.runtime.model.DefaultComponent
- - org.nuxeo.ecm.core.work.WorkManagerImpl
  - - org.nuxeo.ecm.core.work.StreamWorkManager

All Implemented Interfaces:

WorkManager, Adaptable, Component, Extensible, TimestampedService
```
public class StreamWorkManager
extends WorkManagerImpl
```
WorkManager impl that appends works into a Log. Works are therefore immutable (no state update) and can not be listed for performance reason.

Since:

9.3

Nested Class Summary

Nested Classes
Modifier and Type Class Description

class StreamWorkManager.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	Description
`protected long`	`CACHE_LAST_METRIC_DURATION_MS`
`static String`	`COMPUTATION_FILTER_CLASS_KEY`
`static String`	`COMPUTATION_FILTER_PREFIX_KEY`
`static String`	`COMPUTATION_FILTER_STORE_KEY`
`static String`	`COMPUTATION_FILTER_STORE_TTL_KEY`
`static String`	`COMPUTATION_FILTER_THRESHOLD_SIZE_KEY`
`static int`	`DEFAULT_CONCURRENCY`
`static String`	`DEFAULT_WORK_CODEC`
`static String`	`DEFAULT_WORK_LOG_CONFIG`	Deprecated.
`static String`	`DEFAULT_WORK_OVER_PROVISIONING`
`protected WorkQueueMetrics`	`lastMetrics`
`protected long`	`lastMetricTime`
`protected static Log`	`log`
`protected LogManager`	`logManager`
`static String`	`NAMESPACE_PREFIX`
`protected Settings`	`settings`
`protected long`	`stateTTL`
`static long`	`STATETTL_DEFAULT_VALUE`
`static String`	`STATETTL_KEY`
`protected boolean`	`storeState`
`static String`	`STORESTATE_KEY`
`protected StreamManager`	`streamManager`
`protected StreamProcessor`	`streamProcessor`
`protected Topology`	`topology`
`protected Topology`	`topologyDisabled`
`static String`	`WORK_CODEC_PROP`
`static String`	`WORK_LOG_CONFIG_PROP`
`static String`	`WORK_OVER_PROVISIONING_PROP`

Fields inherited from class org.nuxeo.ecm.core.work.WorkManagerImpl
active, categoryToQueueId, completionSynchronizer, DEAD_LETTER_QUEUE, DEAD_LETTER_QUEUE_CODEC, DEFAULT_CATEGORY, DEFAULT_QUEUE_ID, executors, GLOBAL_METRIC_PREFIX, IMPL_EP, NAME, QUEUES_EP, queuing, registry, SHUTDOWN_DELAY_MS_KEY, shutdownInProgress, started, THREAD_PREFIX, WORKMANAGER_PROCESSING_DISABLE, WORKMANAGER_PROCESSING_ENABLED

Fields inherited from class org.nuxeo.runtime.model.DefaultComponent
lastModified, name

Constructor Summary

Constructors
Constructor Description

StreamWorkManager()

Method Summary

All Methods Instance Methods Concrete Methods Deprecated Methods
Modifier and Type	Method	Description
`protected void`	`activateQueueMetrics(String queueId)`
`boolean`	`awaitCompletion(String queueId, long duration, TimeUnit unit)`	Waits for completion of work in a given queue.
`protected boolean`	`awaitCompletionOnQueue(String queueId, long duration, TimeUnit unit)`
`boolean`	`awaitCompletionWithWaterMark(String queueId, long duration, TimeUnit unit)`	Deprecated. since 10.2 because unused
`protected void`	`deactivateQueueMetrics(String queueId)`
`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)`.
`protected Codec<Record>`	`getCodec()`
`protected String`	`getCodecName()`
`protected String`	`getLogConfig()`	Deprecated.
`protected LogManager`	`getLogManager()`
`protected long`	`getLowWaterMark(String queueId)`
`WorkQueueMetrics`	`getMetrics(String queueId)`	Gets the metrics for the `queueId`
`protected WorkQueueMetrics`	`getMetricsWithNuxeoClassLoader(String queueId)`
`protected 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.
`protected RecordFilterChain`	`getRecordFilter()`
`protected String`	`getRecordFilterClass()`
`protected Map<String,String>`	`getRecordFilterOptions()`
`protected StreamManager`	`getStreamManager()`
`Work.State`	`getWorkState(String workId)`	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 void`	`initTopology()`
`boolean`	`isProcessingEnabled(String queueId)`	Is processing enabled for a given queue id.
`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.
`boolean`	`supportsProcessingDisabling()`

Methods inherited from class org.nuxeo.ecm.core.work.WorkManagerImpl
awaitCompletion, enableProcessing, enableProcessing, getCategoryQueueId, getExecutor, getTimestampAfter, getWorkQueueDescriptor, getWorkQueueIds, hasWorkInState, index, initDeadLetterQueueStream, isProcessingDisabled, isProcessingEnabled, isQueuingEnabled, isStarted, noScheduledOrRunningWork, registerContribution, remainingMillis, removeExecutor, schedule, schedule, schedule, setName, shutdownExecutors

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
  - log
```
protected static final Log log
```
  - WORK_LOG_CONFIG_PROP
```
public static final String WORK_LOG_CONFIG_PROP
```
    See Also:
    
    Constant Field Values
  - DEFAULT_WORK_LOG_CONFIG
```
@Deprecated(since="2021.0")
public static final String DEFAULT_WORK_LOG_CONFIG
```
    Deprecated.
    
    See Also:
    
    Constant Field Values
  - WORK_CODEC_PROP
```
public static final String WORK_CODEC_PROP
```
    See Also:
    
    Constant Field Values
  - DEFAULT_WORK_CODEC
```
public static final String DEFAULT_WORK_CODEC
```
    See Also:
    
    Constant Field Values
  - WORK_OVER_PROVISIONING_PROP
```
public static final String WORK_OVER_PROVISIONING_PROP
```
    See Also:
    
    Constant Field Values
  - DEFAULT_WORK_OVER_PROVISIONING
```
public static final String DEFAULT_WORK_OVER_PROVISIONING
```
    See Also:
    
    Constant Field Values
  - DEFAULT_CONCURRENCY
```
public static final int DEFAULT_CONCURRENCY
```
    See Also:
    
    Constant Field Values
  - lastMetrics
```
protected WorkQueueMetrics lastMetrics
```
  - lastMetricTime
```
protected long lastMetricTime
```
  - CACHE_LAST_METRIC_DURATION_MS
```
protected long CACHE_LAST_METRIC_DURATION_MS
```
  - NAMESPACE_PREFIX
```
public static final String NAMESPACE_PREFIX
```
    See Also:
    
    Constant Field Values
  - STATETTL_KEY
```
public static final String STATETTL_KEY
```
    Since:
    
    10.2
    
    See Also:
    
    Constant Field Values
  - STORESTATE_KEY
```
public static final String STORESTATE_KEY
```
    Since:
    
    10.2
    
    See Also:
    
    Constant Field Values
  - STATETTL_DEFAULT_VALUE
```
public static final long STATETTL_DEFAULT_VALUE
```
    Since:
    
    10.2
    
    See Also:
    
    Constant Field Values
  - COMPUTATION_FILTER_CLASS_KEY
```
public static final String COMPUTATION_FILTER_CLASS_KEY
```
    Since:
    
    11.1
    
    See Also:
    
    Constant Field Values
  - COMPUTATION_FILTER_STORE_KEY
```
public static final String COMPUTATION_FILTER_STORE_KEY
```
    Since:
    
    11.1
    
    See Also:
    
    Constant Field Values
  - COMPUTATION_FILTER_STORE_TTL_KEY
```
public static final String COMPUTATION_FILTER_STORE_TTL_KEY
```
    Since:
    
    11.1
    
    See Also:
    
    Constant Field Values
  - COMPUTATION_FILTER_THRESHOLD_SIZE_KEY
```
public static final String COMPUTATION_FILTER_THRESHOLD_SIZE_KEY
```
    Since:
    
    11.1
    
    See Also:
    
    Constant Field Values
  - COMPUTATION_FILTER_PREFIX_KEY
```
public static final String COMPUTATION_FILTER_PREFIX_KEY
```
    Since:
    
    11.1
    
    See Also:
    
    Constant Field Values
  - topology
```
protected Topology topology
```
  - topologyDisabled
```
protected Topology topologyDisabled
```
  - settings
```
protected Settings settings
```
  - streamProcessor
```
protected StreamProcessor streamProcessor
```
  - logManager
```
protected LogManager logManager
```
  - streamManager
```
protected StreamManager streamManager
```
  - storeState
```
protected boolean storeState
```
  - stateTTL
```
protected long stateTTL
```
- Constructor Detail
  - StreamWorkManager
```
public StreamWorkManager()
```
- Method Detail
  - getOverProvisioningFactor
```
protected int getOverProvisioningFactor()
```
  - getCodecName
```
protected String getCodecName()
```
  - getCodec
```
protected Codec<Record> getCodec()
```
  - 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)
  - 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
  - getRecordFilter
```
protected RecordFilterChain getRecordFilter()
```
  - getRecordFilterOptions
```
protected Map<String,String> getRecordFilterOptions()
```
  - getRecordFilterClass
```
protected String getRecordFilterClass()
```
  - 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
  - getLogManager
```
protected LogManager getLogManager()
```
  - getStreamManager
```
protected StreamManager getStreamManager()
```
  - getLogConfig
```
@Deprecated(since="2021.0")
protected String getLogConfig()
```
    Deprecated.
  - 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
    
    Overrides:
    
    isProcessingEnabled in class WorkManagerImpl
  - initTopology
```
protected void initTopology()
```
  - getPartitions
```
protected int getPartitions(int maxThreads)
```
  - activateQueueMetrics
```
protected void activateQueueMetrics(String queueId)
```
  - deactivateQueueMetrics
```
protected void deactivateQueueMetrics(String queueId)
```
  - shutdownQueue
```
public boolean shutdownQueue(String queueId,
                             long timeout,
                             TimeUnit unit)
```
    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
  - shutdown
```
public boolean shutdown(long timeout,
                        TimeUnit timeUnit)
```
    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
  - 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, or null for non-completed ( SCHEDULED or RUNNING)
    
    Returns:
    
    the number of work instances in the given state
  - getMetricsWithNuxeoClassLoader
```
protected WorkQueueMetrics getMetricsWithNuxeoClassLoader(String queueId)
```
  - 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
  - awaitCompletionOnQueue
```
protected boolean awaitCompletionOnQueue(String queueId,
                                         long duration,
                                         TimeUnit unit)
                                  throws InterruptedException
```
    Throws:
    
    InterruptedException
  - awaitCompletionWithWaterMark
```
@Deprecated
public boolean awaitCompletionWithWaterMark(String queueId,
                                            long duration,
                                            TimeUnit unit)
                                     throws InterruptedException
```
    Deprecated.
    since 10.2 because unused
    
    Throws:
    
    InterruptedException
  - getLowWaterMark
```
protected long getLowWaterMark(String queueId)
```
  - 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
    
    Overrides:
    
    getWorkState in class WorkManagerImpl
    
    Parameters:
    
    workId - 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
  - supportsProcessingDisabling
```
public boolean supportsProcessingDisabling()
```
    Specified by:
    
    supportsProcessingDisabling in interface WorkManager
    
    Overrides:
    
    supportsProcessingDisabling in class WorkManagerImpl
    
    Returns:
    
    true if the implementation supports processing disabling

Class StreamWorkManager

Nested Class Summary

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

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

Field Summary

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

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

Constructor Summary

Method Summary

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

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

Methods inherited from class java.lang.Object

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

Field Detail

log

WORK_LOG_CONFIG_PROP

DEFAULT_WORK_LOG_CONFIG

WORK_CODEC_PROP

DEFAULT_WORK_CODEC

WORK_OVER_PROVISIONING_PROP

DEFAULT_WORK_OVER_PROVISIONING

DEFAULT_CONCURRENCY

lastMetrics

lastMetricTime

CACHE_LAST_METRIC_DURATION_MS

NAMESPACE_PREFIX

STATETTL_KEY

STORESTATE_KEY

STATETTL_DEFAULT_VALUE

COMPUTATION_FILTER_CLASS_KEY

COMPUTATION_FILTER_STORE_KEY

COMPUTATION_FILTER_STORE_TTL_KEY

COMPUTATION_FILTER_THRESHOLD_SIZE_KEY

COMPUTATION_FILTER_PREFIX_KEY

topology

topologyDisabled

settings

streamProcessor

logManager

streamManager

storeState

stateTTL

Constructor Detail

StreamWorkManager

Method Detail

getOverProvisioningFactor

getCodecName

getCodec

schedule

getApplicationStartedOrder

start

getRecordFilter

getRecordFilterOptions

getRecordFilterClass

init

getLogManager

getStreamManager

getLogConfig

isProcessingEnabled

initTopology

getPartitions

activateQueueMetrics

deactivateQueueMetrics

shutdownQueue

shutdown

getQueueSize

getMetricsWithNuxeoClassLoader

getMetrics

awaitCompletion

awaitCompletionOnQueue

awaitCompletionWithWaterMark

getLowWaterMark

getWorkState

find

listWork

listWorkIds

scheduleAfterCommit

supportsProcessingDisabling