Polyconseil / aioamqp

AMQP implementation using asyncio
Other
280 stars 88 forks source link

Publishing to non existing exchange #116

Closed chemiron closed 5 years ago

chemiron commented 7 years ago

Hi, according to the documentation https://www.rabbitmq.com/amqp-0-9-1-reference.html#basic.publish basic_publish method have to return not-found error code if the exchange doesn't exist, but it returns nothing and doesn't raise any exception:

await channel.basic_publish(
       payload=pickle.dumps(request),
       exchange_name='non_existing_exc',
       routing_key=q_name)

is it an issue or it have to work the such way ? and also is there any way how I could to check if the exchange or queue exists ? Thanks

dzen commented 7 years ago

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 commented 7 years ago

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

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 commented 6 years ago

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 commented 5 years ago

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 '/'")