@loutrejs/bullmq

@loutrejs/loutre/tasksのQueue Consumer executionをBullMQ Workerへ接続するDriver packageです。

Install

npm install @loutrejs/loutre @loutrejs/bullmq bullmq

bullmqはpeer dependencyです。

Usage

import { defineApplication, defineModule } from '@loutrejs/loutre'
import { consume, queue, task } from '@loutrejs/loutre/tasks'
import { bindBullMqQueue } from '@loutrejs/bullmq'
import { z } from 'zod'

const orders = queue({
  name: 'orders',
  payload: z.object({ id: z.string() }),
})

const processOrder = task<{ id: string }, void>({
  name: 'orders.process',
  factory:
    () =>
    async ({ id }) => {
      console.log(id)
    },
})

const orderConsumer = consume({
  name: 'orders.consumer',
  queue: orders,
  task: processOrder,
})

const QueueModule = defineModule({
  providers: [
    bindBullMqQueue(orders, {
      connection: { host: '127.0.0.1', port: 6379 },
    }),
  ],
  executions: [orderConsumer],
})

export default defineApplication({ modules: [QueueModule] })

Trigger execution is started through the Tasks Extension Host API:

const app = await bootstrapApplication({ application })
await app.tasks.start()

Options

bindBullMqQueue()ではBullMQ Workerへ渡すconnectionに加えて、concurrency、prefix、workerOptionsを指定できます。

低レベルAPIとしてcreateBullMqQueueDriver()も公開しています。

Documentation

• Getting Started
• Loutre

License

MIT