Support user created loops on the asyncio client (Fixes miguelgrinber…

…g/Flask-SocketIO#1065)

engineio/asyncio_client.py

@@ -427,7 +427,8 @@ async def _ping_loop(self):
         interval.
         """
         self.pong_received = True
-        self.ping_loop_event.clear()
+        if self.ping_loop_event is None:
+            self.ping_loop_event = self.create_event()
         while self.state == 'connected':
             if not self.pong_received:
                 self.logger.info(
@@ -477,7 +478,8 @@ async def _read_loop_polling(self):
         self.logger.info('Waiting for write loop task to end')
         await self.write_loop_task
         self.logger.info('Waiting for ping loop task to end')
-        self.ping_loop_event.set()
+        if self.ping_loop_event:  # pragma: no cover
+            self.ping_loop_event.set()
         await self.ping_loop_task
         if self.state == 'connected':
             await self._trigger_event('disconnect', run_async=False)
@@ -512,7 +514,8 @@ async def _read_loop_websocket(self):
         self.logger.info('Waiting for write loop task to end')
         await self.write_loop_task
         self.logger.info('Waiting for ping loop task to end')
-        self.ping_loop_event.set()
+        if self.ping_loop_event:  # pragma: no cover
+            self.ping_loop_event.set()
         await self.ping_loop_task
         if self.state == 'connected':
             await self._trigger_event('disconnect', run_async=False)

engineio/client.py

@@ -90,7 +90,7 @@ def __init__(self,
         self.read_loop_task = None
         self.write_loop_task = None
         self.ping_loop_task = None
-        self.ping_loop_event = self.create_event()
+        self.ping_loop_event = None
         self.queue = None
         self.state = 'disconnected'
         self.ssl_verify = ssl_verify
@@ -515,7 +515,8 @@ def _ping_loop(self):
         interval.
         """
         self.pong_received = True
-        self.ping_loop_event.clear()
+        if self.ping_loop_event is None:
+            self.ping_loop_event = self.create_event()
         while self.state == 'connected':
             if not self.pong_received:
                 self.logger.info(
@@ -560,7 +561,8 @@ def _read_loop_polling(self):
         self.logger.info('Waiting for write loop task to end')
         self.write_loop_task.join()
         self.logger.info('Waiting for ping loop task to end')
-        self.ping_loop_event.set()
+        if self.ping_loop_event:  # pragma: no cover
+            self.ping_loop_event.set()
         self.ping_loop_task.join()
         if self.state == 'connected':
             self._trigger_event('disconnect', run_async=False)
@@ -595,7 +597,8 @@ def _read_loop_websocket(self):
         self.logger.info('Waiting for write loop task to end')
         self.write_loop_task.join()
         self.logger.info('Waiting for ping loop task to end')
-        self.ping_loop_event.set()
+        if self.ping_loop_event:  # pragma: no cover
+            self.ping_loop_event.set()
         self.ping_loop_task.join()
         if self.state == 'connected':
             self._trigger_event('disconnect', run_async=False)

tests/asyncio/test_asyncio_client.py

@@ -838,6 +838,7 @@ def test_ping_loop_disconnect(self):
         async def fake_wait():
             c.state, c.pong_received = states.pop(0)
 
+        c.ping_loop_event = c.create_event()
         c.ping_loop_event.wait = fake_wait
         _run(c._ping_loop())
         self.assertEqual(
@@ -858,6 +859,7 @@ def test_ping_loop_missing_pong(self):
         async def fake_wait():
             c.state, c.pong_received = states.pop(0)
 
+        c.ping_loop_event = c.create_event()
         c.ping_loop_event.wait = fake_wait
         _run(c._ping_loop())
         self.assertEqual(c.state, 'connected')
@@ -880,6 +882,7 @@ def test_ping_loop_missing_pong_websocket(self):
         async def fake_wait():
             c.state, c.pong_received = states.pop(0)
 
+        c.ping_loop_event = c.create_event()
         c.ping_loop_event.wait = fake_wait
         _run(c._ping_loop())
         self.assertEqual(c.state, 'connected')

tests/common/test_client.py

@@ -791,6 +791,7 @@ def fake_wait(timeout):
             self.assertEqual(timeout, 10)
             c.state, c.pong_received = states.pop(0)
 
+        c.ping_loop_event = c.create_event()
         c.ping_loop_event.wait = fake_wait
         c._ping_loop()
         self.assertEqual(
@@ -811,6 +812,7 @@ def fake_wait(timeout):
             self.assertEqual(timeout, 10)
             c.state, c.pong_received = states.pop(0)
 
+        c.ping_loop_event = c.create_event()
         c.ping_loop_event.wait = fake_wait
         c._ping_loop()
         self.assertEqual(c.state, 'connected')
@@ -832,6 +834,7 @@ def fake_wait(timeout):
             self.assertEqual(timeout, 10)
             c.state, c.pong_received = states.pop(0)
 
+        c.ping_loop_event = c.create_event()
         c.ping_loop_event.wait = fake_wait
         c._ping_loop()
         self.assertEqual(c.state, 'connected')