Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
await channel1.close()
await channel2.close()
channel = await amqp_connection.channel()
queue = asyncio.Queue()
deaclare_ok = await channel.queue_declare(auto_delete=True)
consume_ok = await channel.basic_consume(deaclare_ok.queue, queue.put)
await channel.basic_publish(b"foo", routing_key=deaclare_ok.queue)
message = await queue.get() # type: DeliveredMessage
assert message.body == b"foo"
with pytest.raises(aiormq.exceptions.PublishError) as e:
await channel.basic_publish(
b"bar", routing_key=deaclare_ok.queue + "foo", mandatory=True
)
message = e.value.message
assert message.delivery.routing_key == deaclare_ok.queue + "foo"
assert message.body == b"bar"
assert "'NO_ROUTE' for routing key" in repr(e.value)
cancel_ok = await channel.basic_cancel(consume_ok.consumer_tag)
assert cancel_ok.consumer_tag == consume_ok.consumer_tag
await channel.queue_delete(deaclare_ok.queue)
deaclare_ok = await channel.queue_declare(auto_delete=True)
await channel.basic_publish(b"foo bar", routing_key=deaclare_ok.queue)
def _credentials_class(start_frame: spec.Connection.Start):
for mechanism in start_frame.mechanisms.decode().split():
with suppress(KeyError):
return AuthMechanism[mechanism]
raise exc.AuthenticationError(
start_frame.mechanisms, [m.name for m in AuthMechanism]
)
def __exception_by_code(frame: spec.Channel.Close): # pragma: nocover
if frame.reply_code == 403:
return exc.ChannelAccessRefused(frame.reply_text)
elif frame.reply_code == 404:
return exc.ChannelNotFoundEntity(frame.reply_text)
elif frame.reply_code == 405:
return exc.ChannelLockedResource(frame.reply_text)
elif frame.reply_code == 406:
return exc.ChannelPreconditionFailed(frame.reply_text)
else:
return exc.ChannelClosed(frame.reply_code, frame.reply_text)
try:
self.reader, self.writer = await asyncio.open_connection(
self.url.host, self.url.port, ssl=ssl_context
)
except OSError as e:
raise ConnectionError(*e.args) from e
try:
protocol_header = ProtocolHeader()
self.writer.write(protocol_header.marshal())
res = await self.__receive_frame()
_, _, frame = res # type: spec.Connection.Start
self.heartbeat_last_received = self.loop.time()
except EOFError as e:
raise exc.IncompatibleProtocolError(*e.args) from e
credentials = self._credentials_class(frame)
self.server_properties = frame.server_properties
# noinspection PyTypeChecker
self.connection_tune = await self.__rpc(
spec.Connection.StartOk(
client_properties=self._client_properties(
**(client_properties or {})
),
mechanism=credentials.name,
response=credentials.value(self).marshal(),
)
) # type: spec.Connection.Tune
def __exception_by_code(frame: spec.Connection.Close):
if frame.reply_code == 501:
return exc.ConnectionFrameError(frame.reply_text)
elif frame.reply_code == 502:
return exc.ConnectionSyntaxError(frame.reply_text)
elif frame.reply_code == 503:
return exc.ConnectionCommandInvalid(frame.reply_text)
elif frame.reply_code == 504:
return exc.ConnectionChannelError(frame.reply_text)
elif frame.reply_code == 505:
return exc.ConnectionUnexpectedFrame(frame.reply_text)
elif frame.reply_code == 506:
return exc.ConnectionResourceError(frame.reply_text)
elif frame.reply_code == 530:
return exc.ConnectionNotAllowed(frame.reply_text)
elif frame.reply_code == 540:
return exc.ConnectionNotImplemented(frame.reply_text)
elif frame.reply_code == 541:
return exc.ConnectionInternalError(frame.reply_text)
else:
return exc.ConnectionClosed(frame.reply_code, frame.reply_text)
def lock(self):
if self.is_closed:
raise exc.ChannelInvalidStateError("%r closed" % self)
return self.__lock
message_id = message.header.properties.message_id
delivery_tag = self.message_id_delivery_tag.get(message_id)
if delivery_tag is None: # pragma: nocover
log.error("Unhandled message %r returning", message)
return
confirmation = self.confirmations.pop(delivery_tag, None)
if confirmation is None: # pragma: nocover
return
self.confirmations[delivery_tag] = self.Returning
if self.on_return_raises:
confirmation.set_exception(exc.PublishError(message, frame))
return
for cb in self.on_return_callbacks:
try:
cb(message)
except Exception:
log.exception("Unhandled return callback exception")
confirmation.set_result(message)