Package org.nuxeo.ecm.core.work

Class StreamWorkManager

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

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.start(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)
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

Field Details

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 Details

StreamWorkManager

public StreamWorkManager()

Method Details

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

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

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