Delayed Message Exchange Not Working with Topic Type

[mreouven]

Describe the bug

Delayed messages are not delivered when using x-delayed-type: "topic" with the RabbitMQ delayed message exchange plugin. The message is published but never consumed.

Steps to Reproduce

1. Set up a delayed exchange with x-delayed-type: "topic":

   await ch.assertExchange(EXCHANGE, "x-delayed-message", {
     durable: true,
     arguments: { "x-delayed-type": "topic" },
   });

2. Bind a queue with a topic pattern (e.g., "demo.node.#")
3. Publish a delayed message with routing key "demo.node.hello" and x-delay header
4. The message is published but never received by the consumer

Actual Behavior

• Message is published successfully
• Consumer never receives the message
• Script hangs waiting for message delivery

Workaround

Changing x-delayed-type to "direct" resolves the issue:

arguments: { "x-delayed-type": "direct" },

Environment

• RabbitMQ with delayed message exchange plugin enabled
• Node.js with amqplib
• Exchange type: x-delayed-message
• Binding pattern: "demo.node.#" (topic pattern)
• Routing key: "demo.node.hello"

Additional Notes

This appears to be a bug or limitation in the delayed message exchange plugin when using topic routing patterns. Direct routing works as expected.

---

Reproduction steps

import amqp from "amqplib";

const AMQP_URL = process.env.AMQP_URL ?? "amqp://admin:admin@localhost:5672";
const EXCHANGE = process.env.EXCHANGE ?? "demo.delayed.topic";
const QUEUE = process.env.QUEUE ?? "demo.node.q1";
const BIND = process.env.BIND ?? "demo.node.#";
const ROUTING_KEY = process.env.ROUTING_KEY ?? "demo.node.hello";
const DELAY_MS = Number(process.env.DELAY_MS ?? 5000);

const run = async () => {
console.log("Connecting to RabbitMQ");
const conn = await amqp.connect(AMQP_URL);
console.log("Connected to RabbitMQ");
const ch = await conn.createChannel();
console.log("Created channel");

await ch.assertExchange(EXCHANGE, "x-delayed-message", {
 durable: true,
 arguments: { "x-delayed-type": "topic" },
});

await ch.assertQueue(QUEUE, { durable: true });
await ch.bindQueue(QUEUE, EXCHANGE, BIND);
await ch.prefetch(10);

ch.consume(
 QUEUE,
 (m) => {
   if (!m) return;
   const body = m.content.toString("utf8");
   console.log({
     receivedAt: new Date().toISOString(),
     routingKey: m.fields.routingKey,
     headers: m.properties.headers ?? {},
     body: tryJson(body),
   });
   ch.ack(m);
 },
 { noAck: false }
);

const msg = {
 ts: new Date().toISOString(),
 routingKey: ROUTING_KEY,
 delayMs: DELAY_MS,
 payload: process.env.PAYLOAD ?? "hello from node",
};

console.log("Publishing message");
ch.publish(EXCHANGE, ROUTING_KEY, Buffer.from(JSON.stringify(msg)), {
 contentType: "application/json",
 persistent: true,
 headers: { "x-delay": DELAY_MS },
});

function tryJson(s) {
 try {
   return JSON.parse(s);
 } catch {
   return s;
 }
}
};

run();

Expected behavior

The delayed message should be delivered after the specified delay period.

Additional context

rabbit 4.2.0-management + plugin rabbitmq-delayed-message-exchange

[michaelklishin]

There is no inherent reason why this plugin would have to do anything differently when re-publishing to topic exchanges.

It has many serious limitations in general, all very clearly laid out in the README.

If topic exchanges were unsupported we'd be flooded with complaints but that's not the case. The issue is almost certainly specific to your environment and it's on you to investigate it, not on us to prove that there is no issue. In particular since this project is no longer maintained beyond building it for every new RabbitMQ plugin series.

[michaelklishin]

In fact, the test suite covers all the commonly used built-in exchange types, including the topic exchange, and it has been the case since the early days, for a good decade.

The re-published message is not routed the way you expect it to.