Added HTTP proxy support

Fix for issue Azure#41

Author: annatisch

azure/eventhub/_async/__init__.py

@@ -55,8 +55,10 @@ def _create_auth(self, username=None, password=None):  # pylint: disable=no-self
         username = username or self._auth_config['username']
         password = password or self._auth_config['password']
         if "@sas.root" in username:
-            return authentication.SASLPlain(self.address.hostname, username, password)
-        return authentication.SASTokenAsync.from_shared_access_key(self.auth_uri, username, password, timeout=60)
+            return authentication.SASLPlain(
+                self.address.hostname, username, password, http_proxy=self.http_proxy)
+        return authentication.SASTokenAsync.from_shared_access_key(
+            self.auth_uri, username, password, timeout=60, http_proxy=self.http_proxy)
 
     async def _close_clients_async(self):
         """

azure/eventhub/client.py

@@ -89,7 +89,7 @@ class EventHubClient(object):
     events to and receiving events from the Azure Event Hubs service.
     """
 
-    def __init__(self, address, username=None, password=None, debug=False):
+    def __init__(self, address, username=None, password=None, debug=False, http_proxy=None):
         """
         Constructs a new EventHubClient with the given address URL.
 
@@ -105,10 +105,15 @@ def __init__(self, address, username=None, password=None, debug=False):
         :param debug: Whether to output network trace logs to the logger. Default
          is `False`.
         :type debug: bool
+        :param http_proxy: HTTP proxy settings. This must be a dictionary with the following
+         keys: 'proxy_hostname' (str value) and 'proxy_port' (int value).
+         Additionally the following keys may also be present: 'username', 'password'.
+        :type http_proxy: dict[str, Any]
         """
         self.container_id = "eventhub.pysdk-" + str(uuid.uuid4())[:8]
         self.address = urlparse(address)
         self.eh_name = self.address.path.lstrip('/')
+        self.http_proxy = http_proxy
         self.mgmt_target = "amqps://{}/{}".format(self.address.hostname, self.eh_name)
         url_username = unquote_plus(self.address.username) if self.address.username else None
         username = username or url_username
@@ -169,8 +174,10 @@ def _create_auth(self, username=None, password=None):
         username = username or self._auth_config['username']
         password = password or self._auth_config['password']
         if "@sas.root" in username:
-            return authentication.SASLPlain(self.address.hostname, username, password)
-        return authentication.SASTokenAuth.from_shared_access_key(self.auth_uri, username, password, timeout=60)
+            return authentication.SASLPlain(
+                self.address.hostname, username, password, http_proxy=self.http_proxy)
+        return authentication.SASTokenAuth.from_shared_access_key(
+            self.auth_uri, username, password, timeout=60, http_proxy=self.http_proxy)
 
     def create_properties(self):  # pylint: disable=no-self-use
         """

azure/eventprocessorhost/eh_partition_pump.py

@@ -66,7 +66,8 @@ async def open_clients_async(self):
         # Create event hub client and receive handler and set options
         self.eh_client = EventHubClientAsync(
             self.host.eh_config.client_address,
-            debug=self.host.eph_options.debug_trace)
+            debug=self.host.eph_options.debug_trace,
+            http_proxy=self.host.eph_options.http_proxy)
         self.partition_receive_handler = self.eh_client.add_async_receiver(
             self.partition_context.consumer_group_name,
             self.partition_context.partition_id,

azure/eventprocessorhost/eph.py

@@ -73,3 +73,4 @@ def __init__(self):
         self.release_pump_on_timeout = False
         self.initial_offset_provider = "-1"
         self.debug_trace = False
+        self.http_proxy = None

azure/eventprocessorhost/partition_manager.py

@@ -38,7 +38,8 @@ async def get_partition_ids_async(self):
             try:
                 eh_client = EventHubClientAsync(
                     self.host.eh_config.client_address,
-                    debug=self.host.eph_options.debug_trace)
+                    debug=self.host.eph_options.debug_trace,
+                    http_proxy=self.host.eph_options.http_proxy)
                 try:
                     eh_info = await eh_client.get_eventhub_info_async()
                     self.partition_ids = eh_info['partition_ids']