RedisWorkQueuing (Nuxeo ECM Projects 8.2 API)

java.lang.Object
- org.nuxeo.ecm.core.redis.contribs.RedisWorkQueuing

All Implemented Interfaces:

WorkQueuing
```
public class RedisWorkQueuing
extends Object
implements WorkQueuing
```
Implementation of a WorkQueuing storing Work instances in Redis.

Since:

5.8

Nested Class Summary

Nested Classes
Modifier and Type	Class and Description
`static class`	`RedisWorkQueuing.SScanner` Helper to call SSCAN but fall back on a custom implementation based on SMEMBERS if the backend (embedded) does not support SSCAN.

Field Summary

Fields
Modifier and Type	Field and Description
`protected Map<String,BlockingQueue<Runnable>>`	`allQueued`
`protected String`	`cleanCompletedWorkSha`
`protected String`	`completedWorkSha`
`protected static String`	`KEY_COMPLETED_PREFIX` Per-queue set of completed Work instance ids.
`protected static String`	`KEY_DATA` Global hash of Work instance id -> serialized Work instance.
`protected static String`	`KEY_QUEUE_PREFIX` Per-queue list of scheduled Work instance ids.
`protected static String`	`KEY_RUNNING_PREFIX` Per-queue set of running Work instance ids.
`protected static String`	`KEY_SCHEDULED_PREFIX` Per-queue set of scheduled Work instance ids.
`protected static String`	`KEY_STATE` Global hash of Work instance id -> Work state.
`protected static String`	`KEY_SUSPENDED_PREFIX` Per-queue list of suspended Work instance ids.
`protected WorkManagerImpl`	`mgr`
`protected RedisAdmin`	`redisAdmin`
`protected RedisExecutor`	`redisExecutor`
`protected String`	`redisNamespace`
`protected String`	`runningWorkSha`
`protected String`	`schedulingWorkSha`
`protected static byte[]`	`STATE_CANCELED`
`protected static byte`	`STATE_CANCELED_B`
`protected static byte[]`	`STATE_COMPLETED`
`protected static byte`	`STATE_COMPLETED_B`
`protected static byte[]`	`STATE_RUNNING`
`protected static byte`	`STATE_RUNNING_B`
`protected static byte[]`	`STATE_SCHEDULED`
`protected static byte`	`STATE_SCHEDULED_B`
`protected static String`	`UTF_8`

Constructor Summary

Constructors
Constructor and Description

RedisWorkQueuing(WorkManagerImpl mgr, WorkQueueDescriptorRegistry workQueueDescriptors)

Constructors
Constructor and Description
`RedisWorkQueuing(WorkManagerImpl mgr, WorkQueueDescriptorRegistry workQueueDescriptors)`

Method Summary

All Methods Instance Methods Concrete Methods
Modifier and Type	Method and Description
`void`	`addScheduledWork(String queueId, Work work)` Persists a work instance and adds it to the scheduled queue.
`protected byte[]`	`bytes(String string)`
`void`	`clearCompletedWork(String queueId, long completionTime)` Clears the list of completed work instances older than the given time in the given queue.
`protected byte[]`	`completedKey(String queueId)`
`protected String`	`completedKeyString(String queueId)`
`int`	`count(String queueId, Work.State state)` Gets the number of work instances in the given state in a given queue.
`protected byte[]`	`dataKey()`
`protected String`	`dataKeyString()`
`protected Work`	`deserializeWork(byte[] workBytes)`
`Work`	`find(String workId, Work.State state)` Finds a work instance in the scheduled or running or completed sets.
`Set<String>`	`getCompletedQueueIds()` Finds which queues have completed work.
`protected int`	`getCompletedSize(String queueId)`
`protected Set<String>`	`getQueueIds(String queuePrefix)` Finds which queues have work for a given state prefix.
`protected int`	`getRedisListSize(byte[] key)`
`protected int`	`getRedisSetSize(byte[] key)`
`protected Set<String>`	`getRunningQueueIds()`
`protected int`	`getRunningSize(String queueId)`
`protected Set<String>`	`getScheduledQueueIds()`
`protected int`	`getScheduledSize(String queueId)`
`protected Set<String>`	`getSuspendedQueueIds()` Finds which queues have suspended work.
`protected Work`	`getWork(byte[] workIdBytes)`
`protected Work`	`getWorkData(byte[] workIdBytes)`
`protected Work`	`getWorkFromQueue(String queueId)` Removes first work from work queue.
`BlockingQueue<Runnable>`	`getWorkQueue(String queueId)`
`Work.State`	`getWorkState(String workId)` Gets the state in which a work instance is.
`protected Work.State`	`getWorkStateInfo(String workId)` Gets the work state.
`void`	`init()` Starts up this `WorkQueuing` and attempts to resume work previously suspended and saved at shutdown time.
`BlockingQueue<Runnable>`	`initWorkQueue(String queueId)` Creates a blocking queue of work used by the `ThreadPoolExecutor`.
`boolean`	`isWorkInState(String workId, Work.State state)` Checks if a work instance with the given id is in the given state.
`protected byte[]`	`keyBytes(String prefix)`
`protected byte[]`	`keyBytes(String prefix, String queueId)`
`protected List<Work>`	`listCompleted(String queueId)`
`protected List<String>`	`listCompletedIds(String queueId)`
`protected List<String>`	`listNonCompletedIds(String queueId)`
`protected List<Work>`	`listRunning(String queueId)`
`protected List<String>`	`listRunningIds(String queueId)`
`protected List<Work>`	`listScheduled(String queueId)`
`protected List<String>`	`listScheduledIds(String queueId)`
`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 List<String>`	`listWorkIdsList(byte[] queueBytes)`
`protected List<String>`	`listWorkIdsSet(byte[] queueBytes)`
`protected List<Work>`	`listWorkList(byte[] queueBytes)`
`protected List<Work>`	`listWorkSet(byte[] queueBytes)`
`protected BlockingQueue<Runnable>`	`newBlockingQueue(String queueId)`
`protected byte[]`	`queuedKey(String queueId)`
`protected void`	`removeAllCompletedWork(String queueId)`
`protected void`	`removeCompletedWork(String queueId, long completionTime)`
`Work`	`removeScheduled(String queueId, String workId)` Finds a scheduled work instance and removes it from the scheduled queue.
`protected Work`	`removeScheduledWork(String queueId, String workId)` Removes a given work from queue, move the work from scheduled to completed set.
`protected byte[]`	`runningKey(String queueId)`
`protected byte[]`	`scheduledKey(String queueId)`
`int`	`scheduleSuspendedWork(String queueId)` Resumes all suspended work instances by moving them to the scheduled queue.
`protected byte[]`	`serializeWork(Work work)`
`int`	`setSuspending(String queueId)` Notifies this queuing that all work should be suspending.
`protected byte[]`	`stateKey()`
`protected String`	`stateKeyString()`
`protected String`	`string(byte[] bytes)`
`protected byte[]`	`suspendedKey(String queueId)`
`int`	`suspendScheduledWork(String queueId)` Suspends all scheduled work instances by moving them to the suspended queue.
`void`	`workCompleted(String queueId, Work work)` Moves a work instance from the running set to the completed set.
`void`	`workRunning(String queueId, Work work)` Put the work instance into the running set.
`boolean`	`workSchedule(String queueId, Work work)` Submit a work to the `ThreadPoolExecutor` and put it in the scheduled set.
`protected void`	`workSetCompleted(String queueId, Work work)` Switches a work to state completed, and saves its new state.
`protected void`	`workSetRunning(String queueId, Work work)` Switches a work to state running.

Methods inherited from class java.lang.Object
clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

- Field Detail
  - UTF_8
```
protected static final String UTF_8
```
    See Also:
    
    Constant Field Values
  - KEY_DATA
```
protected static final String KEY_DATA
```
    Global hash of Work instance id -> serialized Work instance.
    
    See Also:
    
    Constant Field Values
  - KEY_STATE
```
protected static final String KEY_STATE
```
    Global hash of Work instance id -> Work state. The completed state ( 67) is followed by a completion time in milliseconds.
    
    See Also:
    
    Constant Field Values
  - KEY_SUSPENDED_PREFIX
```
protected static final String KEY_SUSPENDED_PREFIX
```
    Per-queue list of suspended Work instance ids.
    
    See Also:
    
    Constant Field Values
  - KEY_QUEUE_PREFIX
```
protected static final String KEY_QUEUE_PREFIX
```
    Per-queue list of scheduled Work instance ids.
    
    See Also:
    
    Constant Field Values
  - KEY_SCHEDULED_PREFIX
```
protected static final String KEY_SCHEDULED_PREFIX
```
    Per-queue set of scheduled Work instance ids.
    
    See Also:
    
    Constant Field Values
  - KEY_RUNNING_PREFIX
```
protected static final String KEY_RUNNING_PREFIX
```
    Per-queue set of running Work instance ids.
    
    See Also:
    
    Constant Field Values
  - KEY_COMPLETED_PREFIX
```
protected static final String KEY_COMPLETED_PREFIX
```
    Per-queue set of completed Work instance ids.
    
    See Also:
    
    Constant Field Values
  - STATE_SCHEDULED_B
```
protected static final byte STATE_SCHEDULED_B
```
    See Also:
    
    Constant Field Values
  - STATE_CANCELED_B
```
protected static final byte STATE_CANCELED_B
```
    See Also:
    
    Constant Field Values
  - STATE_RUNNING_B
```
protected static final byte STATE_RUNNING_B
```
    See Also:
    
    Constant Field Values
  - STATE_COMPLETED_B
```
protected static final byte STATE_COMPLETED_B
```
    See Also:
    
    Constant Field Values
  - STATE_SCHEDULED
```
protected static final byte[] STATE_SCHEDULED
```
  - STATE_CANCELED
```
protected static final byte[] STATE_CANCELED
```
  - STATE_RUNNING
```
protected static final byte[] STATE_RUNNING
```
  - STATE_COMPLETED
```
protected static final byte[] STATE_COMPLETED
```
  - mgr
```
protected final WorkManagerImpl mgr
```
  - allQueued
```
protected Map<String,BlockingQueue<Runnable>> allQueued
```
  - redisExecutor
```
protected RedisExecutor redisExecutor
```
  - redisAdmin
```
protected RedisAdmin redisAdmin
```
  - redisNamespace
```
protected String redisNamespace
```
  - schedulingWorkSha
```
protected String schedulingWorkSha
```
  - runningWorkSha
```
protected String runningWorkSha
```
  - completedWorkSha
```
protected String completedWorkSha
```
  - cleanCompletedWorkSha
```
protected String cleanCompletedWorkSha
```
- Constructor Detail
  - RedisWorkQueuing
```
public RedisWorkQueuing(WorkManagerImpl mgr,
                        WorkQueueDescriptorRegistry workQueueDescriptors)
```
- Method Detail
  - init
```
public void init()
```
    Description copied from interface: WorkQueuing
    
    Starts up this WorkQueuing and attempts to resume work previously suspended and saved at shutdown time.
    
    Specified by:
    
    init in interface WorkQueuing
  - initWorkQueue
```
public BlockingQueue<Runnable> initWorkQueue(String queueId)
```
    Description copied from interface: WorkQueuing
    
    Creates a blocking queue of work used by the ThreadPoolExecutor.
    
    Specified by:
    
    initWorkQueue in interface WorkQueuing
    
    Returns:
  - workSchedule
```
public boolean workSchedule(String queueId,
                            Work work)
```
    Description copied from interface: WorkQueuing
    
    Submit a work to the ThreadPoolExecutor and put it in the scheduled set.
    
    Specified by:
    
    workSchedule in interface WorkQueuing
    
    Parameters:
    
    queueId - the queue id
    
    work - the work instance
  - getWorkQueue
```
public BlockingQueue<Runnable> getWorkQueue(String queueId)
```
  - workRunning
```
public void workRunning(String queueId,
                        Work work)
```
    Description copied from interface: WorkQueuing
    
    Put the work instance into the running set.
    
    Specified by:
    
    workRunning in interface WorkQueuing
    
    Parameters:
    
    queueId - the queue id
    
    work - the work instance
  - workCompleted
```
public void workCompleted(String queueId,
                          Work work)
```
    Description copied from interface: WorkQueuing
    
    Moves a work instance from the running set to the completed set.
    
    Specified by:
    
    workCompleted in interface WorkQueuing
    
    Parameters:
    
    queueId - the queue id
    
    work - the work instance
  - newBlockingQueue
```
protected BlockingQueue<Runnable> newBlockingQueue(String queueId)
```
  - listWork
```
public List<Work> listWork(String queueId,
                           Work.State state)
```
    Description copied from interface: WorkQueuing
    
    Lists the work instances in a given queue in a defined state.
    Note that an instance requested as RUNNING could be found SUSPENDING or SUSPENDED, and an instance requested as COMPLETED could be found FAILED.
    
    Specified by:
    
    listWork in interface WorkQueuing
    
    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: WorkQueuing
    
    Lists the work ids in a given queue in a defined state.
    
    Specified by:
    
    listWorkIds in interface WorkQueuing
    
    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
  - listScheduled
```
protected List<Work> listScheduled(String queueId)
```
  - listRunning
```
protected List<Work> listRunning(String queueId)
```
  - listCompleted
```
protected List<Work> listCompleted(String queueId)
```
  - listScheduledIds
```
protected List<String> listScheduledIds(String queueId)
```
  - listRunningIds
```
protected List<String> listRunningIds(String queueId)
```
  - listNonCompletedIds
```
protected List<String> listNonCompletedIds(String queueId)
```
  - listCompletedIds
```
protected List<String> listCompletedIds(String queueId)
```
  - count
```
public int count(String queueId,
                 Work.State state)
```
    Description copied from interface: WorkQueuing
    
    Gets the number of work instances in the given state in a given queue.
    
    Specified by:
    
    count in interface WorkQueuing
    
    Parameters:
    
    queueId - the queue id
    
    state - the state, SCHEDULED, RUNNING or COMPLETED
    
    Returns:
    
    the number of scheduled work instances in the queue
  - getScheduledSize
```
protected int getScheduledSize(String queueId)
```
  - getRunningSize
```
protected int getRunningSize(String queueId)
```
  - getCompletedSize
```
protected int getCompletedSize(String queueId)
```
  - find
```
public Work find(String workId,
                 Work.State state)
```
    Description copied from interface: WorkQueuing
    
    Finds a work instance in the scheduled or running or completed sets.
    
    Specified by:
    
    find in interface WorkQueuing
    
    Parameters:
    
    workId - the id of the work to find
    
    state - the state defining the state to look into, SCHEDULED, RUNNING, COMPLETED, or null for SCHEDULED or RUNNING
    
    Returns:
    
    the found work instance, or null if not found
  - isWorkInState
```
public boolean isWorkInState(String workId,
                             Work.State state)
```
    Description copied from interface: WorkQueuing
    
    Checks if a work instance with the given id is in the given state.
    
    Specified by:
    
    isWorkInState in interface WorkQueuing
    
    Parameters:
    
    workId - the work id
    
    state - the state, SCHEDULED, RUNNING, COMPLETED, or null for non-completed
    
    Returns:
    
    true if a work instance with the given id is in the given state
  - removeScheduled
```
public Work removeScheduled(String queueId,
                            String workId)
```
    Description copied from interface: WorkQueuing
    
    Finds a scheduled work instance and removes it from the scheduled queue.
    
    Specified by:
    
    removeScheduled in interface WorkQueuing
    
    Parameters:
    
    queueId - the queue id
    
    workId - the id of the work to find
    
    Returns:
    
    the work if found, otherwise null
  - getWorkState
```
public Work.State getWorkState(String workId)
```
    Description copied from interface: WorkQueuing
    
    Gets the state in which a work instance is.
    This can be Work.State.SCHEDULED, Work.State.RUNNING, Work.State.COMPLETED, Work.State.FAILED, or Work.State.CANCELED.
    
    Specified by:
    
    getWorkState in interface WorkQueuing
    
    Parameters:
    
    workId - the id of the work to find
    
    Returns:
    
    the work state, or null if not found
  - setSuspending
```
public int setSuspending(String queueId)
```
    Description copied from interface: WorkQueuing
    
    Notifies this queuing that all work should be suspending.
    
    Specified by:
    
    setSuspending in interface WorkQueuing
    
    Returns:
    
    the number of scheduled instances removed from queue
  - clearCompletedWork
```
public void clearCompletedWork(String queueId,
                               long completionTime)
```
    Description copied from interface: WorkQueuing
    
    Clears the list of completed work instances older than the given time in the given queue.
    
    Specified by:
    
    clearCompletedWork in interface WorkQueuing
    
    Parameters:
    
    queueId - the queue id
    
    completionTime - the completion time (milliseconds since epoch) before which completed work instances are cleared, or 0 for all
  - string
```
protected String string(byte[] bytes)
```
  - bytes
```
protected byte[] bytes(String string)
```
  - keyBytes
```
protected byte[] keyBytes(String prefix,
                          String queueId)
```
  - keyBytes
```
protected byte[] keyBytes(String prefix)
```
  - suspendedKey
```
protected byte[] suspendedKey(String queueId)
```
  - queuedKey
```
protected byte[] queuedKey(String queueId)
```
  - runningKey
```
protected byte[] runningKey(String queueId)
```
  - scheduledKey
```
protected byte[] scheduledKey(String queueId)
```
  - completedKey
```
protected byte[] completedKey(String queueId)
```
  - completedKeyString
```
protected String completedKeyString(String queueId)
```
  - stateKey
```
protected byte[] stateKey()
```
  - stateKeyString
```
protected String stateKeyString()
```
  - dataKey
```
protected byte[] dataKey()
```
  - dataKeyString
```
protected String dataKeyString()
```
  - serializeWork
```
protected byte[] serializeWork(Work work)
                        throws IOException
```
    Throws:
    
    IOException
  - deserializeWork
```
protected Work deserializeWork(byte[] workBytes)
```
  - getRedisListSize
```
protected int getRedisListSize(byte[] key)
                        throws IOException
```
    Throws:
    
    IOException
  - getRedisSetSize
```
protected int getRedisSetSize(byte[] key)
                       throws IOException
```
    Throws:
    
    IOException
  - addScheduledWork
```
public void addScheduledWork(String queueId,
                             Work work)
                      throws IOException
```
    Persists a work instance and adds it to the scheduled queue.
    
    Parameters:
    
    queueId - the queue id
    
    work - the work instance
    
    Throws:
    
    IOException
  - getSuspendedQueueIds
```
protected Set<String> getSuspendedQueueIds()
                                    throws IOException
```
    Finds which queues have suspended work.
    
    Returns:
    
    a set of queue ids
    
    Throws:
    
    IOException
    
    Since:
    
    5.8
  - getScheduledQueueIds
```
protected Set<String> getScheduledQueueIds()
```
  - getRunningQueueIds
```
protected Set<String> getRunningQueueIds()
```
  - getCompletedQueueIds
```
public Set<String> getCompletedQueueIds()
```
    Description copied from interface: WorkQueuing
    
    Finds which queues have completed work.
    
    Specified by:
    
    getCompletedQueueIds in interface WorkQueuing
    
    Returns:
    
    a set of queue ids
  - getQueueIds
```
protected Set<String> getQueueIds(String queuePrefix)
```
    Finds which queues have work for a given state prefix.
    
    Returns:
    
    a set of queue ids
    
    Since:
    
    5.8
  - scheduleSuspendedWork
```
public int scheduleSuspendedWork(String queueId)
                          throws IOException
```
    Resumes all suspended work instances by moving them to the scheduled queue.
    
    Parameters:
    
    queueId - the queue id
    
    Returns:
    
    the number of work instances scheduled
    
    Throws:
    
    IOException
    
    Since:
    
    5.8
  - suspendScheduledWork
```
public int suspendScheduledWork(String queueId)
                         throws IOException
```
    Suspends all scheduled work instances by moving them to the suspended queue.
    
    Parameters:
    
    queueId - the queue id
    
    Returns:
    
    the number of work instances suspended
    
    Throws:
    
    IOException
    
    Since:
    
    5.8
  - workSetRunning
```
protected void workSetRunning(String queueId,
                              Work work)
                       throws IOException
```
    Switches a work to state running.
    
    Parameters:
    
    queueId - the queue id
    
    work - the work
    
    Throws:
    
    IOException
  - workSetCompleted
```
protected void workSetCompleted(String queueId,
                                Work work)
                         throws IOException
```
    Switches a work to state completed, and saves its new state.
    
    Throws:
    
    IOException
  - getWorkStateInfo
```
protected Work.State getWorkStateInfo(String workId)
```
    Gets the work state.
    
    Parameters:
    
    workId - the work id
    
    Returns:
    
    the state, or null if not found
  - listWorkIdsList
```
protected List<String> listWorkIdsList(byte[] queueBytes)
                                throws IOException
```
    Throws:
    
    IOException
  - listWorkIdsSet
```
protected List<String> listWorkIdsSet(byte[] queueBytes)
                               throws IOException
```
    Throws:
    
    IOException
  - listWorkList
```
protected List<Work> listWorkList(byte[] queueBytes)
                           throws IOException
```
    Throws:
    
    IOException
  - listWorkSet
```
protected List<Work> listWorkSet(byte[] queueBytes)
                          throws IOException
```
    Throws:
    
    IOException
  - getWork
```
protected Work getWork(byte[] workIdBytes)
```
  - getWorkData
```
protected Work getWorkData(byte[] workIdBytes)
                    throws IOException
```
    Throws:
    
    IOException
  - getWorkFromQueue
```
protected Work getWorkFromQueue(String queueId)
                         throws IOException
```
    Removes first work from work queue.
    
    Parameters:
    
    queueId - the queue id
    
    Returns:
    
    the work, or null if the scheduled queue is empty
    
    Throws:
    
    IOException
  - removeScheduledWork
```
protected Work removeScheduledWork(String queueId,
                                   String workId)
                            throws IOException
```
    Removes a given work from queue, move the work from scheduled to completed set.
    
    Throws:
    
    IOException
  - removeAllCompletedWork
```
protected void removeAllCompletedWork(String queueId)
                               throws IOException
```
    Throws:
    
    IOException
  - removeCompletedWork
```
protected void removeCompletedWork(String queueId,
                                   long completionTime)
                            throws IOException
```
    Throws:
    
    IOException

Class RedisWorkQueuing

Nested Class Summary

Field Summary

Constructor Summary

Method Summary

Methods inherited from class java.lang.Object

Field Detail

UTF_8

KEY_DATA

KEY_STATE

KEY_SUSPENDED_PREFIX

KEY_QUEUE_PREFIX

KEY_SCHEDULED_PREFIX

KEY_RUNNING_PREFIX

KEY_COMPLETED_PREFIX

STATE_SCHEDULED_B

STATE_CANCELED_B

STATE_RUNNING_B

STATE_COMPLETED_B

STATE_SCHEDULED

STATE_CANCELED

STATE_RUNNING

STATE_COMPLETED

mgr

allQueued

redisExecutor

redisAdmin

redisNamespace

schedulingWorkSha

runningWorkSha

completedWorkSha

cleanCompletedWorkSha

Constructor Detail

RedisWorkQueuing

Method Detail

init

initWorkQueue

workSchedule

getWorkQueue

workRunning

workCompleted

newBlockingQueue

listWork

listWorkIds

listScheduled

listRunning

listCompleted

listScheduledIds

listRunningIds

listNonCompletedIds

listCompletedIds

count

getScheduledSize

getRunningSize

getCompletedSize

find

isWorkInState

removeScheduled

getWorkState

setSuspending

clearCompletedWork

string

bytes

keyBytes

keyBytes

suspendedKey

queuedKey

runningKey

scheduledKey

completedKey

completedKeyString

stateKey

stateKeyString

dataKey

dataKeyString

serializeWork

deserializeWork

getRedisListSize

getRedisSetSize

addScheduledWork