You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

65 lines
1.9 KiB

import {Directive, OptionDefinition} from '../../Directive'
import {Inject, Injectable} from '../../../di'
import {Bus, PushedToQueue, Queue} from '../../../support/bus'
import {Queueables} from '../../../service/Queueables'
export class ListenDirective extends Directive {
protected readonly queue!: Queue
protected readonly queueables!: Queueables
protected readonly bus!: Bus
getDescription(): string {
return 'listen for jobs pushed to the queue and attempt to execute them'
getKeywords(): string | string[] {
return 'queue-listen'
getOptions(): OptionDefinition[] {
return []
async handle(): Promise<void> {'Subscribing to queue events...')
await this.bus.subscribe(PushedToQueue, async () => {
// A new job has been pushed to the queue, so try to pop it and execute it.
// We may get undefined if some other worker is running and picked up this job first.
await this.tryExecuteJob()
})'Setting periodic poll...')
const handle = setInterval(async () => {
await this.tryExecuteJob()
}, 5000)'Listening for jobs...')
await this.untilInterrupt()'Shutting down...')
protected async tryExecuteJob(): Promise<void> {
try {
const job = await this.queue.pop()
if ( !job ) {
return // Some other worker already picked up this job
}`Executing: ${job.constructor?.name || 'unknown job'}`)
await job.execute()
this.success('Execution finished.')
} catch (e: unknown) {
this.error('Failed to execute job.')