@@ -47,17 +47,28 @@ def __init__(self, url='amqp://guest:guest@localhost:5672//',
4747 self .listener_connection = None
4848 self .listener_channel = None
4949 self .listener_queue = None
50+
51+ self .aiopika_connection = None
52+ self .aiopika_channel = None
53+ self .aiopika_exchange = None
54+
5055 super ().__init__ (channel = channel , write_only = write_only , logger = logger )
5156
5257 async def _connection (self ):
53- return await aio_pika .connect_robust (self .url )
58+ if (self .aiopika_connection is None or self .aiopika_connection .is_closed is False ):
59+ return await aio_pika .connect_robust (self .url )
60+ return self .aiopika_connection
5461
5562 async def _channel (self , connection ):
56- return await connection .channel ()
63+ if self .aiopika_channel is None :
64+ return await connection .channel ()
65+ return self .aiopika_channel
5766
5867 async def _exchange (self , channel ):
59- return await channel .declare_exchange (self .channel ,
68+ if self .aiopika_exchange is None :
69+ return await channel .declare_exchange (self .channel ,
6070 aio_pika .ExchangeType .FANOUT )
71+ return self .aiopika_exchange
6172
6273 async def _queue (self , channel , exchange ):
6374 queue = await channel .declare_queue (durable = False ,
@@ -66,10 +77,10 @@ async def _queue(self, channel, exchange):
6677 return queue
6778
6879 async def _publish (self , data ):
69- connection = await self ._connection ()
70- channel = await self ._channel (connection )
71- exchange = await self ._exchange (channel )
72- await exchange .publish (
80+ self . aiopika_connection = await self ._connection ()
81+ self . aiopika_channel = await self ._channel (self . aiopika_connection )
82+ self . aiopika_exchange = await self ._exchange (self . aiopika_channel )
83+ await self . aiopika_exchange .publish (
7384 aio_pika .Message (body = pickle .dumps (data ),
7485 delivery_mode = aio_pika .DeliveryMode .PERSISTENT ),
7586 routing_key = '*'
0 commit comments