Support priority queues

[Alxandir]

We are using this plugin in an instance where most queues also have priority queues enabled.
It would be great if this plugin could be adapted to optionally support the "increase" and "decrease" of priority of a message if a duplicate message of a different priority is submitted to the queue. Ideally being able to enable increase and decrease separately to support the most use-cases.

Example 1 - Increasing priority when "duplicate priority increase" enabled:

1. A message with deduplication header "123" is present on Queue A, priority 7.
2. A second message with deduplication header "123" is published to Queue A, with priority set to 9.
3. The plugin determines that there is a cache hit, but the priority of the existing message is lower.
4. Drop the existing priority 7 message.
5. Publish the new message to priority 9.

Example 2 - Decreasing priority when "duplicate priority decrease" enabled:

1. A message with deduplication header "123" is present on Queue A, priority 7.
2. A second message with deduplication header "123" is published to Queue A, with priority set to 2.
3. The plugin determines that there is a cache hit, but the priority of the existing message is higher.
4. Drop the existing priority 7 message.
5. Publish the new message to priority 2.

Example 3 - Ignoring priority increase when "duplicate priority increase" is disabled:

1. A message with deduplication header "123" is present on Queue A, priority 7.
2. A second message with deduplication header "123" is published to Queue A, with priority set to 9.
3. The plugin determines that there is a cache hit, but the priority of the existing message is lower.
4. The incoming message is dropped.

Example 4 - Ignoring priority decrease when "duplicate priority decrease" is disabled:

1. A message with deduplication header "123" is present on Queue A, priority 7.
2. A second message with deduplication header "123" is published to Queue A, with priority set to 2.
3. The plugin determines that there is a cache hit, but the priority of the existing message is higher.
4. The incoming message is dropped.

[noxdafox]

Hello,

apologies for the late reply.

The behaviour you are requesting cannot be easily implemented. RabbitMQ queues do not support the removal of messages which are not at the head or tail of the queues. This is the reason they are called queues and not, for example, lists or streams.

The way the queue deduplication logic works is by hooking into the queue behaviour before the actual implementation is executed. In other words, when a message is about to be published into the queue, we check if the deduplication header is already in the deduplication cache. If so we discard it, otherwise we add its deduplication header to the cache. When a message is consumed and acknowledged, its deduplication header is removed from the cache.

This makes possible to implement the deduplication logic on top of the queue and it's quite optimal as we do not need to scan the entire queue to assess if the message is already present or not.

The drawback for this implementation is we are bound to the rabbit_backing_queue behaviour which is only used for classic and priority queues. This is why we cannot support streams or quorum queue (so far).

It would be very inefficient to do what you are suggesting as the way the messages are stored internally (in memory or disk) is opaque. This lets the broker implement the most performant mechanisms for storage but the consequence is we cannot fetch a given message at will.