[EventHubs] merge pyamqp to main

sdk/eventhub/azure-eventhub/CHANGELOG.md

@@ -1,6 +1,6 @@
 # Release History
 
-## 5.10.2 (Unreleased)
+## 5.11.0 (Unreleased)
 
 ### Features Added
 
@@ -88,9 +88,66 @@ This version and all future versions will require Python 3.7+, Python 3.6 is no
 
 ## 5.9.0b1 (2022-02-09)
 
+- The following features have been temporarily pulled out of async `EventHubProducerClient` and `EventHubConsumerClient` which will be added back in future previews as we work towards a stable release:
+  - Passing the following keyword arguments to the constructors and `from_connection_string` methods of the `EventHubProducerClient` and `EventHubConsumerClient` is not supported:  `transport_type`, `http_proxy`, `custom_endpoint_address`, and `connection_verify`.
+
+## 5.8.0b2 (2022-10-11)
+
+### Features Added
+
+- Updated the optional dependency for async transport using AMQP over WebSocket from `websocket-client` to `aiohttp` (Issue #24315, thanks @hansmbakker for the suggestion).
+
+## 5.8.0b1 (2022-09-22)
+
+This version and all future versions will require Python 3.7+. Python 3.6 is no longer supported.
+
+### Other Changes
+
+- Added the `uamqp_transport` optional parameter to the clients, to allow switching to the `uamqp` library as the transport.
+
+## 5.8.0a5 (2022-07-19)
+
+### Bugs Fixed
+
+- Fixed bug that prevented token refresh at regular intervals.
+- Fixed bug that was improperly passing the debug keyword argument, so that network trace debug logs are output when requested.
+
+### Other Changes
+
+- Added logging added in to track proper token refreshes & fetches, output exception reason for producer init failure.
+
+## 5.8.0a4 (2022-06-07)
+
+### Features Added
+
+- Added support for connection using websocket and http proxy.
+- Added support for custom endpoint connection over websocket.
+
+## 5.8.0a3 (2022-03-08)
+
+### Other Changes
+
+- Improved the performance of async sending and receiving.
+
+## 5.8.0a2 (2022-02-09)
+
 ### Features Added
 
-- The classmethod `from_message_data` has been added to `EventData` for interoperability with the Schema Registry Avro Encoder library, and takes `data` and `content_type` as positional parameters.
+- Added support for async `EventHubProducerClient` and `EventHubConsumerClient`.
+
+## 5.8.0a1 (2022-01-13)
+
+Version 5.8.0a1 is our first efforts to build an Azure Event Hubs client library based on pure python implemented AMQP stack.
+
+### Breaking changes
+
+- The following features have been temporarily pulled out which will be added back in future previews as we work towards a stable release:
+  - Async is not supported.
+  - Passing the following keyword arguments to the constructors and `from_connection_string` methods of the `EventHubProducerClient` and `EventHubConsumerClient` is not supported:  `transport_type`, `http_proxy`, `custom_endpoint_address`, and `connection_verify`.
+
+### Other Changes
+
+- uAMQP dependency is removed.
 
 ## 5.7.0 (2022-01-12)
 
@@ -598,4 +655,6 @@ Version 5.0.0b1 is a preview of our efforts to create a client library that is u
 - Further testing and minor bug fixes.
 
 
-![Impressions](https://azure-sdk-impressions.azurewebsites.net/api/impressions/azure-sdk-for-python/sdk/eventhub/azure-eventhub/HISTORY.png)
+## 0.2.0a2 (2018-04-02)
+
+- Updated uAQMP dependency.

sdk/eventhub/azure-eventhub/azure/eventhub/_buffered_producer/_buffered_producer.py

@@ -32,6 +32,7 @@ def __init__(
         max_message_size_on_link: int,
         executor: ThreadPoolExecutor,
         *,
+        amqp_transport: AmqpTransport,
         max_buffer_length: int,
         max_wait_time: float = 1
     ):
@@ -50,10 +51,11 @@ def __init__(
         self._max_message_size_on_link = max_message_size_on_link
         self._check_max_wait_time_future = None
         self.partition_id = partition_id
+        self._amqp_transport = amqp_transport
 
     def start(self):
         with self._lock:
-            self._cur_batch = EventDataBatch(self._max_message_size_on_link)
+            self._cur_batch = EventDataBatch(self._max_message_size_on_link, amqp_transport=self._amqp_transport)
             self._running = True
             if self._max_wait_time:
                 self._last_send_time = time.time()
@@ -113,12 +115,12 @@ def put_events(self, events, timeout_time=None):
                     self._buffered_queue.put(self._cur_batch)
                 self._buffered_queue.put(events)
             # create a new batch for incoming events
-            self._cur_batch = EventDataBatch(self._max_message_size_on_link)
+            self._cur_batch = EventDataBatch(self._max_message_size_on_link, amqp_transport=self._amqp_transport)
         except ValueError:
             # add single event exceeds the cur batch size, create new batch
             with self._lock:
                 self._buffered_queue.put(self._cur_batch)
-            self._cur_batch = EventDataBatch(self._max_message_size_on_link)
+            self._cur_batch = EventDataBatch(self._max_message_size_on_link, amqp_transport=self._amqp_transport)
             self._cur_batch.add(events)
         with self._lock:
             self._cur_buffered_len += new_events_len
@@ -197,7 +199,7 @@ def flush(self, timeout_time=None, raise_error=True):
             self._last_send_time = time.time()
             #reset buffered count
             self._cur_buffered_len = 0
-            self._cur_batch = EventDataBatch(self._max_message_size_on_link)
+            self._cur_batch = EventDataBatch(self._max_message_size_on_link, amqp_transport=self._amqp_transport)
             _LOGGER.info("Partition %r finished flushing.", self.partition_id)
 
     def check_max_wait_time_worker(self):

sdk/eventhub/azure-eventhub/azure/eventhub/_buffered_producer/_buffered_producer_dispatcher.py

@@ -14,8 +14,8 @@
 from ..exceptions import EventDataSendError, ConnectError, EventHubError
 
 if TYPE_CHECKING:
-    from .._producer_client import SendEventTypes
     from .._transport._base import AmqpTransport
+    from .._producer_client import SendEventTypes
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -31,6 +31,7 @@ def __init__(
         eventhub_name: str,
         max_message_size_on_link: int,
         *,
+        amqp_transport: AmqpTransport,
         max_buffer_length: int = 1500,
         max_wait_time: float = 1,
         executor: Optional[Union[ThreadPoolExecutor, int]] = None
@@ -47,6 +48,7 @@ def __init__(
         self._max_wait_time = max_wait_time
         self._max_buffer_length = max_buffer_length
         self._existing_executor = False
+        self._amqp_transport = amqp_transport
 
         if not executor:
             self._executor = ThreadPoolExecutor()
@@ -88,6 +90,7 @@ def enqueue_events(
                     executor=self._executor,
                     max_wait_time=self._max_wait_time,
                     max_buffer_length=self._max_buffer_length,
+                    amqp_transport = self._amqp_transport,
                 )
                 buffered_producer.start()
                 self._buffered_producers[pid] = buffered_producer

sdk/eventhub/azure-eventhub/azure/eventhub/_client_base.py

@@ -11,7 +11,7 @@
 import collections
 from typing import Any, Dict, Tuple, List, Optional, TYPE_CHECKING, cast, Union
 try:
-    from typing import TypeAlias
+    from typing import TypeAlias    # type: ignore
 except ImportError:
     from typing_extensions import TypeAlias
 from datetime import timedelta
@@ -25,11 +25,15 @@
 from azure.core.utils import parse_connection_string as core_parse_connection_string
 from azure.core.pipeline.policies import RetryMode
 
-
-from ._transport._uamqp_transport import UamqpTransport
+try:
+    from ._transport._uamqp_transport import UamqpTransport
+except ImportError:
+    UamqpTransport = None   # type: ignore
+from ._transport._pyamqp_transport import PyamqpTransport
 from .exceptions import ClientClosedError
 from ._configuration import Configuration
-from ._utils import utc_from_timestamp, parse_sas_credential, generate_sas_token
+from ._utils import utc_from_timestamp, parse_sas_credential
+from ._pyamqp.utils import generate_sas_token
 from ._connection_manager import get_connection_manager
 from ._constants import (
     CONTAINER_PREFIX,
@@ -43,8 +47,14 @@
 
 if TYPE_CHECKING:
     from azure.core.credentials import TokenCredential
-    from uamqp import Message as uamqp_Message
-    from uamqp.authentication import JWTTokenAuth as uamqp_JWTTokenAuth
+    try:
+        from uamqp import Message as uamqp_Message
+        from uamqp.authentication import JWTTokenAuth as uamqp_JWTTokenAuth
+    except ImportError:
+        uamqp_Message = None
+        uamqp_JWTTokenAuth = None
+    from ._pyamqp.message import Message
+    from ._pyamqp.authentication import JWTTokenAuth
 
 _LOGGER = logging.getLogger(__name__)
 _Address = collections.namedtuple("_Address", "hostname path")
@@ -165,7 +175,7 @@ def _get_backoff_time(retry_mode, backoff_factor, backoff_max, retried_times):
     if retry_mode == RetryMode.Fixed:
         backoff_value = backoff_factor
     else:
-        backoff_value = backoff_factor * (2**retried_times)
+        backoff_value = backoff_factor * (2 ** retried_times)
     return min(backoff_max, backoff_value)
 
 
@@ -262,6 +272,7 @@ def get_token(self, *scopes, **kwargs):  # pylint:disable=unused-argument
         return AccessToken(signature, expiry)
 
 
+# separate TYPE_CHECKING block here for EventHubSharedKeyCredential, o/w mypy raised error even with forward referencing
 if TYPE_CHECKING:
     from azure.core.credentials import TokenCredential
 
@@ -281,8 +292,10 @@ def __init__(
         credential: CredentialTypes,
         **kwargs: Any,
     ) -> None:
-        uamqp_transport = kwargs.pop("uamqp_transport", True)
-        self._amqp_transport = kwargs.pop("amqp_transport", UamqpTransport)
+        uamqp_transport = kwargs.pop("uamqp_transport", False)
+        if uamqp_transport and not UamqpTransport:
+            raise ValueError("To use the uAMQP transport, please install `uamqp>=1.6.0,<2.0.0`.")
+        self._amqp_transport = kwargs.pop("amqp_transport", UamqpTransport if uamqp_transport else PyamqpTransport)
 
         self.eventhub_name = eventhub_name
         if not eventhub_name:
@@ -305,7 +318,10 @@ def __init__(
             **kwargs,
         )
         self._debug = self._config.network_tracing
-        self._conn_manager = get_connection_manager(**kwargs)
+        kwargs["custom_endpoint_address"] = self._config.custom_endpoint_address
+        self._conn_manager = get_connection_manager(
+            amqp_transport=self._amqp_transport,
+            **kwargs)
         self._idle_timeout = kwargs.get("idle_timeout", None)
 
     @staticmethod
@@ -322,7 +338,7 @@ def _from_connection_string(conn_str, **kwargs):
             kwargs["credential"] = EventHubSharedKeyCredential(policy, key)
         return kwargs
 
-    def _create_auth(self) -> uamqp_JWTTokenAuth:
+    def _create_auth(self) -> Union[uamqp_JWTTokenAuth, JWTTokenAuth]:
         """
         Create an ~uamqp.authentication.SASTokenAuth instance
          to authenticate the session.
@@ -381,7 +397,7 @@ def _backoff(
             raise last_exception
 
     def _management_request(
-        self, mgmt_msg: uamqp_Message, op_type: bytes
+        self, mgmt_msg: Union[uamqp_Message, Message], op_type: bytes
     ) -> Any:
         # pylint:disable=assignment-from-none
         retried_times = 0
@@ -401,26 +417,29 @@ def _management_request(
                 mgmt_msg.application_properties[
                     "security_token"
                 ] = self._amqp_transport.get_updated_token(mgmt_auth)
-                response = self._amqp_transport.mgmt_client_request(
+                status_code, description, response = self._amqp_transport.mgmt_client_request(
                     mgmt_client,
                     mgmt_msg,
                     operation=READ_OPERATION,
                     operation_type=op_type,
                     status_code_field=MGMT_STATUS_CODE,
                     description_fields=MGMT_STATUS_DESC,
                 )
-                status_code = int(response.application_properties[MGMT_STATUS_CODE])
-                description = response.application_properties.get(
-                    MGMT_STATUS_DESC
-                )  # type: Optional[Union[str, bytes]]
+                status_code = int(status_code)
                 if description and isinstance(description, bytes):
                     description = description.decode("utf-8")
                 if status_code < 400:
                     return response
                 raise self._amqp_transport.get_error(status_code, description)
             except Exception as exception:  # pylint: disable=broad-except
+                # is_consumer=True passed in here, ALTHOUGH this method is shared by the producer and consumer.
+                # is_consumer will only be checked if FileNotFoundError is raised by self.mgmt_client.open() due to
+                # invalid/non-existent connection_verify filepath. The producer will encounter the FileNotFoundError
+                # when opening the SendClient, so is_consumer=True will not be passed to amqp_transport.handle_exception
+                # there. This is for uamqp exception parity, which raises FileNotFoundError in the consumer and
+                # EventHubError in the producer. TODO: Remove `is_consumer` kwarg when resolving issue #27128.
                 last_exception = self._amqp_transport._handle_exception(  # pylint: disable=protected-access
-                    exception, self
+                    exception, self, is_consumer=True
                 )
                 self._backoff(
                     retried_times=retried_times, last_exception=last_exception
@@ -540,10 +559,10 @@ def _close_connection(self):
         self._close_handler()
         self._client._conn_manager.reset_connection_if_broken()  # pylint: disable=protected-access
 
-    def _handle_exception(self, exception):
+    def _handle_exception(self, exception, *, is_consumer=False):
         exception = self._amqp_transport.check_timeout_exception(self, exception)
         return self._amqp_transport._handle_exception(  # pylint: disable=protected-access
-            exception, self
+            exception, self, is_consumer=is_consumer
         )
 
     def _do_retryable_operation(self, operation, timeout=None, **kwargs):