tttp / node-rabbitmq-client Goto Github PK

import {Connection} from 'rabbitmq-client'

// Initialize:
const rabbit = new Connection('amqp://guest:guest@localhost:5672')
rabbit.on('error', (err) => {
  console.log('RabbitMQ connection error', err)
})
rabbit.on('connection', () => {
  console.log('Connection successfully (re)established')
})

// Consume messages from a queue:
// See API docs for all options
const sub = rabbit.createConsumer({
  queue: 'user-events',
  queueOptions: {durable: true},
  // handle 2 messages at a time
  qos: {prefetchCount: 2},
  // Optionally ensure an exchange exists
  exchanges: [{exchange: 'my-events', type: 'topic'}],
  // With a "topic" exchange, messages matching this pattern are routed to the queue
  queueBindings: [{exchange: 'my-events', routingKey: 'users.*'}],
}, async (msg) => {
  console.log('received message (user-events)', msg)
  // The message is automatically acknowledged when this function ends.
  // If this function throws an error, then msg is NACK'd (rejected) and
  // possibly requeued or sent to a dead-letter exchange
})

sub.on('error', (err) => {
  // Maybe the consumer was cancelled, or the connection was reset before a
  // message could be acknowledged.
  console.log('consumer error (user-events)', err)
})

// Declare a publisher
// See API docs for all options
const pub = rabbit.createPublisher({
  // Enable publish confirmations, similar to consumer acknowledgements
  confirm: true,
  // Enable retries
  maxAttempts: 2,
  // Optionally ensure the existence of an exchange before we use it
  exchanges: [{exchange: 'my-events', type: 'topic'}]
})

// Publish a message to a custom exchange
await pub.send(
  {exchange: 'my-events', routingKey: 'users.visit'}, // metadata
  {id: 1, name: 'Alan Turing'}) // message content

// Or publish directly to a queue
await pub.send('user-events', {id: 1, name: 'Alan Turing'})

// Clean up when you receive a shutdown signal
async function onShutdown() {
  // Waits for pending confirmations and closes the underlying Channel
  await pub.close()
  // Stop consuming. Wait for any pending message handlers to settle.
  await sub.close()
  await rabbit.close()
}

// Will wait for the connection to establish and then create a Channel
const ch = await rabbit.acquire()

// Channels can emit some events too (see documentation)
ch.on('close', () => {
  console.log('channel was closed')
})

// Create a queue for the duration of this connection
await ch.queueDeclare({queue: 'my-queue'})

// Enable publisher acknowledgements
await ch.confirmSelect()

const data = {title: 'just some object'}

// Resolves when the data has been flushed through the socket or if
// ch.confirmSelect() was called: will wait for an acknowledgement
await ch.basicPublish({routingKey: 'my-queue'}, data)

const msg = ch.basicGet('my-queue')
console.log(msg)

await ch.queueDelete('my-queue')

// It's your responsibility to close any acquired channels
await ch.close()

// rpc-server.js
const rabbit = new Connection()

const rpcServer = rabbit.createConsumer({
  queue: 'my-rpc-queue'
}, async (req, reply) => {
  console.log('request:', req.body)
  await reply('pong')
})

process.on('SIGINT', async () => {
  await rpcServer.close()
  await rabbit.close()
})

// rpc-client.js
const rabbit = new Connection()

const rpcClient = rabbit.createRPCClient({confirm: true})

const res = await rpcClient.send('my-rpc-queue', 'ping')
console.log('response:', res.body) // pong

await rpcClient.close()
await rabbit.close()