fix(servicebus): add create_mgmt_client_async to UamqpTransportAsync

_SessionBrowserAsync calls self._amqp_transport.create_mgmt_client_async(...) but the uamqp async transport never exposed it, so list_queue_sessions / list_subscription_sessions on the async client would AttributeError when uamqp_transport=True. Added the factory mirroring the sync UamqpTransport.create_mgmt_client implementation.

Author: EldertGrootenboerMS

sdk/servicebus/azure-servicebus/azure/servicebus/aio/_transport/_uamqp_transport_async.py

@@ -10,6 +10,7 @@
 try:
     from uamqp import (
         constants,
+        AMQPClientAsync,
         SendClientAsync,
         ReceiveClientAsync,
     )
@@ -64,6 +65,28 @@ async def close_connection_async(connection: "ConnectionAsync") -> None:
             """
             await connection.destroy_async()
 
+        @staticmethod
+        def create_mgmt_client_async(config: "Configuration", **kwargs: Any) -> "AMQPClientAsync":
+            """Creates and returns an async uamqp AMQPClient for management-only operations.
+
+            :param ~azure.servicebus._common._configuration.Configuration config: The configuration.
+            :keyword JWTTokenAuthAsync auth: Required.
+            :keyword retry_policy: Required.
+            :keyword str client_name: Required.
+            :keyword dict properties: Required.
+            :return: AMQPClientAsync
+            :rtype: ~uamqp.AMQPClientAsync
+            """
+            retry_policy = kwargs.pop("retry_policy")
+            return AMQPClientAsync(
+                "amqps://" + config.hostname,
+                debug=config.logging_enable,
+                error_policy=retry_policy,
+                keep_alive_interval=config.keep_alive,
+                encoding=config.encoding,
+                **kwargs,
+            )
+
         @staticmethod
         def create_send_client_async(config: "Configuration", **kwargs: Any) -> "SendClientAsync": # pylint:disable=docstring-keyword-should-match-keyword-only
             """