How detect that a certain block of work (a group of messages) were consumed, processed and acknowledged?

[hamidovz]

hi ! i am trying to build video processing app. what i want to do is , first process videos and convert them to 480p variant and if there is no more messages in current queue , then start to process 720p variants.

here is my current code

480p consumer

const amqp = require('amqplib');

async function process480p() {
  const conn = await amqp.connect('amqp://localhost');
  const channel = await conn.createChannel();
  
  const queue480p = '480p_queue';
  const queue720p = '720p_queue';
  
  await channel.assertQueue(queue480p, { durable: true });
  await channel.assertQueue(queue720p, { durable: true });
  await channel.prefetch(1,false)
  
  channel.consume(queue480p, async (msg) => {
    if (msg !== null) {

      const videoDetails = JSON.parse(msg.content.toString());
      console.log('Processing 480p for video:', videoDetails.videoId);

      // Simulate video processing
      await new Promise(resolve => setTimeout(resolve, 2000)); // 1 second processing time

      console.log('Processed 480p for video:', videoDetails.videoId);
      channel.ack(msg);

      // Check if the 480p queue is empty
      const { messageCount } = await channel.checkQueue(queue480p);
      if (messageCount === 0) {
        console.log('480p queue empty, enqueueing for 720p processing:', videoDetails.videoId);
        channel.sendToQueue(queue720p, Buffer.from(JSON.stringify(videoDetails)), { persistent: true });
      }
    }
  });
}

process480p();

720p consumer

const amqp = require('amqplib');

async function process720p() {
  const conn = await amqp.connect('amqp://localhost');
  const channel = await conn.createChannel();
  
  const queue720p = '720p_queue';
  await channel.assertQueue(queue720p, { durable: true });

  channel.consume(queue720p, (msg) => {
    if (msg !== null) {
      const videoDetails = JSON.parse(msg.content.toString());
      console.log('Processing 720p for video:', videoDetails.videoId);

      // Simulate video processing
      setTimeout(() => {
        console.log('Processed 720p for video:', videoDetails.videoId);
        channel.ack(msg);
      }, 1000); // 1 second processing time
    }
  });
}

process720p();

this is my publisher

const amqp = require('amqplib');

async function sendToQueue(videos) {
  const conn = await amqp.connect('amqp://localhost');
  const channel = await conn.createChannel();
  
  const queue480p = '480p_queue';
  await channel.assertQueue(queue480p, { durable: true });

  videos.forEach(videoDetails => {
    const message = JSON.stringify(videoDetails);
    channel.sendToQueue(queue480p, Buffer.from(message), { persistent: true });
    console.log('Sent video to 480p processing queue:', videoDetails);
  });

  await channel.close();
  await conn.close();
  
}

// Array of video details
const videoDetailsList = [
  { videoId: 101, path: '/path/to/video1.mp4' },
  { videoId: 102, path: '/path/to/video2.mp4' },
  { videoId: 103, path: '/path/to/video3.mp4' }
];

sendToQueue(videoDetailsList);

this is my results for 480p consumer from terminal:

Processed 480p for video: 101
Processing 480p for video: 102
Processed 480p for video: 102
Processing 480p for video: 103
480p queue empty, enqueueing for 720p processing: 102
Processed 480p for video: 103
480p queue empty, enqueueing for 720p processing: 103

it looks like there is a problem about asynchronicity

[lukebakken]

it looks like there is a problem about asynchronicity

No, it appears that everything is working as intended.

How do you define "empty" when a queue might have more messages published to it?

When video 102 is done processing, the message for 103 must not have been in the queue yet. I don't know how await channel.checkQueue(queue480p) actually gets the current message count in the queue.

Your publisher should send a DONE message when it is guaranteed that no more messages will be published.

[hamidovz]

Why not do exactly what you state in 1, 2, 3?

• When 100ms passes without a message, cancel consuming from the 480p queue, and start consuming from the 720p queue
• After every 720p encoding, check the 480p message count
• If the 480p queue has messages, stop consuming from the 720p queue, and re-consume from the 480p queue

i tried to do it but ended up race condition on message consuming , even i set up prefetch 1. when i watched terminal , there result was something similar to this:

Processing 480p for video: 101
Processed 480p for video: 101
Processing 480p for video: 102
480p queue empty, enqueueing for 720p processing
Processed 480p for video: 102
Processing 480p for video: 103
Processed 480p for video: 103
480p queue empty, enqueueing for 720p processing

it seemed like second and first message attempt to check queue emptiness at the same time and i think it is because of asynchonous nature of javascript. i thought when i set prefetch 1 , no matter what happens messages will be consumed synchoronously.

[lukebakken]

i thought when i set prefetch 1 , no matter what happens messages will be consumed synchoronously.

A prefetch of 1 means that RabbitMQ will only send your consumer 1 message, then wait for an ack, then send the next message, etc.

This is probably what is happening in your case:

• Your application receives a message, and processes it.
• The application sends the ack back to RabbitMQ, and then asks RabbitMQ for the count of "ready" messages in the queue. The count is 0 because the only other message is in the "Unacked" state (i.e., being processed currently)
• Your application then goes on to the 720p queue.

[hamidovz]

Yes, this is always a possibility.

is there any repo or code for reference for situation similar to mine ? i can use external store but i just want to stick with rabbitmq itself. i am completely newbie to message queues

[lukebakken]

At this time, the best way for me to help you is for you to provide code I can run. Please provide a git repository I can clone with detailed instructions for setting up your code and running it. Don't assume I know anything about node/javascript other than how to install node.js. Assume I have a running RabbitMQ server, but make everything else as easy as possible to get going. I don't have any free time to fix syntax errors, that sort of thing.

I can fork your repository and poke around a bit.

[hamidovz]

At this time, the best way for me to help you is for you to provide code I can run. Please provide a git repository I can clone with detailed instructions for setting up your code and running it. Don't assume I know anything about node/javascript other than how to install node.js. Assume I have a running RabbitMQ server, but make everything else as easy as possible to get going. I don't have any free time to fix syntax errors, that sort of thing.

I can fork your repository and poke around a bit.

thank you for your help and time ! here is the repo: rabbitmq sample app i wrote instructions inside of the readme file. thank you again !

[michaelklishin]

This is a consumer coordination problem, how your consumers depend on each other or interact with each other is not something RabbitMQ is aware of or should be aware of.

What your applications should do when they are done with a certain stage is to publish a message of a special type to a separate queue. In general, in a pipeline of N different stages using N queues is guaranteed to be easier to reason about than one MegaQueue. Single Giant Shared Queue™ is likely the most problematic and the most infamous anti-pattern in messaging and the RabbitMQ community specifically.

Not that when a design problem does not fit the queue consumption-acknowledgement model, there's nothing wrong with using a stream, for which there are now tutorials. Just an idea.

[michaelklishin]

@hamidovz using data services such as Consul, etcd or Redis (or anything else that you like) for consumer coordination is not an anti-pattern, it's exactly what you should do.

Limiting access to queues is not an anti-pattern either, sometimes it is a necessity of your workflow. That's why exclusive queues, Single Active Consumer and (in AMQP 0-9-1) exclusive consumers exist. I would not use exclusive consumers today when SAC exists and has been around for years, and exclusive queues are not meant to be (and cannot) be long lived but your thinking is largely in line with what I'm suggesting above.

A series of queues and consumers that publish (or republish) messages to kick off the next stage in the workflow, with SAC, a single channel with prefetch = 1, and an external data store helping you control access to shared resources, track progress, and so on.

This way there will be no shared queues, no polling, and a lot of freedom to configure individual consumers however you see fit (e.g. a 1080p consumer may use higher timeouts than the 480p one).

[lukebakken]

Just my $0.02, @hamidovz's use case is probably one of the only ones where I would say, "sure, use basic.get". It accomplishes what he needs and there is (apparently) not a huge message workload, so, why not?

[hamidovz]

@hamidovz using data services such as Consul, etcd or Redis (or anything else that you like) for consumer coordination is not an anti-pattern, it's exactly what you should do.

Limiting access to queues is not an anti-pattern either, sometimes it is a necessity of your workflow. That's why exclusive queues, Single Active Consumer and (in AMQP 0-9-1) exclusive consumers exist. I would not use exclusive consumers today when SAC exists and has been around for years, and exclusive queues are not meant to be (and cannot) be long lived but your thinking is largely in line with what I'm suggesting above.

A series of queues and consumers that publish (or republish) messages to kick off the next stage in the workflow, with SAC, a single channel with prefetch = 1, and an external data store helping you control access to shared resources, track progress, and so on.

This way there will be no shared queues, no polling, and a lot of freedom to configure individual consumers however you see fit (e.g. a 1080p consumer may use higher timeouts than the 480p one).

i used locking mechanism and used redis for this. thank you for your time

[hamidovz]

Just my $0.02, @hamidovz's use case is probably one of the only ones where I would say, "sure, use basic.get". It accomplishes what he needs and there is (apparently) not a huge message workload, so, why not?

i used locking mechanism and used redis for this. thank you for your time. but i didn't delete basic.get variant just in case.

[michaelklishin]

Also note that there is a old book that just as relevant today on messaging patterns.

[hamidovz]

thank you very much i will check it out

[hamidovz]

@michaelklishin which book among them do you recommend to read first ?

[michaelklishin]

There aren't that many books on messaging that are not the description of basics. There's another one by a RabbitMQ user from 2024 but Enterprise Integration Patterns has gotten more relevant over time, not less.