Merge pull request #137 from meejah/cancellation

Cancellation

docs/api.rst

@@ -71,19 +71,23 @@ txaio module
     Select the Twisted framework (will fail if Twisted is not installed).
 
 
-.. py:function:: create_future(value=None, error=None)
+.. py:function:: create_future(result=None, error=None, canceller=None)
 
     Create and return a new framework-specific future object. On
     asyncio this returns a `Future`_, on Twisted it returns a
     `Deferred`_.
 
-    :param value: if not ``None``, the future is already fulfilled,
+    :param result: if not ``None``, the future is already fulfilled,
         with the given result.
 
     :param error: if not ``None`` then the future is already failed,
         with the given error.
     :type error: class:`IFailedFuture` or Exception
 
+    :param canceller: a single-argument callable which is invoked if
+        this future is cancelled (the single argument is the future
+        object which has been cancelled)
+
     :raises ValueError: if both ``value`` and ``error`` are provided.
 
     :return: under Twisted a `Deferred`_, under asyncio a `Future`_
@@ -129,6 +133,16 @@ txaio module
     :type error: :class:`IFailedFuture` or :class:`Exception`
 
 
+.. py:function:: cancel(future)
+
+    Cancel the given future. If a ``canceller`` was registered, it is
+    invoked now. It is invalid to ``resolve`` or ``reject`` the future
+    after cancelling it.
+
+    :param future: an unresolved `Deferred`_/`Future`_ as returned by
+                    :meth:`create_future`
+
+
 .. py:function:: resolve(future, value)
 
     Resolve the given future with the provided value. This triggers

docs/releases.rst

@@ -4,6 +4,9 @@ txio releases
 master
 ------
 
+- add API to support cancellation; this means passing a 1-argument
+  callable to ``create_future`` and ``txaio.cancel`` to actually
+  cancel a future
 - ...
 
 

test/test_cancel.py

@@ -0,0 +1,52 @@
+###############################################################################
+#
+# The MIT License (MIT)
+#
+# Copyright (c) Crossbar.io Technologies GmbH
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+# THE SOFTWARE.
+#
+###############################################################################
+
+import txaio
+
+from util import run_once
+
+
+def test_cancel(framework):
+    cancels = []
+
+    def it_died(f):
+        cancels.append(f)
+
+    f = txaio.create_future(canceller=it_died)
+    # both Future and Deferred have .cancel() methods .. but seemed
+    # more "symmetric"/expected to make a method? But could just stick
+    # with "f.cancel()" here ...
+    txaio.cancel(f)
+
+    # at least for Twisted, we have to "handle" the "CancelledError"
+    # -- in practice, dropping a future on the floor with no
+    # error-handler is A Bad Thing anyway
+    txaio.add_callbacks(f, None, lambda _: None)
+
+    run_once()
+    run_once()
+
+    assert cancels == [f]

txaio/__init__.py

@@ -74,6 +74,7 @@ class _Config(object):
     'is_future',                # True for Deferreds in tx and Futures, @coroutines in asyncio
     'reject',                   # errback a Future
     'resolve',                  # callback a Future
+    'cancel',                   # cancel a Future
     'add_callbacks',            # add callback and/or errback
     'gather',                   # return a Future waiting for several other Futures
     'is_called',                # True if the Future has a result

txaio/_unframework.py

@@ -54,6 +54,7 @@ def _throw_usage_error(*args, **kw):
 as_future = _throw_usage_error
 is_future = _throw_usage_error
 reject = _throw_usage_error
+cancel = _throw_usage_error
 resolve = _throw_usage_error
 add_callbacks = _throw_usage_error
 gather = _throw_usage_error

txaio/aio.py

@@ -389,7 +389,7 @@ def failure_format_traceback(self, fail):
         except Exception:
             return u"Failed to format failure traceback for '{0}'".format(fail)
 
-    def create_future(self, result=_unspecified, error=_unspecified):
+    def create_future(self, result=_unspecified, error=_unspecified, canceller=_unspecified):
         if result is not _unspecified and error is not _unspecified:
             raise ValueError("Cannot have both result and error.")
 
@@ -398,6 +398,19 @@ def create_future(self, result=_unspecified, error=_unspecified):
             resolve(f, result)
         elif error is not _unspecified:
             reject(f, error)
+
+        # Twisted's only API for cancelling is to pass a
+        # single-argument callable to the Deferred constructor, so
+        # txaio apes that here for asyncio. The argument is the Future
+        # that has been cancelled.
+        if canceller is not _unspecified:
+            def done(f):
+                try:
+                    f.exception()
+                except asyncio.CancelledError:
+                    canceller(f)
+            f.add_done_callback(done)
+
         return f
 
     def create_future_success(self, result):
@@ -476,6 +489,9 @@ def reject(self, future, error=None):
                 raise RuntimeError("reject requires an IFailedFuture or Exception")
         future.set_exception(error.value)
 
+    def cancel(self, future):
+        future.cancel()
+
     def create_failure(self, exception=None):
         """
         This returns an object implementing IFailedFuture.
@@ -550,6 +566,7 @@ def sleep(self, delay):
 is_called = _default_api.is_called
 resolve = _default_api.resolve
 reject = _default_api.reject
+cancel = _default_api.cancel
 create_failure = _default_api.create_failure
 add_callbacks = _default_api.add_callbacks
 gather = _default_api.gather

txaio/tx.py

@@ -405,11 +405,11 @@ def failure_format_traceback(self, fail):
         except Exception:
             return u"Failed to format failure traceback for '{0}'".format(fail)
 
-    def create_future(self, result=_unspecified, error=_unspecified):
+    def create_future(self, result=_unspecified, error=_unspecified, canceller=None):
         if result is not _unspecified and error is not _unspecified:
             raise ValueError("Cannot have both result and error.")
 
-        f = Deferred()
+        f = Deferred(canceller=canceller)
         if result is not _unspecified:
             resolve(f, result)
         elif error is not _unspecified:
@@ -477,6 +477,9 @@ def reject(self, future, error=None):
                 raise RuntimeError("reject requires a Failure or Exception")
         future.errback(error)
 
+    def cancel(self, future):
+        future.cancel()
+
     def create_failure(self, exception=None):
         """
         Create a Failure instance.
@@ -579,6 +582,7 @@ def get_global_log_level():
 is_called = _default_api.is_called
 resolve = _default_api.resolve
 reject = _default_api.reject
+cancel = _default_api.cancel
 create_failure = _default_api.create_failure
 add_callbacks = _default_api.add_callbacks
 gather = _default_api.gather