source: ralphm-patches/async-observer.patch

wokkel/subprotocols.py

@@ -10 +10 @@
 __all__ = ['XMPPHandler', 'XMPPHandlerCollection', 'StreamManager',
            'IQHandlerMixin']
 
+from functools import wraps
+
 from zope.interface import implements
 
 from twisted.internet import defer
@@ -379 +381 @@
 
 
 
+def asyncObserver(observer):
+    """
+    Decorator for asynchronous stanza observers.
+
+    This decorator makes it easier to do proper error handling for stanza
+    observers and supports deferred results:
+
+        >>> class MyHandler(XMPPHandler):
+        ...    def connectionInitialized(self):
+        ...       self.xmlstream.addObserver('/message', self.onMessage)
+        ...    @asyncObserver
+        ...    def onMessage(self, element):
+        ...        return False
+        ...    @asyncObserver
+        ...    def onMessage(self, element):
+        ...        return True
+        ...    @asyncObserver
+        ...    def onMessage(self, element):
+        ...        raise NotImplementedError
+        ...    @asyncObserver
+        ...    def onMessage(self, element):
+        ...        return defer.fail(StanzaError('bad-request'))
+
+
+    If the stanza had its C{handled} attribute set to C{True}, it will be
+    ignored and the observer will not be called.
+
+    The return value of the wrapped observer is used to set the C{handled}
+    attribute, so that handlers may choose to ignore processing the same
+    stanza.
+
+    If an exception is raised, or the deferred has its errback called, the
+    exception is checked for being a L{error.StanzaError}. If so, an error
+    response is sent. A L{NotImplementedError} will cause an error response
+    with the condition C{service-unavailable}. Any other exception will cause a
+    error response of C{internal-server-error} to be sent.
+
+    The return value of the deferred is not used.
+    """
+    @wraps(observer)
+    def observe(self, element):
+        def checkStanzaType(failure):
+            if element.getAttribute('type') in ('result', 'error'):
+                log.err(failure, 'Cannot return error in response to a '
+                                 'response stanza.')
+            else:
+                return failure
+
+        def trapNotImplemented(failure):
+            failure.trap(NotImplementedError)
+            raise error.StanzaError('service-unavailable')
+
+        def trapOtherError(failure):
+            if failure.check(error.StanzaError):
+                return failure
+            else:
+                log.err(failure, "Unhandled error in observer")
+                raise error.StanzaError('internal-server-error')
+
+        def trapStanzaError(failure):
+            failure.trap(error.StanzaError)
+            self.send(failure.value.toResponse(element))
+
+        if element.handled:
+            return
+
+        try:
+            result = observer(self, element)
+        except Exception:
+            result = defer.fail()
+
+        if isinstance(result, defer.Deferred):
+            result.addErrback(checkStanzaType)
+            result.addErrback(trapNotImplemented)
+            result.addErrback(trapOtherError)
+            result.addErrback(trapStanzaError)
+            result.addErrback(log.err)
+
+        element.handled = result
+    return observe
+
+
+
 class IQHandlerMixin(object):
     """
     XMPP subprotocol mixin for handle incoming IQ stanzas.
@@ -401 +486 @@
 
     A typical way to use this mixin, is to set up L{xpath} observers on the
     C{xmlstream} to call handleRequest, for example in an overridden
-    L{XMPPHandler.connectionMade}. It is likely a good idea to only listen for
-    incoming iq get and/org iq set requests, and not for any iq, to prevent
-    hijacking incoming responses to outgoing iq requests. An example:
+    L{XMPPHandler.connectionInitialized}. It is likely a good idea to only
+    listen for incoming iq get and/org iq set requests, and not for any iq, to
+    prevent hijacking incoming responses to outgoing iq requests. An example:
 
         >>> QUERY_ROSTER = "/query[@xmlns='jabber:iq:roster']"
         >>> class MyHandler(XMPPHandler, IQHandlerMixin):
         ...    iqHandlers = {"/iq[@type='get']" + QUERY_ROSTER: 'onRosterGet',
         ...                  "/iq[@type='set']" + QUERY_ROSTER: 'onRosterSet'}
-        ...    def connectionMade(self):
+        ...    def connectionInitialized(self):
         ...        self.xmlstream.addObserver(
         ...          "/iq[@type='get' or @type='set']" + QUERY_ROSTER,
         ...          self.handleRequest)
@@ -425 +510 @@
 
     iqHandlers = None
 
+    @asyncObserver
     def handleRequest(self, iq):
         """
         Find a handler and wrap the call for sending a response stanza.
@@ -435 +521 @@
             if result:
                 if IElement.providedBy(result):
                     response.addChild(result)
-                else:
-                    for element in result:
-                        response.addChild(element)
 
             return response
 
-        def checkNotImplemented(failure):
+        def trapNotImplemented(failure):
             failure.trap(NotImplementedError)
             raise error.StanzaError('feature-not-implemented')
 
-        def fromStanzaError(failure, iq):
-            failure.trap(error.StanzaError)
-            return failure.value.toResponse(iq)
-
-        def fromOtherError(failure, iq):
-            log.msg("Unhandled error in iq handler:", isError=True)
-            log.err(failure)
-            return error.StanzaError('internal-server-error').toResponse(iq)
-
         handler = None
         for queryString, method in self.iqHandlers.iteritems():
             if xpath.internQuery(queryString).matches(iq):
@@ -461 +535 @@
 
         if handler:
             d = defer.maybeDeferred(handler, iq)
+            d.addCallback(toResult, iq)
+            d.addCallback(self.send)
         else:
             d = defer.fail(NotImplementedError())
 
-        d.addCallback(toResult, iq)
-        d.addErrback(checkNotImplemented)
-        d.addErrback(fromStanzaError, iq)
-        d.addErrback(fromOtherError, iq)
-
-        d.addCallback(self.send)
-
-        iq.handled = True
+        d.addErrback(trapNotImplemented)
+        return d

wokkel/test/test_subprotocols.py

@@ -790 +790 @@
         self.xmlstream.send(obj)
 
 
+
+class AsyncObserverTest(unittest.TestCase):
+    """
+    Tests for L{wokkel.subprotocols.asyncObserver}.
+    """
+
+    def setUp(self):
+        self.output = []
+
+
+    def send(self, element):
+        self.output.append(element)
+
+
+    def test_handled(self):
+        """
+        If the element is marked as handled, it is ignored.
+        """
+        called = []
+
+        @subprotocols.asyncObserver
+        def observer(self, element):
+            called.append(element)
+
+        element = domish.Element((None, u'message'))
+        element.handled = True
+
+        observer(self, element)
+        self.assertFalse(called)
+        self.assertTrue(element.handled)
+        self.assertFalse(self.output)
+
+
+    def test_syncFalse(self):
+        """
+        If the observer returns False, the element is not marked as handled.
+        """
+        @subprotocols.asyncObserver
+        def observer(self, element):
+            return False
+
+        element = domish.Element((None, u'message'))
+
+        observer(self, element)
+        self.assertFalse(element.handled)
+        self.assertFalse(self.output)
+
+
+    def test_syncTrue(self):
+        """
+        If the observer returns True, the element is marked as handled.
+        """
+        @subprotocols.asyncObserver
+        def observer(self, element):
+            return True
+
+        element = domish.Element((None, u'message'))
+
+        observer(self, element)
+        self.assertTrue(element.handled)
+        self.assertFalse(self.output)
+
+
+    def test_syncNotImplemented(self):
+        """
+        NotImplementedError causes a service-unavailable response.
+        """
+        @subprotocols.asyncObserver
+        def observer(self, element):
+            raise NotImplementedError()
+
+        element = domish.Element((None, u'message'))
+
+        observer(self, element)
+        self.assertTrue(element.handled)
+        self.assertEquals(1, len(self.output))
+        exc = error.exceptionFromStanza(self.output[-1])
+        self.assertEquals(u'service-unavailable', exc.condition)
+
+
+    def test_syncStanzaError(self):
+        """
+        A StanzaError is sent back as is.
+        """
+        @subprotocols.asyncObserver
+        def observer(self, element):
+            raise error.StanzaError(u'forbidden')
+
+        element = domish.Element((None, u'message'))
+
+        observer(self, element)
+        self.assertTrue(element.handled)
+        self.assertEquals(1, len(self.output))
+        exc = error.exceptionFromStanza(self.output[-1])
+        self.assertEquals(u'forbidden', exc.condition)
+
+
+    def test_syncOtherError(self):
+        """
+        Other exceptions are logged and cause an internal-service response.
+        """
+        class Error(Exception):
+            pass
+
+        @subprotocols.asyncObserver
+        def observer(self, element):
+            raise Error(u"oops")
+
+        element = domish.Element((None, u'message'))
+
+        observer(self, element)
+        self.assertTrue(element.handled)
+        self.assertEquals(1, len(self.output))
+        exc = error.exceptionFromStanza(self.output[-1])
+        self.assertEquals(u'internal-server-error', exc.condition)
+        self.assertEquals(1, len(self.flushLoggedErrors()))
+
+
+    def test_asyncError(self):
+        """
+        Other exceptions are logged and cause an internal-service response.
+        """
+        class Error(Exception):
+            pass
+
+        @subprotocols.asyncObserver
+        def observer(self, element):
+            return defer.fail(Error("oops"))
+
+        element = domish.Element((None, u'message'))
+
+        observer(self, element)
+        self.assertTrue(element.handled)
+        self.assertEquals(1, len(self.output))
+        exc = error.exceptionFromStanza(self.output[-1])
+        self.assertEquals(u'internal-server-error', exc.condition)
+        self.assertEquals(1, len(self.flushLoggedErrors()))
+
+
+    def test_errorMessage(self):
+        """
+        If the element is an error message, observer exceptions are just logged.
+        """
+        class Error(Exception):
+            pass
+
+        @subprotocols.asyncObserver
+        def observer(self, element):
+            raise Error("oops")
+
+        element = domish.Element((None, u'message'))
+        element[u'type'] = u'error'
+
+        observer(self, element)
+        self.assertTrue(element.handled)
+        self.assertEquals(0, len(self.output))
+        self.assertEquals(1, len(self.flushLoggedErrors()))
+
+
+
 class IQHandlerTest(unittest.TestCase):
+    """
+    Tests for L{subprotocols.IQHandler}.
+    """
 
     def test_match(self):
         """
-        Test that the matching handler gets called.
+        The matching handler gets called.
         """
-
         class Handler(DummyIQHandler):
             called = False
 
@@ -810 +972 @@
         handler.handleRequest(iq)
         self.assertTrue(handler.called)
 
+
     def test_noMatch(self):
         """
-        Test that the matching handler gets called.
+        If the element does not match the handler is not called.
         """
-
         class Handler(DummyIQHandler):
             called = False
 
@@ -828 +990 @@
         handler.handleRequest(iq)
         self.assertFalse(handler.called)
 
+
     def test_success(self):
         """
-        Test response when the request is handled successfully.
+        If None is returned, an empty result iq is returned.
         """
-
         class Handler(DummyIQHandler):
             def onGet(self, iq):
                 return None
@@ -847 +1009 @@
         self.assertEquals('iq', response.name)
         self.assertEquals('result', response['type'])
 
+
     def test_successPayload(self):
         """
-        Test response when the request is handled successfully with payload.
+        If an Element is returned it is added as the payload of the result iq.
         """
-
         class Handler(DummyIQHandler):
             payload = domish.Element(('testns', 'foo'))
 
@@ -870 +1032 @@
         payload = response.elements().next()
         self.assertEqual(handler.payload, payload)
 
+
     def test_successDeferred(self):
         """
-        Test response when where the handler was a deferred.
+        A deferred result is used when fired.
         """
-
         class Handler(DummyIQHandler):
             def onGet(self, iq):
                 return defer.succeed(None)
@@ -889 +1051 @@
         self.assertEquals('iq', response.name)
         self.assertEquals('result', response['type'])
 
+
     def test_failure(self):
         """
-        Test response when the request is handled unsuccessfully.
+        A raised StanzaError causes an error response.
         """
-
         class Handler(DummyIQHandler):
             def onGet(self, iq):
                 raise error.StanzaError('forbidden')
@@ -910 +1072 @@
         e = error.exceptionFromStanza(response)
         self.assertEquals('forbidden', e.condition)
 
+
     def test_failureUnknown(self):
         """
-        Test response when the request handler raises a non-stanza-error.
+        Any Exception cause an internal-server-error response.
         """
-
         class TestError(Exception):
             pass
 
@@ -935 +1097 @@
         self.assertEquals('internal-server-error', e.condition)
         self.assertEquals(1, len(self.flushLoggedErrors(TestError)))
 
+
     def test_notImplemented(self):
         """
-        Test response when the request is recognised but not implemented.
+        A NotImplementedError causes a feature-not-implemented response.
         """
-
         class Handler(DummyIQHandler):
             def onGet(self, iq):
                 raise NotImplementedError()
@@ -956 +1118 @@
         e = error.exceptionFromStanza(response)
         self.assertEquals('feature-not-implemented', e.condition)
 
+
     def test_noHandler(self):
         """
-        Test when the request is not recognised.
+        A missing handler causes a feature-not-implemented response.
         """
-
         iq = domish.Element((None, 'iq'))
         iq['type'] = 'set'
         iq['id'] = 'r1'