Changeset 100:e7199edcb6f5 for wokkel/subprotocols.py

Timestamp:

Aug 3, 2011, 9:49:21 AM (11 years ago)

Author:

Ralph Meijer <ralphm@…>

Branch:

default

Message:

Add iq request (set/get) tracking for StreamManager?.

This adds a new request method to StreamManager? to serialize an iq get or iq
set request object (using toElement), send it over the stream (or queue it
until the next time a stream is initialized) and track the response stanza by
returning a Deferred.

Requests may have a timeout, so that the deferred errbacks after the timeout
(counting from the moment of calling request) has expired.

This implementation is similar to the response tracking done in
twisted.words.protocols.jabber.xmlstream.IQ, but slighly different:

• Requests can be queued until the connection is initialized, even before a XmlStream? object is available. I.e. it is tied to the StreamManager?, not the stream.
• Response timeout starts counting from the moment of calling request, not from the moment request is sent over the wire.

File:

• wokkel/subprotocols.py (modified) (5 diffs)

wokkel/subprotocols.py

@@ -157 +157 @@
     @ivar _packetQueue: internal buffer of unsent data. See L{send} for details.
     @type _packetQueue: L{list}
-    """
+    @ivar timeout: Default IQ request timeout in seconds.
+    @type timeout: C{int}
+    @ivar _reactor: A provider of L{IReactorTime} to track timeouts.
+    """
+    timeout = None
+    _reactor = None
 
     logTraffic = False
 
-    def __init__(self, factory):
+    def __init__(self, factory, reactor=None):
+        """
+        Construct a stream manager.
+
+        @param factory: The stream factory to connect with.
+        @param reactor: A provider of L{IReactorTime} to track timeouts.
+            If not provided, the global reactor will be used.
+        """
         XMPPHandlerCollection.__init__(self)
         self.xmlstream = None
@@ -174 +186 @@
         self.factory = factory
 
+        if reactor is None:
+            from twisted.internet import reactor
+        self._reactor = reactor
+
+        # Set up IQ response tracking
+        self._iqDeferreds = {}
+
 
     def addHandler(self, handler):
@@ -223 +242 @@
         C{connectionInitialized} method.
         """
+
+        xs.addObserver('/iq[@type="result"]', self._onIQResponse)
+        xs.addObserver('/iq[@type="error"]', self._onIQResponse)
+
         # Flush all pending packets
         for p in self._packetQueue:
@@ -269 +292 @@
             e.connectionLost(reason)
 
+        # This errbacks all deferreds of iq's for which no response has
+        # been received with a L{ConnectionLost} failure. Otherwise, the
+        # deferreds will never be fired.
+        iqDeferreds = self._iqDeferreds
+        self._iqDeferreds = {}
+        for d in iqDeferreds.itervalues():
+            d.errback(reason)
+
+
+    def _onIQResponse(self, iq):
+        """
+        Handle iq response by firing associated deferred.
+        """
+        try:
+            d = self._iqDeferreds[iq["id"]]
+        except KeyError:
+            return
+
+        del self._iqDeferreds[iq["id"]]
+        iq.handled = True
+        if iq['type'] == 'error':
+            d.errback(error.exceptionFromStanza(iq))
+        else:
+            d.callback(iq)
+
 
     def send(self, obj):
@@ -284 +332 @@
         else:
             self._packetQueue.append(obj)
+
+
+    def request(self, request):
+        """
+        Send an IQ request and track the response.
+
+        A request is an IQ L{generic.Stanza} of type C{'get'} or C{'set'}. It
+        will have its C{toElement} called to render to a L{domish.Element}
+        which is then sent out over the current stream. If there is no such
+        stream (yet), it is queued and sent whenever a connection is
+        established and initialized, just like L{send}.
+
+        If the request doesn't have an identifier, it will be assigned a fresh
+        one, so the response can be tracked.
+
+        The deferred that is returned will fire with the L{domish.Element}
+        representation of the response if it is a result iq. If the response
+        is an error iq, a corresponding L{error.StanzaError} will be errbacked.
+
+        If the connection is closed before a response was received, the deferred
+        will be errbacked with the reason failure.
+
+        A request may also have a timeout, either by setting a default timeout
+        in L{StreamManager.timeout} or on the C{timeout} attribute of the
+        request.
+
+        @param request: The IQ request.
+        @type request: L{generic.Request}
+        """
+        if (request.stanzaKind != 'iq' or
+            request.stanzaType not in ('get', 'set')):
+            return defer.fail(ValueError("Not a request"))
+
+        element = request.toElement()
+
+        # Make sure we have a trackable id on the stanza
+        if not request.stanzaID:
+            element.addUniqueId()
+            request.stanzaID = element['id']
+
+        # Set up iq response tracking
+        d = defer.Deferred()
+        self._iqDeferreds[element['id']] = d
+
+        timeout = getattr(request, 'timeout', self.timeout)
+
+        if timeout is not None:
+            def onTimeout():
+                del self._iqDeferreds[element['id']]
+                d.errback(xmlstream.TimeoutError("IQ timed out"))
+
+            call = self._reactor.callLater(timeout, onTimeout)
+
+            def cancelTimeout(result):
+                if call.active():
+                    call.cancel()
+
+                return result
+
+            d.addBoth(cancelTimeout)
+        self.send(element)
+        return d