amqplib-as-promised

This module wrapping amqplib node.js library to provide classic Promise interface (instead of original implementation using Bluebird promises).

Publishing messages in a channel or confirm channel will resolve the promise only if operation will be finished.

Also, AMQP protocol closing channel when any error occurs. (even in case of using checkQueue method, error is emitted and channel is closed, when queue does not exist). That library wrapper automatically reconnects channel and returning error as a Promise reject response.

Requirements

This module requires Node >= 12.

Installation

npm install amqplib-as-promised

Usage

(with async await syntax)

const { Connection } = require('amqplib-as-promised')
const connection = new Connection(AMQP_URL)
await connection.init()
const channel = await connection.createChannel() // or createConfirmChannel
await channel.assertQueue(queueName)
await channel.sendToQueue(queueName, Buffer.from(testMessage))
await channel.close()
await connection.close()

TypeScript version

import { Connection } from 'amqplib-as-promised'
const connection = new Connection(AMQP_URL)
await connection.init()
const channel = await connection.createChannel() // or createConfirmChannel
await channel.assertQueue(queueName)
await channel.sendToQueue(queueName, Buffer.from(testMessage))
await channel.close()
await connection.close()

For proper typing with TypeScript, @types/amqplib in devDependencies is needed. (compatible versions: 0.8.0 < 1)

API

Connection

| Method | arguments | return type | notes | | -------------------- | ----------------------------------------------------- | ------------------------ | ----- | | constructor | url: string options?: amqplib.Options.Connect | Connection | | | init | - | Promise<void> | | | createChannel | - | Promise<Channel> | | | createConfirmChannel | - | Promise<ConfirmChannel> | | | close | - | Promise<void> | | | waitForClose | - | Promise<void> | |

Channel

| Method | arguments | return type | notes | | -------------- | ------------------------------------------------------------------------------------------------------------------ | ---------------------------------------- | ----- | | checkQueue | queueName: string | Promise<amqplib.Replies.AssertQueue> | | | assertQueue | queueName: string options?: amqplib.Options.AssertQueue | Promise<amqplib.Replies.AssertQueue> | | | deleteQueue | queueName: string options?: amqplib.Options.DeleteQueue | Promise<amqplib.Replies.DeleteQueue> | | | sendToQueue | queueName: string content: Buffer options?: amqplib.Options.Publish | Promise<unknown> | | | bindQueue | queueName: string, source: string, pattern: string, args?: any | Promise<amqplib.Replies.Empty> | | | unbindQueue | queueName: string, source: string, pattern: string, args?: any | Promise<amqplib.Replies.Empty> | | | assertExchange | exchangeName: string exchangeType: string options?: amqplib.Options.AssertExchange | Promise<amqplib.Replies.AssertExchange> | | | checkExchange | exchangeName: string | Promise<amqplib.Replies.Empty> | | | deleteExchange | exchangeName: string options: amqplib.Options.DeleteExchange | Promise<amqplib.Replies.Empty> | | | bindExchange | destination: string source: string pattern: string args?: any | Promise<amqplib.Replies.Empty> | | | unbindExchange | destination: string source: string pattern: string args?: any | Promise<amqplib.Replies.Empty> | | | publish | exchange: string queue: string content: Buffer options?: amqplib.Options.Publish | Promise<unknown> | | | prefetch | count: number global: boolean | Promise<void> | | | consume | queueName: string handler: (message: amqplib.Message | null) => any options?: amqplib.Options.Consume | Promise<amqplib.Replies.Consume> | | | cancel | consumerTag: string | Promise<amqplib.Replies.Empty> | | | get | queueName: string options?: amqplib.Options.Get | Promise<Message | false> | | | ack | message: amqplib.Message allUpTo?: boolean | void | | | nack | message: amqplib.Message allUpTo?: boolean requeue?: boolean | void | | | close | - | Promise<void> | |

Confirm channel

| Method | arguments | return type | notes | | --------------- | ------------------------------------------------------------------------------------------------ | ------------------------------- | ----- | | sendToQueue | queueName: string content: Buffer options?: amqplib.Options.Publish | Promise<amqplib.Replies.Empty> | | | publish | exchange: string queue: string content: Buffer options?: amqplib.Options.Publish | Promise<amqplib.Replies.Empty> | | | waitForConfirms | exchange: string queue: string content: Buffer options?: amqplib.Options.Publish | Promise<void> | |

License

Copyright (c) 2019-2022 Piotr Roszatycki [email protected]

Copyright (c) 2017-2018 Tadeusz Wawszczak

MIT