diff options
| -rw-r--r-- | kombu/common.py | 14 |
1 files changed, 7 insertions, 7 deletions
diff --git a/kombu/common.py b/kombu/common.py index 6ab33098..5cfb4816 100644 --- a/kombu/common.py +++ b/kombu/common.py @@ -112,11 +112,11 @@ def declaration_cached(entity, channel): return entity in channel.connection.client.declared_entities -def maybe_declare(entity, channel=None, retry=False, **retry_policy): +async def maybe_declare(entity, channel=None, retry=False, **retry_policy): """Declare entity (cached).""" if retry: - return _imaybe_declare(entity, channel, **retry_policy) - return _maybe_declare(entity, channel) + return await _imaybe_declare(entity, channel, **retry_policy) + return await _maybe_declare(entity, channel) def _ensure_channel_is_bound(entity, channel): @@ -135,7 +135,7 @@ def _ensure_channel_is_bound(entity, channel): return entity -def _maybe_declare(entity, channel): +async def _maybe_declare(entity, channel): # _maybe_declare sets name on original for autogen queues orig = entity @@ -156,7 +156,7 @@ def _maybe_declare(entity, channel): if not channel.connection: raise RecoverableConnectionError('channel disconnected') - entity.declare(channel=channel) + await entity.declare(channel=channel) if declared is not None and ident: declared.add(ident) if orig is not None: @@ -164,13 +164,13 @@ def _maybe_declare(entity, channel): return True -def _imaybe_declare(entity, channel, **retry_policy): +async def _imaybe_declare(entity, channel, **retry_policy): _ensure_channel_is_bound(entity, channel) if not entity.channel.connection: raise RecoverableConnectionError('channel disconnected') - return entity.channel.connection.client.ensure( + return await entity.channel.connection.client.ensure( entity, _maybe_declare, **retry_policy)(entity, channel) |