consume does not respect promises?

[mabels]

consume does not respect promises?

[cressie176]

Hi @mabels,

My understanding of the problem you describe is that when there are multiple messages on the queue, they are delivered to your consumer as fast as possible. When onMessage is an asynchronous function, amqplib does not wait for it to yield/resolve, consequently your application is being flooded with multiple messages in parallel.

The above behaviour is by design and necessary due to the amqp protocol. RabbitMQ would still deliver messages as fast as possible to amqplib even if the library did wait for each onMessage function to yield/resolve before calling the next one. This would create a real possibility of running out of memory since all of the messages would be buffered in Node's event loop. Instead the way to throttle messages is to use RabbitMQ's consumer prefetch feature in combination with explicitly acknowledged messages (the default), e.g.

await channel.prefetch(1);
await channel.consume(q.name, async (msg) => {
  console.log("enter onMessage");
  try {
    await fetch("www.google.com");
    channel.ack(msg);
    console.log("success");
  } catch (err) {
    channel.nack(msg);
    console.error(err);
  }
});

[cressie176]

Hi @mabels

I think what you want can be achieved by specifying a consumer prefetch as per the channel api

[mabels]

Hi @cressie176,

i might be my inability to explain the problem but that is not the problem. I try it with an example:

 channel.consume(q.queue, async (msg) => {
  console.log("enter onMessage")
  await fetch("www.google.com")
  console.log("the expectation is that the invokation will not called again before i reach this line")
  console.log("but if consume message does not wait for the promise to resolve than")
  console.log("enter onMessage will called before the fetch is resolved") 
  console.log("which could cause a storm of events. In js there is no way to use consume with async functions")
 })

[mabels]

Hi,

now it get it but i'm still think that this is a stretch.

Thx for the explanation.

meno

[teddybee]

Although @cressie176 's solution solved my problem (I couldn't process messages one-by-one) in a synchronous handler, because of js can't await/force sync run an async function in a sync handler, I think in a recent years devs mostly uses async-await in callbacks.
I have tried to rewrite the whole worker app with sync functions, besides it was a nightmare, I couldn't send feedbacks of processing status.
With async functions, it worked fine (I send a message in a status queue every 10 percent), with fully synced handlers I saw only the initial and the end status on the frontend.
So they wasn`t sent or sent at the end all at once.