[Question] Async ConsumerFunc

[AmauryD]

I noticed that the ConsumerFunc does not support async, so the client doesn't wait for the consumer function to finish before processing more messages. The problem is that my CPU becomes overwhelmed when receiving 1,000,000 messages in a short amount of time.

Is there a way to solve this problem ? The only temporary fix I have is to enqueue all the asynchronous calls in a queue, but that seems a bit overkill and not very efficient.

const consumer = await this.client.declareConsumer({
            ...
 },  (message) => {
        this.worker.enqueue(async () => {
           ...
        })
 })

Thank you,

[icappello]

Hello, sorry for the delay.
We'll get back to you in the following days

[tunniclm]

I am also interested in this because I am trying to replace a single consumer on a queue that needs to respond to a single message at a time in order with potentially async operations (like updating a database). I'm using a classic queue with prefetch of 1 and manual acknowledgements at the moment, but want to use a partitioned superstream instead (I already tried a consistent hash exchange but its lack of partition stability on restarts was a deal breaker).

While investigating superstreams, I was concerned to make sure I could keep the ordered delivery guarantee and I couldn't find much on that in the stream library documentation so digging in the code I spotted this

rabbitmq-stream-js-client/src/client.ts

Lines 554 to 556 in d8083e5

	messages.map((message) => {
	if (messageFilter(message)) consumer.handle(message)
	})

which looked like it would not wait to deliver the next message in a chunk if the handler function wanted to do something async.

Is that correct and is it something that is in scope to add support for?

[AmauryD]

@tunniclm i ended up patching the library code locally. Maybe not the best idea but that works.

Here is the patch (https://pnpm.io/cli/patch) :

rabbitmq-stream-js-client@0.4.2.patch

[tunniclm]

@AmauryD Unfortunately, I'm still seeing overlap in message processing I think because the delivery of new message chunks is overlapping. I suspect this is due to the default credit policy of creditsOnChunkReceived(2, 1) (starts requesting 2 chunks and then requests a new one each time one is received) which can result in new chunks being delivered/start processing before the prior chunk is finished processing.

I'm going to try testing with a policy of creditsOnChunkCompleted(1, 1). It looks like I can't configure this when declaring a super-stream consumer, so I'm going to hard-code it in my client library for the test.

[AmauryD]

@AmauryD Unfortunately, I'm still seeing overlap in message processing I think because the delivery of new message chunks is overlapping. I suspect this is due to the default credit policy of creditsOnChunkReceived(2, 1) (starts requesting 2 chunks and then requests a new one each time one is received) which can result in new chunks being delivered/start processing before the prior chunk is finished processing.

I'm going to try testing with a policy of creditsOnChunkCompleted(1, 1). It looks like I can't configure this when declaring a super-stream consumer, so I'm going to hard-code it in my client library for the test.

I forgot to mention it earlier, but yes, I also added:

{
            creditPolicy: creditsOnChunkCompleted(1, 1),
}

to my Consumer.
However, as you pointed out, this isn't possible on SuperStream :(

[dc1992]

@AmauryD @tunniclm By default every stream consumer will have creditsOnChunkCompleted(1, 1) so that messages will be processed in order, this will be included in the next release