feat: add producer connectivity functionality (#286)

pulsar/asyncio.py

diff --git a/pulsar/asyncio.py b/pulsar/asyncio.py
index 34377d6..a1ca0c0 100644
--- a/pulsar/asyncio.py
+++ b/pulsar/asyncio.py
@@ -220,6 +220,12 @@
         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

diff --git a/tests/asyncio_test.py b/tests/asyncio_test.py
index d1c923a..4440809 100644
--- a/tests/asyncio_test.py
+++ b/tests/asyncio_test.py
@@ -157,6 +157,13 @@
         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):