Skip to content

Commit

Permalink
Merge pull request #637 from Lancetnik/master
Browse files Browse the repository at this point in the history
fix: raise ChannelInvalidStateError at exchange.publish with closed channel
  • Loading branch information
mosquito authored Aug 5, 2024
2 parents 9c1ee8f + 4e93dc4 commit f41c28c
Show file tree
Hide file tree
Showing 2 changed files with 33 additions and 0 deletions.
5 changes: 5 additions & 0 deletions aio_pika/exchange.py
Original file line number Diff line number Diff line change
Expand Up @@ -190,6 +190,11 @@ async def publish(
f"Can not publish to internal exchange: '{self.name}'!",
)

if self.channel.is_closed:
raise aiormq.exceptions.ChannelInvalidStateError(
"%r closed" % self.channel,
)

channel = await self.channel.get_underlay_channel()
return await channel.basic_publish(
exchange=self.name,
Expand Down
28 changes: 28 additions & 0 deletions tests/test_amqp.py
Original file line number Diff line number Diff line change
Expand Up @@ -339,6 +339,34 @@ async def test_simple_publish_and_receive_to_bound_exchange(

await queue.unbind(dest_exchange, routing_key)

async def test_simple_publish_with_closed_channel(
self,
connection: aio_pika.Connection,
declare_exchange: Callable,
declare_queue: Callable,
):
routing_key = get_random_name()

channel = await connection.channel(publisher_confirms=False)

exchange = await declare_exchange(
"direct", auto_delete=True, channel=channel,
)

await connection.close()

body = bytes(shortuuid.uuid(), "utf-8")

with pytest.raises(aiormq.exceptions.ChannelInvalidStateError):
await exchange.publish(
Message(
body,
content_type="text/plain",
headers={"foo": "bar"},
),
routing_key,
)

async def test_incoming_message_info(
self,
channel: aio_pika.Channel,
Expand Down

0 comments on commit f41c28c

Please sign in to comment.