lud

lud

Hi everyone!

I wrote a unit test where I want to ensure that the app is able to connect to RabbitMQ (using Broadway), so the test will start Broadway, then publish to some exchange, and expect a message from the Broadway processor.

The problem is that it takes time for the Broadway producer to declare the exchange and bind the queue, so in my test when I publish the message it is not routed.

My current solution is to add a small sleep but it is not reliable (and slows down tests).

{:ok, rmq_consumer} = consumer_mod.start_link(job_handler: job_handler, start_mode: :normal)

# Generate a random string to validate that we receive this specific
# message.

randstr = Ecto.UUID.generate()
payload = Jason.encode!(%{"randstr" => randstr})

# Connect to RMQ and deliver the payload to the configured exchange.
# Currently I have not figured a way to ensure that the queue was declared
# and to republish otherwise, so we sleep for one second.

config = consumer_mod.config()
{:ok, conn} = AMQP.Connection.open(config.connection)
{:ok, chan} = AMQP.Channel.open(conn)
Process.sleep(1000)
:ok = ensure_routed_publish(chan, config.exchange, "", payload)

# We should receive the same random string from our test handler

assert_receive {:rmq_str, ^randstr}, 1000

I tried to fiddle with AMQP.Confirm.select, the :mandatory option, the wait_confirm_or_die function, but it does not work. Everytime AMQP will happily tell me that everything is fine though my message will never be routed as with select I can see the error in the logs.

How can I write the ensure_routed_publish/4 function so it returns {:error, _} when the message is not routed so I can retry it and avoid the sleeping?

Thank you.

Showing Posts 1 to 5

BradS2S

BradS2S

You can use return to register a handler to deal with returned messages.

If you get this message: {:basic_return, payload, meta}, send it again.

lud

lud OP

Hi, thank you.

Unfortunately this does not work because I receive a message only in case there was an error. So when everything goes right i’ll have to add a timeout to the receive, which is only slightly better than the sleep.

Edit: It seems that if I also add the following:

    AMQP.Confirm.select(chan)
    AMQP.Confirm.register_handler(chan, self())

the :basic_ack message is always delivered after the :basic_return message, so it may be safe to consider that receiving basic_ack without basic_return means that the message was routed.

BradS2S

BradS2S

Would something like this work?

Def ensure_routed_publish(chan, exchange, routing_key, payload) do
AMQP.Basic.return(chan, self())
AMQP.Confirm.select(chan)
AMQP.Confirm.register_handler(chan, self())
:ok = AMQP.Basic.publish(chan, exchange, routing_key, payload, mandatory: true)

receive do
{:basic_ack, _, _} →
:ok
{:basic_return, returned_payload, _} →
{:error, returned_payload}
after
timeout →
{:error, :timeout}
end
end

lud

lud OP

Yes this is more or less what I did:

  defp ensure_routed_publish(chan, exchange, key, payload) do
    AMQP.Confirm.select(chan)
    AMQP.Confirm.register_handler(chan, self())
    AMQP.Basic.return(chan, self())
    publish_loop(chan, exchange, key, payload)
  end

  defp publish_loop(chan, exchange, key, payload) do
    :ok = AMQP.Basic.publish(chan, exchange, key, payload, mandatory: true)

    receive do
      {:basic_return, _, %{reply_text: "NO_ROUTE"}} ->
        receive do
          {:basic_ack, _, _} ->
            Process.sleep(50)
            publish_loop(chan, exchange, key, payload)
        end

      {:basic_ack, _, _} ->
        :ok
    after
      5000 -> flunk("could not deliver message to RabbitMQ")
    end
  end

(It is in a test so we know the queue will exist at some point.)

lud

lud OP

Hi!

Do you know if there is a way to catch when an exchange does not exist?

operation basic.publish caused a channel exception not_found: no exchange ‘does_not_exist’ in vhost ‘/’

I have the following code:

    :ok = AMQP.Confirm.select(chan)
    :ok = AMQP.Confirm.register_handler(chan, self())

And was expecting to receive a basic nack.

I also have this code AMQP.Basic.return(chan, self()) but it does not seem to handle that.
Thank you!

— All posts loaded —

Where Next? Top

Trending in Questions Top

katta
I having some trouble figuring out if I have set myself too strict of standards for my production server. Currently I can handle 75% of r...
New
achenet
Hello, I’m trying to build a basic Phoenix web-app, and I’d like to use Tailwind. However, when I launch mix phx.server, I get an error...
New
bradley
I really like the adapter patterns that ecto, nebulex, waffle, etc. use and would love find something similar for a key management servic...
New
unaware8150
Hello folks! So at work, we are seeing some situations where we have to define some “fixed” strings that are used across the codebase in...
New
Cxx-mlr
I’m working on a small exercise involving update_in/3, and I came up with this solution: data = %{ name: "Periodic Table", category:...
New
ChrisAmelia
I’ve got trouble wrapping my head around the order in which functions are called in this snippet (from Phoenix’s authentication): toke...
New
dillonoconnor
Is there any way to avoid the Hologram compiler running when using iex? It seems like the front-end code could potentially be disregarded...
New

Other Trending Topics Top

GenericJam
Edit: 2026 May 15 - This post is archived. Mob is alive!! Main docs: mob v0.7.11 — Documentation A bit of explanation for the slightly c...
New
garrison
Hobbes is a low-level distributed database for the Elixir programming language. Hobbes provides a simple, safe, and scalable storage lay...
New
budgie
A little off-topic, but I feel like people here have a good head on their shoulders. I used to be quite good at making software. Was luc...
New
KristerV
Hey. Is there anyone here who creates agents in their apps? Not talking about using agents, but creating them. I’m finding it pretty diff...
New
mudasobwa
I fully migrated to my own harness from Anthropic/Gemini and I think it’s time to share it. Welcome DSH, the DeepSeek Harness, fully writ...
New
mcass19
ExRatatui lets you cook up rich terminal UIs in Elixir, powered by Rust’s ratatui via Rustler NIFs. Build interactive terminal applicatio...
New

We're in Beta

About us Mission Statement

Options

Thread Display Mode




Thread Preview

Skip Thread Previews