Publishing to non existing exchange about aioamqp

[dzen]

Hello @chemiron

Well if the doc says so .. it's a bug.
Alternatively, you can call the declare method with passive=True : if the resource does not exist it will raise.
for instance : https://groups.google.com/forum/#!topic/rabbitmq-users/ZTfVwe_HYXc

[mwfrojdman]

There's no way to just check for existence with AMQP, one can just assert. Ie, if exchange does exist, then it works, if not, channel is closed with parameters implying the reason.

The same goes for exchange.declare with passive=True. Channel is closed if it doesn't exist.

So if you want to check if an exchange exists, open a new channel, exchange.declare (with passive or without depending on if you want to create the exchange if it doesn't exist) and then

• receive exchange.declare-ok if it was there, close channel yourself
• get channel.close from broker with parameters class = exchange and method = declare, and probably reply_code 404 (for not found).

If implementing that to a library, it would probably have to be a method of the connection, because of the opening and closing of a new one shot channel.

[jkr78]

I can confirm that publishing to non existing exchange (amqp ver. 0.11.0), while in the consumer, closes channel (as it should be from server side) but lib does not fire exception nor on_error callback. So where is no way how to find out if publish succeeded except check for .is_open.

[dzen]

Hello,

Sorry for the late response, but you have to call confirm_select:

#!/usr/bin/env python
"""
    Rabbitmq.com pub/sub example

    https://www.rabbitmq.com/tutorials/tutorial-four-python.html
"""

import asyncio
import aioamqp

import sys


async def exchange_routing():
    try:
        transport, protocol = await aioamqp.connect('localhost', 5672)
    except aioamqp.AmqpClosedConnection:
        print("closed connections")
        return

    channel = await protocol.channel()
    await channel.confirm_select()
    exchange_name = 'direct_logs'
    severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
    message = ' '.join(sys.argv[2:]) or 'Hello World!'

    #await channel.exchange(exchange_name, 'direct')

    pu = await channel.publish(message, exchange_name=exchange_name, routing_key=severity)
    print(" [x] Sent %r" % (message,))

    await protocol.close()
    transport.close()


asyncio.get_event_loop().run_until_complete(exchange_routing())

which produce :

$ python examples/emit_log_direct.py
only PLAIN login_method is supported, falling back to AMQPLAIN
Traceback (most recent call last):
  File "examples/emit_log_direct.py", line 37, in <module>
    asyncio.get_event_loop().run_until_complete(exchange_routing())
  File "/usr/local/lib/python3.6/asyncio/base_events.py", line 468, in run_until_complete
    return future.result()
  File "examples/emit_log_direct.py", line 29, in exchange_routing
    pu = await channel.publish(message, exchange_name=exchange_name, routing_key=severity)
  File "/home/benoit/Projects/blue/aioamqp/aioamqp/channel.py", line 706, in publish
    await fut
aioamqp.exceptions.ChannelClosed: (404, "NOT_FOUND - no exchange 'direct_logs' in vhost '/'")