@otklib/queue

Install

$ npm install -S @otklib/queue

Usage

Example:

import { QueueConnector, RabbitQueueConnector, QueuePipe } from '@otklib/queue'

const queueConnection: QueueConnector = new RabbitQueueConnector('amqp://<user>:<pass>@<host>:<port>')

// Pipe
const pipe: QueuePipe = new QueuePipe(queueConnection)
pipe
  .from('test-queue-in')
  .filter(
    async (queue: string, data: any): Promise<any> => {
      console.log('filter', queue, data)
      await new Promise((resolve) => setTimeout(resolve, 1000))
      return data
    },
  )
  .to('test-queue-out')
  .run(5)

// Consumer
queueConnection.subscribe('test-queue-out', async (data) => console.log('consume', data), 10)

// Publisher
for (let i = 0; i < 10; i++) {
  queueConnection.send('test-queue-in', { i })
  console.log('publish', { i })
}