Eventhubs new EventProcessor

sdk/eventhub/azure-eventhubs/HISTORY.md

@@ -1,5 +1,20 @@
 # Release History
 
+## 5.0.0b2 (2019-08-06)
+
+**New features**
+
+- Added ability to create and send EventDataBatch object with limited data size.
+- Added new configuration parameters for exponential delay among each retry operation.
+    - `retry_total`: The total number of attempts to redo the failed operation.
+    - `backoff_factor`: The delay time factor.
+    - `backoff_max`: The maximum delay time in total.
+
+**Breaking changes**
+
+- New `EventProcessor` design
+    - The `EventProcessorHost` was waived.
+
 ## 5.0.0b1 (2019-06-25)
 
 Version 5.0.0b1 is a preview of our efforts to create a client library that is user friendly and idiomatic to the Python ecosystem. The reasons for most of the changes in this update can be found in the [Azure SDK Design Guidelines for Python](https://azuresdkspecs.z5.web.core.windows.net/PythonSpec.html). For more information, please visit https://aka.ms/azure-sdk-preview1-python.

sdk/eventhub/azure-eventhubs/azure/eventhub/_consumer_producer_mixin.py

@@ -15,17 +15,16 @@
 
 def _retry_decorator(to_be_wrapped_func):
     def wrapped_func(self, *args, **kwargs):
-        timeout = kwargs.get("timeout", None)
+        timeout = kwargs.pop("timeout", None)
         if not timeout:
             timeout = 100000  # timeout None or 0 mean no timeout. 100000 seconds is equivalent to no timeout
         timeout_time = time.time() + timeout
         max_retries = self.client.config.max_retries
         retry_count = 0
         last_exception = None
-        kwargs.pop("timeout", None)
         while True:
             try:
-                return to_be_wrapped_func(timeout_time=timeout_time, last_exception=last_exception, **kwargs)
+                return to_be_wrapped_func(self, timeout_time=timeout_time, last_exception=last_exception, **kwargs)
             except Exception as exception:
                 last_exception = self._handle_exception(exception, retry_count, max_retries, timeout_time)
                 retry_count += 1
@@ -46,7 +45,7 @@ def __exit__(self, exc_type, exc_val, exc_tb):
 
     def _check_closed(self):
         if self.error:
-            raise EventHubError("{} has been closed. Please create a new consumer to receive event data.".format(self.name))
+            raise EventHubError("{} has been closed. Please create a new one to handle event data.".format(self.name))
 
     def _create_handler(self):
         pass

sdk/eventhub/azure-eventhubs/azure/eventhub/aio/_consumer_producer_mixin_async.py

@@ -15,17 +15,16 @@
 
 def _retry_decorator(to_be_wrapped_func):
     async def wrapped_func(self, *args, **kwargs):
-        timeout = kwargs.get("timeout", None)
+        timeout = kwargs.pop("timeout", None)
         if not timeout:
             timeout = 100000  # timeout None or 0 mean no timeout. 100000 seconds is equivalent to no timeout
         timeout_time = time.time() + timeout
         max_retries = self.client.config.max_retries
         retry_count = 0
         last_exception = None
-        kwargs.pop("timeout", None)
         while True:
             try:
-                return await to_be_wrapped_func(timeout_time=timeout_time, last_exception=last_exception, **kwargs)
+                return await to_be_wrapped_func(self, timeout_time=timeout_time, last_exception=last_exception, **kwargs)
             except Exception as exception:
                 last_exception = await self._handle_exception(exception, retry_count, max_retries, timeout_time)
                 retry_count += 1
@@ -47,7 +46,7 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
 
     def _check_closed(self):
         if self.error:
-            raise EventHubError("{} has been closed. Please create a new consumer to receive event data.".format(self.name))
+            raise EventHubError("{} has been closed. Please create a new one to handle event data.".format(self.name))
 
     def _create_handler(self):
         pass

sdk/eventhub/azure-eventhubs/azure/eventhub/aio/consumer_async.py

@@ -147,10 +147,8 @@ async def _open(self, timeout_time=None):
             self.source = self.redirected.address
         await super(EventHubConsumer, self)._open(timeout_time)
 
-    async def _receive(self, **kwargs):
-        timeout_time = kwargs.get("timeout_time")
+    async def _receive(self, timeout_time=None, max_batch_size=None, **kwargs):
         last_exception = kwargs.get("last_exception")
-        max_batch_size = kwargs.get("max_batch_size")
         data_batch = kwargs.get("data_batch")
 
         await self._open(timeout_time)
@@ -171,6 +169,10 @@ async def _receive(self, **kwargs):
             data_batch.append(event_data)
         return data_batch
 
+    @_retry_decorator
+    async def _receive_with_try(self, timeout_time=None, max_batch_size=None, **kwargs):
+        return await self._receive(timeout_time=timeout_time, max_batch_size=max_batch_size, **kwargs)
+
     @property
     def queue_size(self):
         # type: () -> int
@@ -217,8 +219,7 @@ async def receive(self, *, max_batch_size=None, timeout=None):
         max_batch_size = max_batch_size or min(self.client.config.max_batch_size, self.prefetch)
         data_batch = []  # type: List[EventData]
 
-        return await _retry_decorator(self._receive)(self, timeout=timeout,
-                                                     max_batch_size=max_batch_size, data_batch=data_batch)
+        return await self._receive_with_try(timeout=timeout, max_batch_size=max_batch_size, data_batch=data_batch)
 
     async def close(self, exception=None):
         # type: (Exception) -> None

sdk/eventhub/azure-eventhubs/azure/eventhub/aio/producer_async.py

@@ -110,6 +110,10 @@ async def _open(self, timeout_time=None, **kwargs):
             self.target = self.redirected.address
         await super(EventHubProducer, self)._open(timeout_time)
 
+    @_retry_decorator
+    async def _open_with_retry(self, timeout_time=None, **kwargs):
+        return await self._open(timeout_time=timeout_time, **kwargs)
+
     async def _send_event_data(self, timeout_time=None, last_exception=None):
         if self.unsent_events:
             await self._open(timeout_time)
@@ -131,6 +135,10 @@ async def _send_event_data(self, timeout_time=None, last_exception=None):
                 _error(self._outcome, self._condition)
         return
 
+    @_retry_decorator
+    async def _send_event_data_with_retry(self, timeout_time=None, last_exception=None):
+        return await self._send_event_data(timeout_time=timeout_time, last_exception=last_exception)
+
     def _on_outcome(self, outcome, condition):
         """
         Called when the outcome is received for a delivery.
@@ -155,10 +163,19 @@ async def create_batch(self, max_size=None, partition_key=None):
         :type partition_key: str
         :return: an EventDataBatch instance
         :rtype: ~azure.eventhub.EventDataBatch
+
+        Example:
+            .. literalinclude:: ../examples/async_examples/test_examples_eventhub_async.py
+                :start-after: [START eventhub_client_async_create_batch]
+                :end-before: [END eventhub_client_async_create_batch]
+                :language: python
+                :dedent: 4
+                :caption: Create EventDataBatch object within limited size
+
         """
 
         if not self._max_message_size_on_link:
-            await _retry_decorator(self._open)(self, timeout=self.client.config.send_timeout)
+            await self._open_with_retry(timeout=self.client.config.send_timeout)
 
         if max_size and max_size > self._max_message_size_on_link:
             raise ValueError('Max message size: {} is too large, acceptable max batch size is: {} bytes.'
@@ -187,7 +204,7 @@ async def send(self, event_data, *, partition_key=None, timeout=None):
         :rtype: None
 
         Example:
-            .. literalinclude:: ../examples/test_examples_eventhub.py
+            .. literalinclude:: ../examples/async_examples/test_examples_eventhub_async.py
                 :start-after: [START eventhub_client_async_send]
                 :end-before: [END eventhub_client_async_send]
                 :language: python
@@ -212,7 +229,7 @@ async def send(self, event_data, *, partition_key=None, timeout=None):
                 wrapper_event_data = EventDataBatch._from_batch(event_data, partition_key)  # pylint: disable=protected-access
         wrapper_event_data.message.on_send_complete = self._on_outcome
         self.unsent_events = [wrapper_event_data.message]
-        await _retry_decorator(self._send_event_data)(self, timeout=timeout)
+        await self._send_event_data_with_retry(timeout=timeout)
 
     async def close(self,  exception=None):
         # type: (Exception) -> None

sdk/eventhub/azure-eventhubs/azure/eventhub/consumer.py

@@ -142,10 +142,8 @@ def _open(self, timeout_time=None):
             self.source = self.redirected.address
         super(EventHubConsumer, self)._open(timeout_time)
 
-    def _receive(self, **kwargs):
-        timeout_time = kwargs.get("timeout_time")
+    def _receive(self, timeout_time=None, max_batch_size=None, **kwargs):
         last_exception = kwargs.get("last_exception")
-        max_batch_size = kwargs.get("max_batch_size")
         data_batch = kwargs.get("data_batch")
 
         self._open(timeout_time)
@@ -165,6 +163,10 @@ def _receive(self, **kwargs):
             data_batch.append(event_data)
         return data_batch
 
+    @_retry_decorator
+    def _receive_with_try(self, timeout_time=None, max_batch_size=None, **kwargs):
+        return self._receive(timeout_time=timeout_time, max_batch_size=max_batch_size, **kwargs)
+
     @property
     def queue_size(self):
         # type:() -> int
@@ -210,7 +212,7 @@ def receive(self, max_batch_size=None, timeout=None):
         max_batch_size = max_batch_size or min(self.client.config.max_batch_size, self.prefetch)
         data_batch = []  # type: List[EventData]
 
-        return _retry_decorator(self._receive)(self, timeout=timeout, max_batch_size=max_batch_size, data_batch=data_batch)
+        return self._receive_with_try(timeout=timeout, max_batch_size=max_batch_size, data_batch=data_batch)
 
     def close(self, exception=None):
         # type:(Exception) -> None

sdk/eventhub/azure-eventhubs/azure/eventhub/producer.py

@@ -117,6 +117,10 @@ def _open(self, timeout_time=None, **kwargs):
             self.target = self.redirected.address
         super(EventHubProducer, self)._open(timeout_time)
 
+    @_retry_decorator
+    def _open_with_retry(self, timeout_time=None, **kwargs):
+        return self._open(timeout_time=timeout_time, **kwargs)
+
     def _send_event_data(self, timeout_time=None, last_exception=None):
         if self.unsent_events:
             self._open(timeout_time)
@@ -138,6 +142,10 @@ def _send_event_data(self, timeout_time=None, last_exception=None):
                 _error(self._outcome, self._condition)
         return
 
+    @_retry_decorator
+    def _send_event_data_with_retry(self, timeout_time=None, last_exception=None):
+        return self._send_event_data(timeout_time=timeout_time, last_exception=last_exception)
+
     def _on_outcome(self, outcome, condition):
         """
         Called when the outcome is received for a delivery.
@@ -162,10 +170,19 @@ def create_batch(self, max_size=None, partition_key=None):
         :type partition_key: str
         :return: an EventDataBatch instance
         :rtype: ~azure.eventhub.EventDataBatch
+
+        Example:
+            .. literalinclude:: ../examples/test_examples_eventhub.py
+                :start-after: [START eventhub_client_sync_create_batch]
+                :end-before: [END eventhub_client_sync_create_batch]
+                :language: python
+                :dedent: 4
+                :caption: Create EventDataBatch object within limited size
+
         """
 
         if not self._max_message_size_on_link:
-            _retry_decorator(self._open)(self, timeout=self.client.config.send_timeout)
+            self._open_with_retry(timeout=self.client.config.send_timeout)
 
         if max_size and max_size > self._max_message_size_on_link:
             raise ValueError('Max message size: {} is too large, acceptable max batch size is: {} bytes.'
@@ -219,7 +236,7 @@ def send(self, event_data, partition_key=None, timeout=None):
                 wrapper_event_data = EventDataBatch._from_batch(event_data, partition_key)  # pylint: disable=protected-access
         wrapper_event_data.message.on_send_complete = self._on_outcome
         self.unsent_events = [wrapper_event_data.message]
-        _retry_decorator(self._send_event_data)(self, timeout=timeout)
+        self._send_event_data_with_retry(timeout=timeout)
 
     def close(self, exception=None):
         # type:(Exception) -> None

sdk/eventhub/azure-eventhubs/examples/async_examples/test_examples_eventhub_async.py

@@ -23,7 +23,7 @@ async def test_example_eventhub_async_send_and_receive(live_eventhub_config):
         os.environ['EVENT_HUB_HOSTNAME'],
         os.environ['EVENT_HUB_SAS_POLICY'],
         os.environ['EVENT_HUB_SAS_KEY'],
-        os.environ['EVENT_HUB_NAME'])    
+        os.environ['EVENT_HUB_NAME'])
     client = EventHubClient.from_connection_string(connection_str)
     # [END create_eventhub_client_async]
 
@@ -49,6 +49,17 @@ async def test_example_eventhub_async_send_and_receive(live_eventhub_config):
 
     await consumer.receive(timeout=1)
 
+    # [START eventhub_client_async_create_batch]
+    event_data_batch = await producer.create_batch(max_size=10000)
+    while True:
+        try:
+            event_data_batch.try_add(EventData('Message inside EventBatchData'))
+        except ValueError:
+            # The EventDataBatch object reaches its max_size.
+            # You can send the full EventDataBatch object and create a new one here.
+            break
+    # [END eventhub_client_async_create_batch]
+
     # [START eventhub_client_async_send]
     async with producer:
         event_data = EventData(b"A single event")

sdk/eventhub/azure-eventhubs/examples/event_data_batch.py

@@ -0,0 +1,64 @@
+#!/usr/bin/env python
+
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+"""
+An example to show creating and sending EventBatchData within limited size.
+"""
+
+# pylint: disable=C0111
+
+import logging
+import time
+import os
+
+from azure.eventhub import EventHubClient, EventData, EventHubSharedKeyCredential
+
+import examples
+logger = examples.get_logger(logging.INFO)
+
+
+HOSTNAME = os.environ.get('EVENT_HUB_HOSTNAME')  # <mynamespace>.servicebus.windows.net
+EVENT_HUB = os.environ.get('EVENT_HUB_NAME')
+
+USER = os.environ.get('EVENT_HUB_SAS_POLICY')
+KEY = os.environ.get('EVENT_HUB_SAS_KEY')
+
+
+def create_batch_data(producer):
+    event_data_batch = producer.create_batch(max_size=10000)
+    while True:
+        try:
+            event_data_batch.try_add(EventData('Message inside EventBatchData'))
+        except ValueError:
+            # EventDataBatch object reaches max_size.
+            # New EventDataBatch object can be created here to send more data
+            break
+    return event_data_batch
+
+
+try:
+    if not HOSTNAME:
+        raise ValueError("No EventHubs URL supplied.")
+
+    client = EventHubClient(host=HOSTNAME, event_hub_path=EVENT_HUB, credential=EventHubSharedKeyCredential(USER, KEY),
+                            network_tracing=False)
+    producer = client.create_producer()
+
+    try:
+        start_time = time.time()
+        with producer:
+            event_data_batch = create_batch_data(producer)
+            producer.send(event_data_batch)
+    except:
+        raise
+    finally:
+        end_time = time.time()
+        run_time = end_time - start_time
+        logger.info("Runtime: {} seconds".format(run_time))
+
+except KeyboardInterrupt:
+    pass

sdk/eventhub/azure-eventhubs/examples/test_examples_eventhub.py

@@ -86,6 +86,17 @@ def test_example_eventhub_sync_send_and_receive(live_eventhub_config):
         event_data = EventData(body=list_data)
         # [END create_event_data]
 
+        # [START eventhub_client_sync_create_batch]
+        event_data_batch = producer.create_batch(max_size=10000)
+        while True:
+            try:
+                event_data_batch.try_add(EventData('Message inside EventBatchData'))
+            except ValueError:
+                # The EventDataBatch object reaches its max_size.
+                # You can send the full EventDataBatch object and create a new one here.
+                break
+        # [END eventhub_client_sync_create_batch]
+
         # [START eventhub_client_sync_send]
         with producer:
             event_data = EventData(b"A single event")