added test for subscribe connection error

pulsar/apps/data/channels.py

@@ -186,6 +186,7 @@ async def _connect(self, next_time):
             )
             await gather(*[c.connect() for c in self.channels.values()])
         except ConnectionRefusedError:
+            self.status = StatusType.disconnected
             next_time = backoff(next_time) if next_time else RECONNECT_LAG
             self.logger.critical(
                 '%s cannot subscribe - connection error - '
@@ -194,7 +195,7 @@ async def _connect(self, next_time):
                 next_time
             )
             self._loop.call_later(next_time,
-                                  self._loop.make_task,
+                                  self._loop.create_task,
                                   self.connect(next_time))
 
 

tests/stores/channels.py

@@ -53,6 +53,27 @@ def fire(channel, event, data):
         await channels.close()
         self.assertEqual(channels.status, StatusType.closed)
 
+    async def test_fail_subscribe(self):
+        channels = self.channels()
+        original, warning, critical = self._patch(
+            channels, channels.pubsub, 'subscribe'
+        )
+        await channels.connect()
+        args, kw = await critical.end
+        self.assertEqual(len(args), 3)
+        self.assertEqual(args[1], channels)
+        self.assertEqual(args[2], 2)
+        critical.end = create_future()
+        args, kw = await critical.end
+        self.assertEqual(len(args), 3)
+        self.assertEqual(args[1], channels)
+        self.assertEqual(args[2], 2.25)
+        channels.pubsub.subscribe = original
+        args, kw = await warning.end
+        self.assertEqual(len(args), 3)
+        self.assertEqual(args[1], channels)
+        self.assertEqual(args[2], channels.status_channel)
+
     async def test_fail_publish(self):
         channels = self.channels()
         original, warning, critical = self._patch(