Merge pull request #51 from annatisch/eh_scenarios

EPH improvements

Author: annatisch

.gitignore

@@ -110,4 +110,5 @@ ENV/
 azure/mgmt/
 azure/common/
 azure/profiles/
+azure/servicebus/
 features/steps/mgmt_settings_real.py

HISTORY.rst

@@ -3,6 +3,22 @@
 Release History
 ===============
 
+0.2.0 (2018-08-06)
+++++++++++++++++++
+
+- Stability improvements for EPH.
+- Updated uAMQP version.
+- Added new configuration options for Sender and Receiver; `keep_alive` and `auto_reconnect`.
+  These flags have been added to the following:
+
+  - `EventHubClient.add_receiver`
+  - `EventHubClient.add_sender`
+  - `EventHubClientAsync.add_async_receiver`
+  - `EventHubClientAsync.add_async_sender`
+  - `EPHOptions.keey_alive_interval`
+  - `EPHOptions.auto_reconnect_on_error`
+
+
 0.2.0rc2 (2018-07-29)
 +++++++++++++++++++++
 

azure/eventhub/__init__.py

@@ -3,7 +3,7 @@
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
 
-__version__ = "0.2.0rc2"
+__version__ = "0.2.0"
 
 from azure.eventhub.common import EventData, EventHubError, Offset
 from azure.eventhub.client import EventHubClient

azure/eventhub/_async/__init__.py

@@ -77,6 +77,7 @@ async def _start_client_async(self, client):
         try:
             await client.open_async()
         except Exception as exp:  # pylint: disable=broad-except
+            log.info("Encountered error while starting handler: {}".format(exp))
             await client.close_async(exception=exp)
 
     async def _handle_redirect(self, redirects):
@@ -164,7 +165,9 @@ async def get_eventhub_info_async(self):
         finally:
             await mgmt_client.close_async()
 
-    def add_async_receiver(self, consumer_group, partition, offset=None, prefetch=300, operation=None, loop=None):
+    def add_async_receiver(
+            self, consumer_group, partition, offset=None, prefetch=300,
+            operation=None, keep_alive=30, auto_reconnect=True, loop=None):
         """
         Add an async receiver to the client for a particular consumer group and partition.
 
@@ -184,11 +187,15 @@ def add_async_receiver(self, consumer_group, partition, offset=None, prefetch=30
         path = self.address.path + operation if operation else self.address.path
         source_url = "amqps://{}{}/ConsumerGroups/{}/Partitions/{}".format(
             self.address.hostname, path, consumer_group, partition)
-        handler = AsyncReceiver(self, source_url, offset=offset, prefetch=prefetch, loop=loop)
+        handler = AsyncReceiver(
+            self, source_url, offset=offset, prefetch=prefetch,
+            keep_alive=keep_alive, auto_reconnect=auto_reconnect, loop=loop)
         self.clients.append(handler)
         return handler
 
-    def add_async_epoch_receiver(self, consumer_group, partition, epoch, prefetch=300, operation=None, loop=None):
+    def add_async_epoch_receiver(
+            self, consumer_group, partition, epoch, prefetch=300,
+            operation=None, keep_alive=30, auto_reconnect=True, loop=None):
         """
         Add an async receiver to the client with an epoch value. Only a single epoch receiver
         can connect to a partition at any given time - additional epoch receivers must have
@@ -211,11 +218,13 @@ def add_async_epoch_receiver(self, consumer_group, partition, epoch, prefetch=30
         path = self.address.path + operation if operation else self.address.path
         source_url = "amqps://{}{}/ConsumerGroups/{}/Partitions/{}".format(
             self.address.hostname, path, consumer_group, partition)
-        handler = AsyncReceiver(self, source_url, prefetch=prefetch, epoch=epoch, loop=loop)
+        handler = AsyncReceiver(
+            self, source_url, prefetch=prefetch, epoch=epoch,
+            keep_alive=keep_alive, auto_reconnect=auto_reconnect, loop=loop)
         self.clients.append(handler)
         return handler
 
-    def add_async_sender(self, partition=None, operation=None, loop=None):
+    def add_async_sender(self, partition=None, operation=None, keep_alive=30, auto_reconnect=True, loop=None):
         """
         Add an async sender to the client to send ~azure.eventhub.common.EventData object
         to an EventHub.
@@ -232,6 +241,8 @@ def add_async_sender(self, partition=None, operation=None, loop=None):
         target = "amqps://{}{}".format(self.address.hostname, self.address.path)
         if operation:
             target = target + operation
-        handler = AsyncSender(self, target, partition=partition, loop=loop)
+        handler = AsyncSender(
+            self, target, partition=partition, keep_alive=keep_alive,
+            auto_reconnect=auto_reconnect, loop=loop)
         self.clients.append(handler)
         return handler

azure/eventhub/_async/receiver_async.py

@@ -4,6 +4,8 @@
 # --------------------------------------------------------------------------------------------
 
 import asyncio
+import uuid
+import logging
 
 from uamqp import errors, types
 from uamqp import ReceiveClientAsync, Source
@@ -12,13 +14,17 @@
 from azure.eventhub.receiver import Receiver
 from azure.eventhub.common import _error_handler
 
+log = logging.getLogger(__name__)
+
 
 class AsyncReceiver(Receiver):
     """
     Implements the async API of a Receiver.
     """
 
-    def __init__(self, client, source, offset=None, prefetch=300, epoch=None, loop=None):  # pylint: disable=super-init-not-called
+    def __init__(  # pylint: disable=super-init-not-called
+            self, client, source, offset=None, prefetch=300, epoch=None,
+            keep_alive=None, auto_reconnect=True, loop=None):
         """
         Instantiate an async receiver.
 
@@ -39,10 +45,14 @@ def __init__(self, client, source, offset=None, prefetch=300, epoch=None, loop=N
         self.offset = offset
         self.prefetch = prefetch
         self.epoch = epoch
+        self.keep_alive = keep_alive
+        self.auto_reconnect = auto_reconnect
         self.retry_policy = errors.ErrorPolicy(max_retries=3, on_error=_error_handler)
         self.redirected = None
         self.error = None
         self.properties = None
+        partition = self.source.split('/')[-1]
+        self.name = "EHReceiver-{}-partition{}".format(uuid.uuid4(), partition)
         source = Source(self.source)
         if self.offset is not None:
             source.set_filter(self.offset.selector())
@@ -56,7 +66,9 @@ def __init__(self, client, source, offset=None, prefetch=300, epoch=None, loop=N
             link_properties=self.properties,
             timeout=self.timeout,
             error_policy=self.retry_policy,
-            keep_alive_interval=30,
+            keep_alive_interval=self.keep_alive,
+            client_name=self.name,
+            properties=self.client.create_properties(),
             loop=self.loop)
 
     async def open_async(self):
@@ -85,7 +97,9 @@ async def open_async(self):
                 link_properties=self.properties,
                 timeout=self.timeout,
                 error_policy=self.retry_policy,
-                keep_alive_interval=30,
+                keep_alive_interval=self.keep_alive,
+                client_name=self.name,
+                properties=self.client.create_properties(),
                 loop=self.loop)
         await self._handler.open_async()
         while not await self.has_started():
@@ -110,7 +124,8 @@ async def reconnect_async(self):
             link_properties=self.properties,
             timeout=self.timeout,
             error_policy=self.retry_policy,
-            keep_alive_interval=30,
+            keep_alive_interval=self.keep_alive,
+            client_name=self.name,
             properties=self.client.create_properties(),
             loop=self.loop)
         await self._handler.open_async()
@@ -189,17 +204,27 @@ async def receive(self, max_batch_size=None, timeout=None):
                 data_batch.append(event_data)
             return data_batch
         except (errors.LinkDetach, errors.ConnectionClose) as shutdown:
-            if shutdown.action.retry:
+            if shutdown.action.retry and self.auto_reconnect:
+                log.info("AsyncReceiver detached. Attempting reconnect.")
                 await self.reconnect_async()
                 return data_batch
             else:
+                log.info("AsyncReceiver detached. Shutting down.")
+                error = EventHubError(str(shutdown), shutdown)
+                await self.close_async(exception=error)
+                raise error
+        except errors.MessageHandlerError as shutdown:
+            if self.auto_reconnect:
+                log.info("AsyncReceiver detached. Attempting reconnect.")
+                await self.reconnect_async()
+                return data_batch
+            else:
+                log.info("AsyncReceiver detached. Shutting down.")
                 error = EventHubError(str(shutdown), shutdown)
                 await self.close_async(exception=error)
                 raise error
-        except errors.MessageHandlerError:
-            await self.reconnect_async()
-            return data_batch
         except Exception as e:
+            log.info("Unexpected error occurred ({}). Shutting down.".format(e))
             error = EventHubError("Receive failed: {}".format(e))
             await self.close_async(exception=error)
             raise error

azure/eventhub/_async/sender_async.py

@@ -3,6 +3,7 @@
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
 
+import uuid
 import asyncio
 
 from uamqp import constants, errors
@@ -17,7 +18,7 @@ class AsyncSender(Sender):
     Implements the async API of a Sender.
     """
 
-    def __init__(self, client, target, partition=None, loop=None):  # pylint: disable=super-init-not-called
+    def __init__(self, client, target, partition=None, keep_alive=None, auto_reconnect=True, loop=None):  # pylint: disable=super-init-not-called
         """
         Instantiate an EventHub event SenderAsync handler.
 
@@ -31,18 +32,23 @@ def __init__(self, client, target, partition=None, loop=None):  # pylint: disabl
         self.client = client
         self.target = target
         self.partition = partition
+        self.keep_alive = keep_alive
+        self.auto_reconnect = auto_reconnect
         self.retry_policy = errors.ErrorPolicy(max_retries=3, on_error=_error_handler)
+        self.name = "EHSender-{}".format(uuid.uuid4())
         self.redirected = None
         self.error = None
         if partition:
             self.target += "/Partitions/" + partition
+            self.name += "-partition{}".format(partition)
         self._handler = SendClientAsync(
             self.target,
             auth=self.client.get_auth(),
             debug=self.client.debug,
             msg_timeout=Sender.TIMEOUT,
             error_policy=self.retry_policy,
-            keep_alive_interval=30,
+            keep_alive_interval=self.keep_alive,
+            client_name=self.name,
             properties=self.client.create_properties(),
             loop=self.loop)
         self._outcome = None
@@ -65,7 +71,8 @@ async def open_async(self):
                 debug=self.client.debug,
                 msg_timeout=Sender.TIMEOUT,
                 error_policy=self.retry_policy,
-                keep_alive_interval=30,
+                keep_alive_interval=self.keep_alive,
+                client_name=self.name,
                 properties=self.client.create_properties(),
                 loop=self.loop)
         await self._handler.open_async()
@@ -85,7 +92,8 @@ async def reconnect_async(self):
             debug=self.client.debug,
             msg_timeout=Sender.TIMEOUT,
             error_policy=self.retry_policy,
-            keep_alive_interval=30,
+            keep_alive_interval=self.keep_alive,
+            client_name=self.name,
             properties=self.client.create_properties(),
             loop=self.loop)
         await self._handler.open_async()
@@ -158,14 +166,19 @@ async def send(self, event_data):
             if self._outcome != constants.MessageSendResult.Ok:
                 raise Sender._error(self._outcome, self._condition)
         except (errors.LinkDetach, errors.ConnectionClose) as shutdown:
-            if shutdown.action.retry:
+            if shutdown.action.retry and self.auto_reconnect:
+                await self.reconnect_async()
+            else:
+                error = EventHubError(str(shutdown), shutdown)
+                await self.close_async(exception=error)
+                raise error
+        except errors.MessageHandlerError as shutdown:
+            if self.auto_reconnect:
                 await self.reconnect_async()
             else:
                 error = EventHubError(str(shutdown), shutdown)
                 await self.close_async(exception=error)
                 raise error
-        except errors.MessageHandlerError:
-            await self.reconnect_async()
         except Exception as e:
             error = EventHubError("Send failed: {}".format(e))
             await self.close_async(exception=error)
@@ -182,13 +195,18 @@ async def wait_async(self):
         try:
             await self._handler.wait_async()
         except (errors.LinkDetach, errors.ConnectionClose) as shutdown:
-            if shutdown.action.retry:
+            if shutdown.action.retry and self.auto_reconnect:
+                await self.reconnect_async()
+            else:
+                error = EventHubError(str(shutdown), shutdown)
+                await self.close_async(exception=error)
+                raise error
+        except errors.MessageHandlerError as shutdown:
+            if self.auto_reconnect:
                 await self.reconnect_async()
             else:
                 error = EventHubError(str(shutdown), shutdown)
                 await self.close_async(exception=error)
                 raise error
-        except errors.MessageHandlerError:
-            await self.reconnect_async()
         except Exception as e:
             raise EventHubError("Send failed: {}".format(e))

azure/eventhub/client.py

@@ -303,7 +303,9 @@ def get_eventhub_info(self):
         finally:
             mgmt_client.close()
 
-    def add_receiver(self, consumer_group, partition, offset=None, prefetch=300, operation=None):
+    def add_receiver(
+            self, consumer_group, partition, offset=None, prefetch=300,
+            operation=None, keep_alive=30, auto_reconnect=True):
         """
         Add a receiver to the client for a particular consumer group and partition.
 
@@ -323,11 +325,15 @@ def add_receiver(self, consumer_group, partition, offset=None, prefetch=300, ope
         path = self.address.path + operation if operation else self.address.path
         source_url = "amqps://{}{}/ConsumerGroups/{}/Partitions/{}".format(
             self.address.hostname, path, consumer_group, partition)
-        handler = Receiver(self, source_url, offset=offset, prefetch=prefetch)
+        handler = Receiver(
+            self, source_url, offset=offset, prefetch=prefetch,
+            keep_alive=keep_alive, auto_reconnect=auto_reconnect)
         self.clients.append(handler)
         return handler
 
-    def add_epoch_receiver(self, consumer_group, partition, epoch, prefetch=300, operation=None):
+    def add_epoch_receiver(
+            self, consumer_group, partition, epoch, prefetch=300,
+            operation=None, keep_alive=30, auto_reconnect=True):
         """
         Add a receiver to the client with an epoch value. Only a single epoch receiver
         can connect to a partition at any given time - additional epoch receivers must have
@@ -350,11 +356,13 @@ def add_epoch_receiver(self, consumer_group, partition, epoch, prefetch=300, ope
         path = self.address.path + operation if operation else self.address.path
         source_url = "amqps://{}{}/ConsumerGroups/{}/Partitions/{}".format(
             self.address.hostname, path, consumer_group, partition)
-        handler = Receiver(self, source_url, prefetch=prefetch, epoch=epoch)
+        handler = Receiver(
+            self, source_url, prefetch=prefetch, epoch=epoch,
+            keep_alive=keep_alive, auto_reconnect=auto_reconnect)
         self.clients.append(handler)
         return handler
 
-    def add_sender(self, partition=None, operation=None):
+    def add_sender(self, partition=None, operation=None, keep_alive=30, auto_reconnect=True):
         """
         Add a sender to the client to send ~azure.eventhub.common.EventData object
         to an EventHub.
@@ -371,6 +379,6 @@ def add_sender(self, partition=None, operation=None):
         target = "amqps://{}{}".format(self.address.hostname, self.address.path)
         if operation:
             target = target + operation
-        handler = Sender(self, target, partition=partition)
+        handler = Sender(self, target, partition=partition, keep_alive=keep_alive, auto_reconnect=auto_reconnect)
         self.clients.append(handler)
         return handler