feat: add producer connectivity functionality

Author: nachatz

pulsar/asyncio.py

@@ -220,6 +220,12 @@ def last_sequence_id(self):
         message was ever published.
         """
         return self._producer.last_sequence_id()
+    
+    def is_connected(self) -> bool:
+        """
+        Check if the producer is connected or not.
+        """
+        return self._producer.is_connected()
 
 class Consumer:
     """

tests/asyncio_test.py

@@ -156,6 +156,13 @@ async def test_close_producer(self):
         except PulsarException as e:
             self.assertEqual(e.error(), pulsar.Result.AlreadyClosed)
 
+    async def test_producer_is_connected(self):
+        topic = f'asyncio-test-producer-is-connected-{time.time()}'
+        producer = await self._client.create_producer(topic)
+        self.assertTrue(producer.is_connected())
+        await producer.close()
+        self.assertFalse(producer.is_connected())
+
     async def _prepare_messages(self, producer: Producer) -> List[pulsar.MessageId]:
         msg_ids = []
         for i in range(5):