Search code examples
clojurerabbitmqlangohr

RabbitMQ delayed_message_exchange declaration fails in langohr clojure client

I am trying to configure a RabbitMQ delayed message exchange as described in this document: https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/

The below is my attempt at translating the code example from that README into clojure/langohr. 

(let [conn (langohr.core/connect {:host "localhost" :port 61666})
      ch (langohr.channel/open conn)]
  (langohr.exchange/declare ch "my-exchange" "x-delayed-message"
                            {"x-delayed-type" "direct"}))

Sadly when I run this, I receive this error:

ShutdownSignalException channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - Invalid argument, 'x-delayed-type' must be an existing exchange type, class-id=40, method-id=10)  com.rabbitmq.client.impl.ChannelN.asyncShutdown (ChannelN.java:509)

Looking at the source for the rabbit_exchange_type_delayed_message plugin, I can see that my {"x-delayed-type" "direct"} seems to be falling through to the default hander in that case statement.

https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/blob/543efef2555338c46f776a89df3ee134066221e2/src/rabbit_exchange_type_delayed_message.erl#L72

I've no idea why this is happening though, or how to fix it.

Solution

  • It looks like you need to wrap the options map inside another :arguments map. Here's the langohr.exchange/declare code showing the call to .exchangeDeclare using the destructured arguments binding as a map.

    (let [conn (langohr.core/connect {:host "localhost" :port 61666})
      ch (langohr.channel/open conn)]
  (langohr.exchange/declare ch "my-exchange" "x-delayed-message"
                            {:arguments {"x-delayed-type" "direct"}}))