Merge pull request #28 from annatisch/send-async

vNext

Author: annatisch

HISTORY.rst

@@ -3,6 +3,55 @@
 Release History
 ===============
 
+0.2.0 (2018-07-25)
+++++++++++++++++++
+
+- **Breaking change** `MessageSender.send_async` has been renamed to `MessageSender.send`, and
+  `MessageSenderAsync.send_async` is now a coroutine.
+- **Breaking change** Removed `detach_received` callback argument from MessageSender, MessageReceiver,
+  MessageSenderAsync, and MessageReceiverAsync in favour of new `error_policy` argument.
+- Added ErrorPolicy class to determine how the client should respond to both generic AMQP errors
+  and custom or vendor-specific errors. A default policy will be used, but a custom policy can
+  be added to any client by using a new `error_policy` argument. Value must be either an instance
+  or subclass of ErrorPolicy.
+
+    - The `error_policy` argument has also been added to MessageSender, MessageReceiver, Connection, and their
+      async counterparts to allow for handling of link DETACH and connection CLOSE events.
+    - The error policy passed to a SendClient determines the number of message send retry
+      attempts. This replaces the previous `constants.MESSAGE_SEND_RETRIES` value which is now
+      deprecated.
+    - Added new ErrorAction object to determine how a client should respond to an error. It has
+      three properties: `retry` (a boolean to determine whether the error is retryable), `backoff`
+      (an integer to determine how long the client should wait before retrying, default is 0) and
+      `increment_retries` (a boolean to determine whether the error should count against the maximum
+      retry attempts, default is `True`). Currently `backoff` and `increment_retries` are only
+      considered for message send failures.
+    - Added `VendorConnectionClose` and `VendorLinkDetach` exceptions for non-standard (unrecognized)
+      connection/link errors.
+
+- Added support for HTTP proxy configuration.
+- Added support for running async clients synchronously.
+- Added keep-alive support for connection - this is a background thread for a synchronous
+  client, and a background async function for an async client. The keep-alive feature is
+  disabled by default, to enable, set the `keep_alive_interval` argument on the client to
+  an integer representing the number of seconds between connection pings.
+- Added support for catching a Connection CLOSE event.
+- Added support for `Connection.sleep` and `ConnectionAsync.sleep_async` to pause the connection.
+- Added support for surfacing message disposition delivery-state (with error information).
+- Added `constants.ErrorCodes` enum to map standard AMQP error conditions. This replaces the previous
+  `constants.ERROR_CONNECTION_REDIRECT` and `constants.ERROR_LINK_REDIRECT` which are now both
+  deprecated.
+- Added new super error `AMQPError` from which all exceptions inherit.
+- Added new `MessageHandlerError` exception, a subclass of `AMQPConnectionError`, for
+  Senders/Receivers that enter an indeterminate error state.
+- `MessageException` is now a subclass of `MessageResponse`.
+- Added `ClientMessageError` exception, a subclass of `MessageException` for send errors raised client-side.
+- Catching Link DETACH event will now work regardless of whether service returns delivery-state.
+- Fixed bug where received messages attempting to settle on a detached link crashed the client.
+- Fixed bug in amqp C DescribedValue.
+- Fixed bug where client crashed on deallocating failed management operation.
+
+
 0.1.1 (2018-07-14)
 ++++++++++++++++++
 

build_armv7l.sh

@@ -4,7 +4,7 @@ set -e
 # To execute this script:
 # docker run --rm -v $PWD:/data local/manylinux_crypto32 /data/build_many_linux_32bit.sh
 
-export UAMQP_VERSION="0.1.1"
+export UAMQP_VERSION="0.2.0"
 
 export CPATH="/opt/uamqp/openssl/include"
 export LIBRARY_PATH="/opt/uamqp/openssl/lib"

build_many_linux_32bit.sh

@@ -4,7 +4,7 @@ set -e
 # To execute this script:
 # docker run --rm -v $PWD:/data local/manylinux_crypto32 /data/build_many_linux_32bit.sh
 
-export UAMQP_VERSION="0.1.1"
+export UAMQP_VERSION="0.2.0"
 
 export CPATH="/opt/pyca/cryptography/openssl/include"
 export LIBRARY_PATH="/opt/pyca/cryptography/openssl/lib"

build_many_linux_64bit.sh

@@ -4,7 +4,7 @@ set -e
 # To execute this script:
 # docker run --rm -v $PWD:/data local/manylinux_crypto64 /data/build_many_linux_64bit.sh
 
-export UAMQP_VERSION="0.1.1"
+export UAMQP_VERSION="0.2.0"
 
 export CPATH="/opt/pyca/cryptography/openssl/include"
 export LIBRARY_PATH="/opt/pyca/cryptography/openssl/lib"

samples/test_azure_event_hubs_receive.py

@@ -15,7 +15,7 @@
     from urllib.parse import quote_plus
 
 import uamqp
-from uamqp import address
+from uamqp import address, errors
 from uamqp import authentication
 
 
@@ -79,6 +79,23 @@ def test_event_hubs_single_batch_receive(live_eventhub_config):
     assert len(message) <= 300
 
 
+def test_event_hubs_client_proxy_settings(live_eventhub_config):
+    proxy_settings={'proxy_hostname':'127.0.0.1', 'proxy_port': 12345}
+    uri = "sb://{}/{}".format(live_eventhub_config['hostname'], live_eventhub_config['event_hub'])
+    sas_auth = authentication.SASTokenAuth.from_shared_access_key(
+        uri, live_eventhub_config['key_name'], live_eventhub_config['access_key'], http_proxy=proxy_settings)
+
+    source = "amqps://{}/{}/ConsumerGroups/{}/Partitions/{}".format(
+        live_eventhub_config['hostname'],
+        live_eventhub_config['event_hub'],
+        live_eventhub_config['consumer_group'],
+        live_eventhub_config['partition'])
+    with pytest.raises(errors.AMQPConnectionError):
+        receive_client = uamqp.ReceiveClient(source, auth=sas_auth, debug=False, timeout=50, prefetch=50)
+        receive_client.receive_message_batch(max_batch_size=10)
+    receive_client.close()
+
+
 def test_event_hubs_client_receive_sync(live_eventhub_config):
     uri = "sb://{}/{}".format(live_eventhub_config['hostname'], live_eventhub_config['event_hub'])
     sas_auth = authentication.SASTokenAuth.from_shared_access_key(

samples/test_azure_event_hubs_send.py

@@ -31,7 +31,7 @@ def test_event_hubs_simple_send(live_eventhub_config):
     sas_auth = authentication.SASTokenAuth.from_shared_access_key(
         uri, live_eventhub_config['key_name'], live_eventhub_config['access_key'])
     target = "amqps://{}/{}".format(live_eventhub_config['hostname'], live_eventhub_config['event_hub'])
-    uamqp.send_message(target, msg_content, auth=sas_auth)
+    uamqp.send_message(target, msg_content, auth=sas_auth, debug=True)
 
 
 def test_event_hubs_client_send_sync(live_eventhub_config):
@@ -104,4 +104,4 @@ def data_generator():
     config['consumer_group'] = "$Default"
     config['partition'] = "0"
 
-    test_event_hubs_batch_send_sync(config)
+    test_event_hubs_client_send_sync(config)

samples/test_azure_event_hubs_send_async.py

@@ -59,6 +59,24 @@ async def test_event_hubs_single_send_async(live_eventhub_config):
     await send_client.close_async()
 
 
+@pytest.mark.asyncio
+async def test_event_hubs_async_sender_sync(live_eventhub_config):
+    annotations={b"x-opt-partition-key": b"PartitionKeyInfo"}
+    msg_content = b"hello world"
+
+    message = uamqp.Message(msg_content, annotations=annotations)
+    uri = "sb://{}/{}".format(live_eventhub_config['hostname'], live_eventhub_config['event_hub'])
+    sas_auth = authentication.SASTokenAsync.from_shared_access_key(uri, live_eventhub_config['key_name'], live_eventhub_config['access_key'])
+
+    target = "amqps://{}/{}".format(live_eventhub_config['hostname'], live_eventhub_config['event_hub'])
+    send_client = uamqp.SendClientAsync(target, auth=sas_auth, debug=False)
+   
+    for _ in range(10):
+        message = uamqp.Message(msg_content, application_properties=annotations, annotations=annotations)
+        send_client.send_message(message)
+    send_client.close()
+
+
 @pytest.mark.asyncio
 async def test_event_hubs_batch_send_async(live_eventhub_config):
     for _ in range(10):

src/amqpvalue.pyx

@@ -875,18 +875,26 @@ cdef class DescribedValue(AMQPValue):
     @property
     def description(self):
         cdef c_amqpvalue.AMQP_VALUE value
+        cdef c_amqpvalue.AMQP_VALUE cloned
         value = c_amqpvalue.amqpvalue_get_inplace_descriptor(self._c_value)
         if <void*>value == NULL:
             self._value_error()
-        return value_factory(value)
+        cloned = c_amqpvalue.amqpvalue_clone(value)
+        if <void*>cloned == NULL:
+            self._value_error()
+        return value_factory(cloned)
 
     @property
     def data(self):
         cdef c_amqpvalue.AMQP_VALUE value
+        cdef c_amqpvalue.AMQP_VALUE cloned
         value = c_amqpvalue.amqpvalue_get_inplace_described_value(self._c_value)
         if <void*>value == NULL:
             self._value_error()
-        return value_factory(value)
+        cloned = c_amqpvalue.amqpvalue_clone(value)
+        if <void*>cloned == NULL:
+            self._value_error()
+        return value_factory(cloned)
 
     @property
     def value(self):

src/connection.pyx

@@ -190,14 +190,17 @@ cdef void on_connection_state_changed(void* context, c_connection.CONNECTION_STA
 cdef void on_io_error(void* context):
     if <void*>context != NULL:
         context_obj = <object>context
-        context_obj._io_error()
+        if hasattr(context_obj, '_io_error'):
+            context_obj._io_error()
 
 
 cdef void on_connection_close_received(void* context, c_amqp_definitions.ERROR_HANDLE error):
+    cdef c_amqp_definitions.ERROR_HANDLE cloned
     context_obj = <object>context
-    cloned = c_amqp_definitions.error_clone(error)
-    wrapped_error = error_factory(cloned)
+    if <void*> error != NULL:
+        cloned = c_amqp_definitions.error_clone(error)
+        wrapped_error = error_factory(cloned)
+    else:
+        wrapped_error = None
     if hasattr(context_obj, '_close_received'):
         context_obj._close_received(wrapped_error)
-    elif callable(context_obj):
-        context_obj(wrapped_error)

src/link.pyx

@@ -150,9 +150,10 @@ cdef class cLink(StructBase):
 cdef void on_link_detach_received(void* context, c_amqp_definitions.ERROR_HANDLE error):
     cdef c_amqp_definitions.ERROR_HANDLE cloned
     context_obj = <object>context
-    cloned = c_amqp_definitions.error_clone(error)
-    wrapped_error = error_factory(cloned)
+    if <void*> error != NULL:
+        cloned = c_amqp_definitions.error_clone(error)
+        wrapped_error = error_factory(cloned)
+    else:
+        wrapped_error = None
     if hasattr(context_obj, '_detach_received'):
         context_obj._detach_received(wrapped_error)
-    elif callable(context_obj):
-        context_obj(wrapped_error)