mirror of https://github.com/celery/kombu.git
Producer.auto_declare now declares on first publish
This commit is contained in:
parent
441a738102
commit
719cd7b2af
|
@ -47,9 +47,8 @@ class Producer(object):
|
||||||
#: Default compression method. Disabled by default.
|
#: Default compression method. Disabled by default.
|
||||||
compression = None
|
compression = None
|
||||||
|
|
||||||
#: By default the exchange is declared at instantiation.
|
#: By default, if a defualt exchange is set,
|
||||||
#: If you want to declare manually then you can set this
|
#: that exchange will be declare when publishing a message.
|
||||||
#: to :const:`False`.
|
|
||||||
auto_declare = True
|
auto_declare = True
|
||||||
|
|
||||||
#: Basic return callback.
|
#: Basic return callback.
|
||||||
|
@ -167,6 +166,10 @@ class Producer(object):
|
||||||
body, serializer, content_type, content_encoding,
|
body, serializer, content_type, content_encoding,
|
||||||
compression, headers)
|
compression, headers)
|
||||||
|
|
||||||
|
if self.auto_declare and self.exchange.name:
|
||||||
|
declare = [] if declare is None else declare
|
||||||
|
declare.append(self.exchange)
|
||||||
|
|
||||||
if retry:
|
if retry:
|
||||||
_publish = self.connection.ensure(self, _publish, **retry_policy)
|
_publish = self.connection.ensure(self, _publish, **retry_policy)
|
||||||
return _publish(
|
return _publish(
|
||||||
|
@ -225,10 +228,6 @@ class Producer(object):
|
||||||
if self.on_return:
|
if self.on_return:
|
||||||
self._channel.events['basic_return'].add(self.on_return)
|
self._channel.events['basic_return'].add(self.on_return)
|
||||||
self.exchange = self.exchange(channel)
|
self.exchange = self.exchange(channel)
|
||||||
if self.auto_declare:
|
|
||||||
# auto_decare is not recommended as this will force
|
|
||||||
# evaluation of the channel.
|
|
||||||
self.declare()
|
|
||||||
|
|
||||||
def __enter__(self):
|
def __enter__(self):
|
||||||
return self
|
return self
|
||||||
|
|
|
@ -59,6 +59,9 @@ class test_Producer:
|
||||||
assert p.exchange is not self.exchange
|
assert p.exchange is not self.exchange
|
||||||
assert p.exchange.is_bound
|
assert p.exchange.is_bound
|
||||||
# auto_declare declares exchange'
|
# auto_declare declares exchange'
|
||||||
|
assert 'exchange_declare' not in channel
|
||||||
|
|
||||||
|
p.publish('foo')
|
||||||
assert 'exchange_declare' in channel
|
assert 'exchange_declare' in channel
|
||||||
|
|
||||||
def test_manual_declare(self):
|
def test_manual_declare(self):
|
||||||
|
@ -124,12 +127,14 @@ class test_Producer:
|
||||||
def test_publish_with_Exchange_instance(self):
|
def test_publish_with_Exchange_instance(self):
|
||||||
p = self.connection.Producer()
|
p = self.connection.Producer()
|
||||||
p.channel = Mock()
|
p.channel = Mock()
|
||||||
|
p.channel.connection.client.declared_entities = set()
|
||||||
p.publish('hello', exchange=Exchange('foo'), delivery_mode='transient')
|
p.publish('hello', exchange=Exchange('foo'), delivery_mode='transient')
|
||||||
assert p._channel.basic_publish.call_args[1]['exchange'] == 'foo'
|
assert p._channel.basic_publish.call_args[1]['exchange'] == 'foo'
|
||||||
|
|
||||||
def test_publish_with_expiration(self):
|
def test_publish_with_expiration(self):
|
||||||
p = self.connection.Producer()
|
p = self.connection.Producer()
|
||||||
p.channel = Mock()
|
p.channel = Mock()
|
||||||
|
p.channel.connection.client.declared_entities = set()
|
||||||
p.publish('hello', exchange=Exchange('foo'), expiration=10)
|
p.publish('hello', exchange=Exchange('foo'), expiration=10)
|
||||||
properties = p._channel.prepare_message.call_args[0][5]
|
properties = p._channel.prepare_message.call_args[0][5]
|
||||||
assert properties['expiration'] == '10000'
|
assert properties['expiration'] == '10000'
|
||||||
|
@ -137,6 +142,8 @@ class test_Producer:
|
||||||
def test_publish_with_reply_to(self):
|
def test_publish_with_reply_to(self):
|
||||||
p = self.connection.Producer()
|
p = self.connection.Producer()
|
||||||
p.channel = Mock()
|
p.channel = Mock()
|
||||||
|
p.channel.connection.client.declared_entities = set()
|
||||||
|
assert not p.exchange.name
|
||||||
p.publish('hello', exchange=Exchange('foo'), reply_to=Queue('foo'))
|
p.publish('hello', exchange=Exchange('foo'), reply_to=Queue('foo'))
|
||||||
properties = p._channel.prepare_message.call_args[0][5]
|
properties = p._channel.prepare_message.call_args[0][5]
|
||||||
assert properties['reply_to'] == 'foo'
|
assert properties['reply_to'] == 'foo'
|
||||||
|
@ -151,6 +158,7 @@ class test_Producer:
|
||||||
def test_publish_retry_calls_ensure(self):
|
def test_publish_retry_calls_ensure(self):
|
||||||
p = Producer(Mock())
|
p = Producer(Mock())
|
||||||
p._connection = Mock()
|
p._connection = Mock()
|
||||||
|
p._connection.declared_entities = set()
|
||||||
ensure = p.connection.ensure = Mock()
|
ensure = p.connection.ensure = Mock()
|
||||||
p.publish('foo', exchange='foo', retry=True)
|
p.publish('foo', exchange='foo', retry=True)
|
||||||
ensure.assert_called()
|
ensure.assert_called()
|
||||||
|
|
Loading…
Reference in New Issue