feat: add shutdown support to async client

pulsar/asyncio.py

@@ -777,6 +777,16 @@ async def subscribe(self, topic: Union[str, List[str]],
         schema.attach_client(self._client)
         return Consumer(await future, schema)
 
+    def shutdown(self) -> None:
+        """
+        Shutdown the client and all the associated producers and consumers
+
+        Raises
+        ------
+        PulsarException
+        """
+        self._client.shutdown()
+
     async def get_topic_partitions(self, topic: str) -> List[str]:
         """
         Get the list of partitions for a given topic in asynchronous mode.

tests/asyncio_test.py

@@ -227,6 +227,17 @@ async def test_producer_is_connected(self):
         await producer.close()
         self.assertFalse(producer.is_connected())
 
+    async def test_shutdown_client(self):
+        producer = await self._client.create_producer("persistent://public/default/partitioned_topic_name_test")
+        await producer.send(b"hello")
+        self._client.shutdown()
+
+        try:
+            await producer.send(b"hello")
+            self.fail("Expected AlreadyClosed exception after client shutdown")
+        except PulsarException as e:
+            self.assertEqual(e.error(), pulsar.Result.AlreadyClosed)
+
     async def _prepare_messages(self, producer: Producer) -> List[pulsar.MessageId]:
         msg_ids = []
         for i in range(5):