LavinMQ with NodeJS

Node.js developers have a number of options for AMQP client libraries. In this tutorial, amqp-client.js is used, an asynchronous client for publishing and consuming of messages.

Before we jump right into writing our code, let’s set up the development environment.

Node.js development environment

First make sure that you have node.js and that it’s available from your command line. You can confirm this by running: node --version
If you do not have node.js installed, go to nodejs.org
While in the project directory, run npm init to initialize that directory as a Nodejs project. Leave all the questions unanswered/empty by pressing enter all the way.
Next add @cloudamqp/amqp-client and dotenv as a dependency to your package.json file. This can be done by running the following command in the terminal: npm install --save @cloudamqp/amqp-client dotenv
Create a .env file in in the root directory
Add CLOUDAMQP_URL="lavinmq_url" to the `.env’ file. Replace lavinmq_url with your correct server url
In order for amqp-client to work you need to add “type”: “module” to your package.json just right under "name": "node_tutorials" like so:

{
  "name": "node_tutorials",
  "type": "module",
}

Publishing messages

//Dependencies
import { AMQPClient } from '@cloudamqp/amqp-client'
import {} from 'dotenv/config'

const cloudAMQPURL = process.env.CLOUDAMQP_URL

async function startProducer() {
  try {
    //Setup a connection to the RabbitMQ server
    const connection = new AMQPClient(cloudAMQPURL)
    await connection.connect()
    const channel = await connection.channel()

    console.log("[✅] Connection over channel established")

    await channel.queue('hello_world', {durable: false})

    //Publish a message to the exchange
    async function sendToQueue(routingKey, body) {
      await channel.basicPublish("", routingKey, body)
      console.log("[📥] Message sent to queue", body)
    }
  
    //Send some messages to the queue
    sendToQueue("hello_world", "Hello World");
    sendToQueue("hello_world", "Hello World");
    sendToQueue("wrong_routing_key", "Hello World");
  
    setTimeout(() => {
      //Close the connection
      connection.close()
      console.log("[❎] Connection closed")
      process.exit(0)
    }, 500);
  } catch (error) {
    console.error(error)
    //Retry after 3 second
    setTimeout(() => {
      startProducer()
    }, 3000)
  }
}

startProducer()

Consuming messages

import { AMQPClient } from '@cloudamqp/amqp-client'
import {} from 'dotenv/config'

const lavinmqUrl = process.env.CLOUDAMQP_URL

async function startConsumer() {
  //Setup a connection to the RabbitMQ server
  const connection = new AMQPClient(lavinmqUrl)
  await connection.connect()
  const channel = await connection.channel()

  console.log("[✅] Connection over channel established")
  console.log("[❎] Waiting for messages. To exit press CTRL+C ")

  const q = await channel.queue('hello_world', {durable: false})

  let counter = 0;

  await q.subscribe({noAck: true}, async (msg) => {
    try {
      console.log(`[📤] Message received (${++counter})`, msg.bodyToString())
    } catch (error) {
      console.error(error)
    }
  })

  //When the process is terminated, close the connection
  process.on('SIGINT', () => {
    channel.close()
    connection.close()
    console.log("[❎] Connection closed")
    process.exit(0)
  });
}

startConsumer().catch(console.error);

Introduction

Configuration

Language Support

AMQP 0-9-1 Overview

More Exchange Types

More Consumer Features

Queue Deep-dive

Other Features

Reliable Message Delivery

High Availability and Backup

Monitoring

Management HTTP API

Tutorials

Networking

LavinMQ CLI

LavinMQ with NodeJS

Node.js development environment

Publishing messages

Consuming messages

Managed LavinMQ instance on CloudAMQP

Help and feedback