RabbitMq node.js token bucket example

new_task.mjs

import amqp from "amqplib";

const delay = (ms) => new Promise((resolve) => setTimeout(resolve, ms));

const connection = await amqp.connect("amqp://localhost");

const channel = await connection.createChannel();

var queue = "task_queue";
var msg = "Hello World!";

await channel.assertQueue(queue, {
  durable: true,
});

while (true) {
  channel.sendToQueue(queue, Buffer.from(msg), {
    persistent: true,
  });
  console.log(" [x] Sent '%s'", msg);

  await delay(100);
}

worker-limit.mjs

import amqp from "amqplib";

const connection = await amqp.connect("amqp://localhost");

const channel = await connection.createChannel();

var limit = 3;
var bucketQueue = `tokens${limit}`;

await channel.assertQueue(bucketQueue, {
  durable: true,
  messageTtl: 1000,
  maxLength: limit,
});

function writeToken() {
  channel.sendToQueue(bucketQueue, Buffer.from(new Date().toISOString()), {
    persistent: true,
  });
  setTimeout(writeToken, 1000 / limit);
}

console.log("Starting token bucket publisher");
writeToken();

worker.mjs

import amqp from "amqplib";

const connection = await amqp.connect("amqp://localhost");
const channel = await connection.createChannel();

var queue = "task_queue";
var bucketQueue = "tokens3";

await channel.assertQueue(queue, { durable: true });
await channel.prefetch(1);

console.log(" [*] Waiting for messages in %s. To exit press CTRL+C", queue);

let c = 0;

const consumer = await channel.consume(
  queue,
  async (msg) => {
    const token = await channel.get(bucketQueue);

    if (token) {
      console.log(" [x] Received %s", msg.content.toString(), new Date());
      channel.ack(msg);
    } else {
      //console.log(" [x] Bucket is full, rejecting message");
      channel.nack(msg);
    }
  },
  { noAck: false }
);