# RabbitMQ Delayed Message Plugin ## This Project is No Longer Maintained **Important**: Team RabbitMQ no longer maintains this project. ## Why? This plugin has serious limitations and is based on Mnesia, the original schema data store in RabbitMQ that [was completely removed from RabbitMQ](https://github.com/rabbitmq/rabbitmq-server/pull/15542) in the `4.3.0` development cycle. A [distributed design](https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/issues/229) of this plugin required significant changes (e.g. switching from a custom exchange type to a custom queue type), and took a few person-years worth of R&D investment. As this is an entirely optional feature, it was decided to ship it in the commercial edition. ## Alternatives Available ### Delayed Queues in VMware Tanzu RabbitMQ [VMware Tanzu RabbitMQ](https://www.vmware.com/products/app-platform/tanzu-rabbitmq) supports a separate queue type, [delayed queues](https://techdocs.broadcom.com/us/en/vmware-tanzu/data-solutions/tanzu-rabbitmq-ova/4-2/tanzu-rabbitmq-ova-virtual-machine/site-delayed-queues.html), that, unlike this plugin, offers: * Raft-based replication (same foundation and characteristics as [quorum queues](https://www.rabbitmq.com/docs/quorum-queues)) * Ability to handle backlogs into tens or even hundreds of millions of delayed messages ### Using Dead Lettering for Message Delays The combination of [Dead Lettering](https://www.rabbitmq.com/docs/dlx) and [TTL](https://www.rabbitmq.com/docs/ttl) has been widely used for basic delays and task retries across the RabbitMQ community. Some messaging frameworks, e.g. NServiceBus, set up the topology for such delayed retries for you. This option gives you much better visibility of what's going on with the delayed messages, an ability to purge the messages, and use replicated queues types for their storage. ### Build Your Own Single Node Alternative If the serious limitations of this plugin are acceptable to you, the single-node nature and significant limitations of this plugin make it easier to reimplement without any use of Mnesia: * Target RabbitMQ 4.x where Khepri is a stable feature, enabled by default as of `4.2.0` * Use Khepri for metadata storage * Use RocksDB for storing messages ## Consider the Limitations This plugin adds delayed-messaging (or scheduled-messaging) to RabbitMQ. Its current design **has multiple significant limitation** (documented below), consider using an external scheduler and a data store that fits your needs first. This plugin badly needs a [new design](https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/issues/229) and a reimplementation from the ground up. If you accept the limitations, please read on. ## The Basics With this plugin [enabled](https://www.rabbitmq.com/docs/plugins), a user can declare an exchange with the type `x-delayed-message` and then publish messages with the custom header `x-delay` expressing in milliseconds a delay time for the message. The message will be delivered to the respective queues after `x-delay` milliseconds. ## Intended Use Cases This plugin was designed for delaying message publishing for a number of seconds, minutes, or hours. A day or two at most. It is **not a longer term scheduling solution**. If you need to delay publishing by days, weeks, months, or years, consider using a data store suitable for long-term storage, and an external scheduling tool of some kind. ## Supported RabbitMQ Versions Every [release](https://github.com/rabbitmq/rabbitmq-delayed-message-exchange) of this plugin targest one RabbitMQ release series. This plugin can be enabled on a RabbitMQ cluster that uses either Mnesia or Khepri as [metadata store](https://www.rabbitmq.com/docs/metadata-store), however, when this plugin is enabled **before** Khepri, it must be restarted (or the node must be) after Khepri is enabled. In other words, there are three possible scenarios w.r.t. the schema data store used: 1. If the cluster uses Mnesia for schema store, it works exactly as it did against RabbitMQ 3.13.x 2. If the cluster uses Khepri and the plugin is enabled after Khepri, it will start Mnesia, set up a node-local Mnesia replica and schema, and works as in scenario 1 3. **Important**: if the cluster uses Mnesia, then the plugin is enabled, and then Khepri is enabled, the plugin must be disabled and re-enabled, or the node must be restarted. Then it will start Mnesia and works as in scenario 2 ## Supported Erlang/OTP Versions The latest version of this plugin [requires Erlang 26.2 or later versions](https://www.rabbitmq.com/docs/which-erlang). ## Installation ### Download a Binary Build Binary builds are distributed [via GitHub releases](https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases). As with all 3rd party plugins, the `.ez` file must be copied into a [node's plugins directory](https://rabbitmq.com/plugins.html#plugin-directories) with sufficient permissions for the effective user of the RabbitMQ process to load it from disk. To find out what the plugins directory is, use `rabbitmq-plugins directories` ``` bash rabbitmq-plugins directories -s ``` ### Enabling the Plugin Then run the following command: ``` bash rabbitmq-plugins enable rabbitmq_delayed_message_exchange ``` ## Usage ## To use the delayed-messaging feature, declare an exchange with the type `x-delayed-message`: ```java // ... elided code ... Map args = new HashMap(); args.put("x-delayed-type", "direct"); channel.exchangeDeclare("my-exchange", "x-delayed-message", true, false, args); // ... more code ... ``` Note that we pass an extra header called `x-delayed-type`, more on it under the _Routing_ section. Once we have the exchange declared we can publish messages providing a header telling the plugin for how long to delay our messages: ```java // ... elided code ... byte[] messageBodyBytes = "delayed payload".getBytes("UTF-8"); Map headers = new HashMap(); headers.put("x-delay", 5000); AMQP.BasicProperties.Builder props = new AMQP.BasicProperties.Builder().headers(headers); channel.basicPublish("my-exchange", "", props.build(), messageBodyBytes); byte[] messageBodyBytes2 = "more delayed payload".getBytes("UTF-8"); Map headers2 = new HashMap(); headers2.put("x-delay", 1000); AMQP.BasicProperties.Builder props2 = new AMQP.BasicProperties.Builder().headers(headers2); channel.basicPublish("my-exchange", "", props2.build(), messageBodyBytes2); // ... more code ... ``` In the above example we publish two messages, specifying the delay time with the `x-delay` header. For this example, the plugin will deliver to our queues first the message with the body `"more delayed payload"` and then the one with the body `"delayed payload"`. If the `x-delay` header is not present, then the plugin will proceed to route the message without delay. ## Routing ## This plugin allows for flexible routing via the `x-delayed-type` arguments that can be passed during `exchange.declare`. In the example above we used `"direct"` as exchange type. That means the plugin will have the same routing behavior shown by the direct exchange. If you want a different routing behavior, then you could provide a different exchange type, like `"topic"` for example. You can also specify exchange types provided by plugins. Note that this argument is **required** and **must** refer to an **existing exchange type**. ## Performance Impact ## Due to the `"x-delayed-type"` argument, one could use this exchange in place of other exchanges, since the `"x-delayed-message"` exchange will just act as proxy. Note that there might be some performance implications if you do this. For each message that crosses an `"x-delayed-message"` exchange, the plugin will try to determine if the message has to be expired by making sure the delay is within range, ie: `Delay > 0, Delay =< ?ERL_MAX_T` (In Erlang a timer can be set up to (2^32)-1 milliseconds in the future). If the previous condition holds, then the message will be persisted to Mnesia and some other logic will kick in to determine if this particular message delay needs to replace the current scheduled timer and so on. This means that while one _could_ use this exchange in place of a _direct_ or _fanout_ exchange (or any other exchange for that matter), _it will be slower_ than using the actual exchange. If you don't need to delay messages, then use the actual exchange. ## Limitations Delayed messages are stored in a Mnesia table (also see Limitations below) with a single disk replica on the current node. They will survive a node restart. While timer(s) that triggered scheduled delivery are not persisted, it will be re-initialised during plugin activation on node start. Obviously, only having one copy of a scheduled message in a cluster means that losing that node or disabling the plugin on it will lose the messages residing on that node. The plugin only performs one attempt at publishing each message but since publishing is local, in practice the only issue that may prevent delivery is the lack of queues (or bindings) to route to. Closely related to the above, the mandatory flag is not supported by this exchange: we cannot be sure that at the future publishing point in time * there is at least one queue we can route to * the original connection is still around to send a `basic.return` to Current design of this plugin doesn't really fit scenarios with a high number of delayed messages (e.g. 100s of thousands or millions). See [#72](https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/issues/72) for details. ## Disabling the Plugin ## You can disable this plugin by calling `rabbitmq-plugins disable rabbitmq_delayed_message_exchange` but note that **ALL DELAYED MESSAGES THAT HAVEN'T BEEN DELIVERED WILL BE LOST**. ## Building the Plugin ```shell PROJECT_VERSION=4.2.0-rc.1 PRODUCT_VERSION=4.2.0-rc.1 VERSION=4.2.0-rc.1 \ gmake dist PROJECT_VERSION=4.2.0-rc.1 PRODUCT_VERSION=4.2.0-rc.1 \ VERSION=4.2.0-rc.1 DIST_AS_EZS=true ``` The EZ file is created in the `plugins` directory. ## Creating a Release 1. Update `broker_version_requirements` in `helpers.bzl` & `Makefile` (Optional) 1. Update the plugin version in `MODULE.bazel` 1. Push a tag (i.e. `v4.0.0`) with the matching version 1. Allow the Release workflow to run and create a draft release 1. Review and publish the release ## LICENSE See the LICENSE file.