Kombu: Declaration problem of a queue with default exchange

[takaomag]

Hi,

I want to use a simple queue which is bound to only "AMQP default exchange (with empty name). But declaration of the queue fails.

from kombu import Connection, Queue, connections
from kombu.common import maybe_declare

connection = Connection()
with connections[connection].acquire(block=True) as conn:
    ch = conn.channel()
    maybe_declare(entity=Queue('my_queue'), channel=ch)


Traceback (most recent call last):
  File "<input>", line 3, in <module>
  File "/opt/airlink0/lib/python2.7/site-packages/kombu/common.py", line 90, in maybe_declare
    return _maybe_declare(entity)
  File "/opt/airlink0/lib/python2.7/site-packages/kombu/common.py", line 99, in _maybe_declare
    entity.declare()
  File "/opt/airlink0/lib/python2.7/site-packages/kombu/entity.py", line 470, in declare
    self.exchange.declare(nowait)
  File "/opt/airlink0/lib/python2.7/site-packages/kombu/entity.py", line 159, in declare
    passive=passive)
  File "/opt/airlink0/lib/python2.7/site-packages/amqp/channel.py", line 603, in exchange_declare
    (40, 11),  # Channel.exchange_declare_ok
  File "/opt/airlink0/lib/python2.7/site-packages/amqp/abstract_channel.py", line 71, in wait
    return self.dispatch_method(method_sig, args, content)
  File "/opt/airlink0/lib/python2.7/site-packages/amqp/abstract_channel.py", line 88, in dispatch_method
    return amqp_method(self, args)
  File "/opt/airlink0/lib/python2.7/site-packages/amqp/channel.py", line 224, in _close
    raise ChannelError(reply_code, reply_text, (class_id, method_id))
ChannelError: 403: (ACCESS_REFUSED - operation not permitted on the default exchange, (40, 10), None)

I think the "kombu.transport.pyamqp.Channel" class should override both "exchange_declare" method and "queue_bind" method of "amqp.channel.Channel" class, and should do nothing in case that exchange's name is empty string.

class Channel(amqp.Channel, base.StdChannel):

    def exchange_declare(
        self,
        exchange,
        type,
        passive = False,
        durable = False,
        auto_delete = True,
        nowait = False,
        arguments = None
    ):
        if exchange == '':
            return None # Or an object of required type. Sorry I don't know what kind of object must be returned in this method.

        return super(Channel, self).exchange_declare(
            exchange = exchange,
            type = type,
            passive = passive,
            durable = durable,
            auto_delete = auto_delete,
            nowait = nowait,
            arguments = arguments
        )

    def queue_bind(
        self,
        queue,
        exchange = '',
        routing_key = '',
        nowait = False,
        arguments = None,
    ):
        if exchange == '':
            return None # Or an object of required type. Sorry I don't know what kind of object must be returned in this method.

        return super(Channel, self).queue_bind(
            queue = queue,
            exchange = exchange,
            routing_key = routing_key,
            nowait = nowait,
            arguments = arguments
        )

Is it right?

[ask]

On Mar 12, 2013, at 10:45 AM, takaomag [email protected] wrote:

Hi,

I want to use a simple queue which is bound to only "AMQP default exchange (with empty name). But declaration of the queue fails.

from kombu import Connection, Queue, connections
from kombu.common import maybe_declare

connection = Connection()
with connections[connection].acquire(block=True) as conn:

ch = conn.channel()

maybe_declare(entity=Queue('my_queue'), channel=ch)

Traceback (most recent call last):

File "", line 3, in

File "/opt/airlink0/lib/python2.7/site-packages/kombu/common.py", line 90, in maybe_declare

return _maybe_declare(entity)

File "/opt/airlink0/lib/python2.7/site-packages/kombu/common.py", line 99, in _maybe_declare

entity.declare()

File "/opt/airlink0/lib/python2.7/site-packages/kombu/entity.py", line 470, in declare

self.exchange.declare(nowait)

File "/opt/airlink0/lib/python2.7/site-packages/kombu/entity.py", line 159, in declare

passive=passive)

File "/opt/airlink0/lib/python2.7/site-packages/amqp/channel.py", line 603, in exchange_declare

(40, 11), # Channel.exchange_declare_ok

File "/opt/airlink0/lib/python2.7/site-packages/amqp/abstract_channel.py", line 71, in wait

return self.dispatch_method(method_sig, args, content)

File "/opt/airlink0/lib/python2.7/site-packages/amqp/abstract_channel.py", line 88, in dispatch_method

return amqp_method(self, args)

File "/opt/airlink0/lib/python2.7/site-packages/amqp/channel.py", line 224, in _close

raise ChannelError(reply_code, reply_text, (class_id, method_id))
ChannelError: 403: (ACCESS_REFUSED - operation not permitted on the default exchange, (40, 10), None)
I think the "kombu.transport.pyamqp.Channel" class should override both "exchange_declare" method and "queue_bind" method of "amqp.channel.Channel" class, and should do nothing in case that exchange's name is empty string.

I think a better solution would be if Exchange.declare did nothing if the
exchange name was empty.

Ask Solem
twitter.com/asksol

[takaomag]

Thanks, your solution would be better !

[ask]

Thanks! Fixed in above commit

[jhalcrow]

This issue still exists, but with a different error. When I run it now I get:

queue.bind: server channel error 403, message: ACCESS_REFUSED - operation not permitted on the default exchange

Since kombu still attempts to bind the queue on the default exchange (which is forbidden by the AMQP spec). To fix it, Queue could check to see if its exchange has a name before calling queue_bind.

[lang100]

there is no need to bind queues to default exchange as explained here:

The default exchange is a direct exchange with no name (empty string) pre-declared by the broker. It has one special property that makes it very useful for simple applications: every queue that is created is automatically bound to it with a routing key which is the same as the queue name.