[ServiceBus] uamqp/pyamqp switch

sdk/servicebus/azure-servicebus/CHANGELOG.md

@@ -4,10 +4,20 @@
 
 ### Features Added
 
+- A new boolean keyword argument `uamqp_transport` has been added to sync and async `ServiceBusClient` constructors which indicates whether to use the `uamqp` library or the default pure Python AMQP library as the underlying transport.
+
 ### Bugs Fixed
 
+- Fixed a bug where sync and async `ServiceBusAdministrationClient` expected `credential` with `get_token` method returning `AccessToken.token` of type `bytes` and not `str`, now matching the documentation.
+- Fixed a bug where `raw_amqp_message.header` and `message.header` properties on `ServiceReceivedBusMessage` were returned with `durable`, `first_acquirer`, and `priority` properties set by default, rather than the values returned by the service.
+
 ### Other Changes
 
+- The `message` attribute on `ServiceBus`/`ServiceBusMessageBatch`/`ServiceBusReceivedMessage`, which previously exposed the `uamqp.Message`/`uamqp.BatchMessage`, has been deprecated.
+  - `LegacyMessage`/`LegacyBatchMessage` objects returned by the `message` attribute on `ServiceBus`/`ServiceBusMessageBatch` have been introduced to help facilitate the transition.
+- Removed uAMQP from required dependencies.
+- Adding `uamqp >= 1.6.3` as an optional dependency for use with the `uamqp_transport` keyword.
+
 ## 7.9.0 (Unreleased)
 
 ### Breaking Changes

sdk/servicebus/azure-servicebus/azure/servicebus/__init__.py

@@ -3,8 +3,6 @@
 # Licensed under the MIT License. See License.txt in the project root for
 # license information.
 # -------------------------------------------------------------------------
-from ._pyamqp import constants
-
 from ._version import VERSION
 
 __version__ = VERSION
@@ -23,6 +21,7 @@
     ServiceBusSubQueue,
     ServiceBusMessageState,
     ServiceBusSessionFilter,
+    TransportType,
     NEXT_AVAILABLE_SESSION,
 )
 from ._common.auto_lock_renewer import AutoLockRenewer
@@ -31,8 +30,6 @@
     ServiceBusConnectionStringProperties,
 )
 
-TransportType = constants.TransportType
-
 __all__ = [
     "ServiceBusMessage",
     "ServiceBusMessageBatch",

sdk/servicebus/azure-servicebus/azure/servicebus/_common/_configuration.py

@@ -1,50 +1,52 @@
-# --------------------------------------------------------------------------------------------
+# # --------------------------------------------------------------------------------------------
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
-from typing import Optional, Dict, Any
+from typing import Optional, Dict, Any, Union, TYPE_CHECKING
 from urllib.parse import urlparse
 
 from azure.core.pipeline.policies import RetryMode
-from .._pyamqp.constants import TransportType
-
-
-DEFAULT_AMQPS_PORT = 1571
-DEFAULT_AMQP_WSS_PORT = 443
-
+from .constants import DEFAULT_AMQPS_PORT, DEFAULT_AMQP_WSS_PORT, TransportType
+if TYPE_CHECKING:
+    from .._transport._base import AmqpTransport
+    from ..aio._transport._base_async import AmqpTransportAsync
 
 class Configuration(object):  # pylint:disable=too-many-instance-attributes
     def __init__(self, **kwargs):
-        self.user_agent = kwargs.get("user_agent")  # type: Optional[str]
-        self.retry_total = kwargs.get("retry_total", 3)  # type: int
+        self.user_agent: Optional[str] = kwargs.get("user_agent")
+        self.retry_total: int = kwargs.get("retry_total", 3)
         self.retry_mode = RetryMode(kwargs.get("retry_mode", 'exponential'))
-        self.retry_backoff_factor = kwargs.get(
+        self.retry_backoff_factor: float = kwargs.get(
             "retry_backoff_factor", 0.8
-        )  # type: float
-        self.retry_backoff_max = kwargs.get("retry_backoff_max", 120)  # type: int
-        self.logging_enable = kwargs.get("logging_enable", False)  # type: bool
-        self.http_proxy = kwargs.get("http_proxy")  # type: Optional[Dict[str, Any]]
+        )
+        self.retry_backoff_max: int = kwargs.get("retry_backoff_max", 120)
+        self.logging_enable: bool = kwargs.get("logging_enable", False)
+        self.http_proxy: Optional[Dict[str, Any]] = kwargs.get("http_proxy")
 
-        self.custom_endpoint_address = kwargs.get("custom_endpoint_address")  # type: Optional[str]
-        self.connection_verify = kwargs.get("connection_verify")  # type: Optional[str]
+        self.custom_endpoint_address: Optional[str] = kwargs.get("custom_endpoint_address")
+        self.connection_verify: Optional[str] = kwargs.get("connection_verify")
         self.connection_port = DEFAULT_AMQPS_PORT
         self.custom_endpoint_hostname = None
+        self.hostname = kwargs.pop("hostname")
+        amqp_transport: Union["AmqpTransport", "AmqpTransportAsync"] = kwargs.pop("amqp_transport")
 
         self.transport_type = (
             TransportType.AmqpOverWebsocket
             if self.http_proxy
             else kwargs.get("transport_type", TransportType.Amqp)
         )
         # The following configs are not public, for internal usage only
-        self.auth_timeout = kwargs.get("auth_timeout", 60)  # type: int
+        self.auth_timeout: float = kwargs.get("auth_timeout", 60)
         self.encoding = kwargs.get("encoding", "UTF-8")
         self.auto_reconnect = kwargs.get("auto_reconnect", True)
         self.keep_alive = kwargs.get("keep_alive", 30)
-        self.timeout = kwargs.get("timeout", 60)  # type: float
+        self.timeout: float = kwargs.get("timeout", 60)
 
         if self.http_proxy or self.transport_type == TransportType.AmqpOverWebsocket:
             self.transport_type = TransportType.AmqpOverWebsocket
             self.connection_port = DEFAULT_AMQP_WSS_PORT
+            if amqp_transport.KIND == "pyamqp":
+                self.hostname += "/$servicebus/websocket"
 
         # custom end point
         if self.custom_endpoint_address:
@@ -55,5 +57,7 @@ def __init__(self, **kwargs):
             endpoint = urlparse(self.custom_endpoint_address)
             self.transport_type = TransportType.AmqpOverWebsocket
             self.custom_endpoint_hostname = endpoint.hostname
+            if amqp_transport.KIND == "pyamqp":
+                self.custom_endpoint_address += "/$servicebus/websocket"
             # in case proxy and custom endpoint are both provided, we default port to 443 if it's not provided
             self.connection_port = endpoint.port or DEFAULT_AMQP_WSS_PORT

sdk/servicebus/azure-servicebus/azure/servicebus/_common/constants.py

@@ -5,7 +5,6 @@
 # -------------------------------------------------------------------------
 from enum import Enum
 from azure.core import CaseInsensitiveEnumMeta
-from .._pyamqp import constants
 
 VENDOR = b"com.microsoft"
 DATETIMEOFFSET_EPOCH = 621355968000000000
@@ -60,6 +59,8 @@
 JWT_TOKEN_SCOPE = "https://servicebus.azure.net//.default"
 USER_AGENT_PREFIX = "azsdk-python-servicebus"
 CONSUMER_IDENTIFIER = VENDOR + b":receiver-name"
+UAMQP_LIBRARY = "uamqp"
+PYAMQP_LIBRARY = "pyamqp"
 
 MANAGEMENT_PATH_SUFFIX = "/$management"
 
@@ -178,12 +179,6 @@ class ServiceBusMessageState(int, Enum):
     DEFERRED = 1
     SCHEDULED = 2
 
-# To enable extensible string enums for the public facing parameter, and translate to the "real" uamqp constants.
-ServiceBusToAMQPReceiveModeMap = {
-    ServiceBusReceiveMode.PEEK_LOCK: constants.ReceiverSettleMode.Second,
-    ServiceBusReceiveMode.RECEIVE_AND_DELETE: constants.ReceiverSettleMode.First,
-}
-
 
 class ServiceBusSessionFilter(Enum):
     NEXT_AVAILABLE = 0
@@ -195,3 +190,17 @@ class ServiceBusSubQueue(str, Enum, metaclass=CaseInsensitiveEnumMeta):
 
 
 NEXT_AVAILABLE_SESSION = ServiceBusSessionFilter.NEXT_AVAILABLE
+
+## all below - previously uamqp
+class TransportType(Enum):
+    """Transport type
+    The underlying transport protocol type:
+     Amqp: AMQP over the default TCP transport protocol, it uses port 5671.
+     AmqpOverWebsocket: Amqp over the Web Sockets transport protocol, it uses
+     port 443.
+    """
+    Amqp = 1
+    AmqpOverWebsocket = 2
+
+DEFAULT_AMQPS_PORT = 5671
+DEFAULT_AMQP_WSS_PORT = 443