Class EventQueue<S, E>

Type Parameters

S extends Record<string, any>
E extends Events<S> | RegExp = Events<S>

Index

Constructors

constructor

new EventQueue<
    S extends Record<string, any>,
    E extends string | number | symbol | RegExp = keyof S,
>(
    opts: QueueOpts & {
        event: RegExp | E;
        observer?: ObserverEngine<S>;
        process: (data: EventData<S, E>) => Promise<void>;
    },
): EventQueue<S, E>
Type Parameters
- S extends Record<string, any>
- E extends string | number | symbol | RegExp = keyof S
Parameters
- opts: QueueOpts & {
      event: RegExp | E;
      observer?: ObserverEngine<S>;
      process: (data: EventData<S, E>) => Promise<void>;
  }
Returns EventQueue<S, E>
- Defined in packages/observer/dist/types/queue/queue.d.ts:89

Accessors

isDraining

get isDraining(): boolean
Whether the queue is draining

Returns boolean
- Defined in packages/observer/dist/types/queue/queue.d.ts:174

isIdle

get isIdle(): boolean
Whether the queue is idle

Returns boolean
- Defined in packages/observer/dist/types/queue/queue.d.ts:158

isPaused

get isPaused(): boolean
Whether the queue is paused

Returns boolean
- Defined in packages/observer/dist/types/queue/queue.d.ts:166

isRunning

get isRunning(): boolean
Whether the queue is running

Returns boolean
- Defined in packages/observer/dist/types/queue/queue.d.ts:162

isStopped

get isStopped(): boolean
Whether the queue is stopped

Returns boolean
- Defined in packages/observer/dist/types/queue/queue.d.ts:170

isWaiting

get isWaiting(): boolean
Whether the queue is waiting for the the

Returns boolean
- Defined in packages/observer/dist/types/queue/queue.d.ts:154

name

get name(): string
The name of the queue

Returns string
- Defined in packages/observer/dist/types/queue/queue.d.ts:149

pending

get pending(): number
The number of items in the queue

Returns number
- Defined in packages/observer/dist/types/queue/queue.d.ts:182

snapshot

get snapshot(): {
    activeRunners: number;
    isDraining: boolean;
    isEmpty: boolean;
    isIdle: boolean;
    isPaused: boolean;
    isRunning: boolean;
    isStopped: boolean;
    isWaiting: boolean;
    name: string;
    pending: number;
    rateLimiter: {
        averageWaitTime: number;
        capacity: number;
        createdAt: number;
        currentTokens: number;
        refillIntervalMs: number;
        rejectedRequests: number;
        rejectionRate: number;
        successfulRequests: number;
        totalRequests: number;
        totalWaitTime: number;
        uptime: number;
        waitCount: number;
    };
    runningNodes: Set<`${number}-${number}`>;
    state: QueueState;
    stats: {
        avgProcessingTime: number;
        error: number;
        processed: number;
        processing: number;
        rejected: number;
        success: number;
    };
}
Returns {
    activeRunners: number;
    isDraining: boolean;
    isEmpty: boolean;
    isIdle: boolean;
    isPaused: boolean;
    isRunning: boolean;
    isStopped: boolean;
    isWaiting: boolean;
    name: string;
    pending: number;
    rateLimiter: {
        averageWaitTime: number;
        capacity: number;
        createdAt: number;
        currentTokens: number;
        refillIntervalMs: number;
        rejectedRequests: number;
        rejectionRate: number;
        successfulRequests: number;
        totalRequests: number;
        totalWaitTime: number;
        uptime: number;
        waitCount: number;
    };
    runningNodes: Set<`${number}-${number}`>;
    state: QueueState;
    stats: {
        avgProcessingTime: number;
        error: number;
        processed: number;
        processing: number;
        rejected: number;
        success: number;
    };
}
- Defined in packages/observer/dist/types/queue/queue.d.ts:194

state

get state(): QueueState
The current state of the queue

Returns QueueState
- Defined in packages/observer/dist/types/queue/queue.d.ts:178

stats

get stats(): {
    avgProcessingTime: number;
    error: number;
    processed: number;
    processing: number;
    rejected: number;
    success: number;
}
The stats of the queue

Returns {
    avgProcessingTime: number;
    error: number;
    processed: number;
    processing: number;
    rejected: number;
    success: number;
}
- Defined in packages/observer/dist/types/queue/queue.d.ts:186

Methods

add

add(data: EventData<S, E>, priority?: number): boolean
Emit an event to the observer, which then gets picked up by the queue and processed.
Parameters
- data: EventData<S, E>
  The data to add to the queue
- Optionalpriority: number
  The priority of the data
Returns boolean
Note
If priority is provided, it will be used to determine the order of the data in the queue. The higher the priority, the sooner the data will be processed.

Note
If priority is not provided, the data will be added to the queue with a priority of 0.
- Defined in packages/observer/dist/types/queue/queue.d.ts:139

debug

debug(on?: boolean | "info" | "verbose"): void
Parameters
- Optionalon: boolean | "info" | "verbose"
Returns void
- Defined in packages/observer/dist/types/queue/queue.d.ts:140

flush

flush(limit?: number): Promise<number>
Executes limit items in the queue and then emits a flush event. Returns the number of items processed.
Parameters
- Optionallimit: number
Returns Promise<number>
- Defined in packages/observer/dist/types/queue/queue.d.ts:121

off

off<K extends keyof QueueEvents<S, E>>(event: K, listener?: Function): void
Type Parameters
- K extends keyof QueueEvents<S, E>
Parameters
- event: K
- Optionallistener: Function
Returns void
- Defined in packages/observer/dist/types/queue/queue.d.ts:145

on

on<K extends keyof QueueEvents<S, E>>(
event: K,
listener: (payload: InternalQueueEvent<QueueEvents<S, E>[K]>) => void,
): Cleanup
Type Parameters
- K extends keyof QueueEvents<S, E>
Parameters
- event: K
- listener: (payload: InternalQueueEvent<QueueEvents<S, E>[K]>) => void
Returns Cleanup
- Defined in packages/observer/dist/types/queue/queue.d.ts:141
on<K extends keyof QueueEvents<S, E>>(
event: K,
): EventGenerator<QueueEvents<S, E>, K>
Type Parameters
- K extends keyof QueueEvents<S, E>
Parameters
- event: K
Returns EventGenerator<QueueEvents<S, E>, K>
- Defined in packages/observer/dist/types/queue/queue.d.ts:142

once

once<K extends keyof QueueEvents<S, E>>(
event: K,
): EventPromise<InternalQueueEvent<QueueEvents<S, E>[K]>>
Type Parameters
- K extends keyof QueueEvents<S, E>
Parameters
- event: K
Returns EventPromise<InternalQueueEvent<QueueEvents<S, E>[K]>>
- Defined in packages/observer/dist/types/queue/queue.d.ts:143
once<K extends keyof QueueEvents<S, E>>(
event: K,
listener: (payload: InternalQueueEvent<QueueEvents<S, E>[K]>) => void,
): Cleanup
Type Parameters
- K extends keyof QueueEvents<S, E>
Parameters
- event: K
- listener: (payload: InternalQueueEvent<QueueEvents<S, E>[K]>) => void
Returns Cleanup
- Defined in packages/observer/dist/types/queue/queue.d.ts:144

pause

pause(): Promise<void> | undefined
Pause the queue

Returns Promise<void> | undefined
- Defined in packages/observer/dist/types/queue/queue.d.ts:105

purge

purge(): number
Clears the queue and emits a purge event. Returns the number of items purged.

Returns number
- Defined in packages/observer/dist/types/queue/queue.d.ts:126

resume

resume(): Promise<void> | undefined
Resume the queue

Returns Promise<void> | undefined
- Defined in packages/observer/dist/types/queue/queue.d.ts:109

shutdown

shutdown(force?: boolean): Promise<number>
Empty and process all items in the queue and then stop the queue. Returns the number of items processed.
Parameters
- Optionalforce: boolean
Returns Promise<number>
- Defined in packages/observer/dist/types/queue/queue.d.ts:115

start

start(): Promise<void> | undefined
Start the queue if it is not already started

Returns Promise<void> | undefined
- Defined in packages/observer/dist/types/queue/queue.d.ts:97

stop

stop(): void
Stop the queue

Returns void
- Defined in packages/observer/dist/types/queue/queue.d.ts:101

Class EventQueue<S, E>

Type Parameters

Index

Constructors

Accessors

Methods

Constructors

constructor

Type Parameters

Parameters

Returns EventQueue<S, E>

Accessors

isDraining

Returns boolean

isIdle

Returns boolean

isPaused

Returns boolean

isRunning

Returns boolean

isStopped

Returns boolean

isWaiting

Returns boolean

name

Returns string

pending

Returns number

snapshot

state

Returns QueueState

stats

Returns { avgProcessingTime: number; error: number; processed: number; processing: number; rejected: number; success: number;}

Methods

add

Parameters

Returns boolean

Note

Note

debug

Parameters

Returns void

flush

Parameters

Returns Promise<number>

off

Type Parameters

Parameters

Returns void

on

Type Parameters

Parameters

Returns Cleanup

Type Parameters

Parameters

Returns EventGenerator<QueueEvents<S, E>, K>

once

Type Parameters

Parameters

Returns EventPromise<InternalQueueEvent<QueueEvents<S, E>[K]>>

Type Parameters

Parameters

Returns Cleanup

pause

Returns Promise<void> | undefined

purge

Returns number

resume

Returns Promise<void> | undefined

shutdown

Parameters

Returns Promise<number>

start

Returns Promise<void> | undefined

stop

Returns void

Settings

On This Page

Returns {
avgProcessingTime: number;
error: number;
processed: number;
processing: number;
rejected: number;
success: number;
}