source: ralphm-patches/c2s_server_factory.patch @ 71:7b5ddbb2b9bb

# HG changeset patch
# Parent 8cb185c88211ac69faa924f6e93425d102daa2bc
Add factory for accepting client connections.

The new XMPPC2SServerFactory is a server factory for accepting client
connections. It uses `XMPPClientListenAuthenticator` to perform the
steps for authentication and binding of a resource, and keeps a list
of all streams.

Upon loss of the connection, the service is called with `unbindResource`.
Received stanzas cause the service's `onElement` to be called.

The factory has a `deliverStanza` method to deliver stanzas to a particular
recipient. This is used for stanzas that have different recipient addressing
than the actual recipient (for presence and messages from a different (or no)
resource).

TODO:

 * Add docstrings.
 * Add tests.

diff --git a/wokkel/client.py b/wokkel/client.py
--- a/wokkel/client.py
+++ b/wokkel/client.py
@@ -21,7 +21,9 @@
 from twisted.words.xish import domish
 
 from wokkel import generic
+from wokkel.compat import XmlStreamServerFactory
 from wokkel.iwokkel import IUserSession
+from wokkel.subprotocols import ServerStreamManager
 from wokkel.subprotocols import StreamManager
 
 NS_CLIENT = 'jabber:client'
@@ -442,3 +444,41 @@
             self.portal = self.portals[self.xmlstream.thisEntity]
         except KeyError:
             raise error.StreamError('host-unknown')
+
+
+
+class XMPPC2SServerFactory(XmlStreamServerFactory):
+
+    def __init__(self, portals):
+        def authenticatorFactory():
+            return XMPPClientListenAuthenticator(portals)
+
+        XmlStreamServerFactory.__init__(self, authenticatorFactory)
+        self.addBootstrap(xmlstream.STREAM_CONNECTED_EVENT,
+                          self.onConnectionMade)
+
+
+    def onConnectionMade(self, xs):
+        """
+        Called when a connection is made.
+
+        This creates a stream manager, calls L{setupHandlers} to attach
+        subprotocol handlers and then signals the stream manager that
+        the connection was made.
+        """
+        sm = ServerStreamManager()
+        sm.logTraffic = self.logTraffic
+
+        for handler in self.setupHandlers():
+            handler.setHandlerParent(sm)
+
+        sm.makeConnection(xs)
+
+
+    def setupHandlers(self):
+        """
+        Set up XMPP subprotocol handlers.
+        """
+        return [
+            generic.StanzaForwarder()
+        ]
diff --git a/wokkel/generic.py b/wokkel/generic.py
--- a/wokkel/generic.py
+++ b/wokkel/generic.py
@@ -615,3 +615,55 @@
 
         self._initializers = self.getInitializers()
         self._initializeStream()
+
+
+
+class StanzaForwarder(XMPPHandler):
+    """
+    XMPP protocol for passing incoming stanzas to the stream avatar.
+
+    This handler adds an observer for all XML Stanzas to forward to the C{send}
+    method on the cred avatar set on the XML Stream, unless it has been handled
+    by other observers.
+
+    Stream errors are logged.
+    """
+
+    def connectionMade(self):
+        """
+        Called when a connection is made.
+        """
+        self.xmlstream.addObserver(xmlstream.STREAM_ERROR_EVENT, self.onError)
+
+
+    def connectionInitialized(self):
+        """
+        Called when the stream has been initialized.
+        """
+        self.xmlstream.addObserver(
+            '/*[@xmlns="%s"]' % self.xmlstream.namespace,
+            self.onStanza, priority=-1)
+
+
+    def onStanza(self, element):
+        """
+        Called when a stanza element was received.
+
+        Unless a stanza has already been handled, or the name of the element is
+        not one of C{'iq'}, C{'message'}, C{'presence'}, the stanza is passed
+        on to the avatar's C{send} method.
+        """
+        if element.handled:
+            return
+
+        if element.name not in ('iq', 'message', 'presence'):
+            return
+
+        self.xmlstream.avatar.send(element)
+
+
+    def onError(reason):
+        """
+        Log a stream error.
+        """
+        log.err(reason, "Stream error")
diff --git a/wokkel/subprotocols.py b/wokkel/subprotocols.py
--- a/wokkel/subprotocols.py
+++ b/wokkel/subprotocols.py
@@ -142,10 +142,13 @@
     @type timeout: C{int}
 
     @ivar _reactor: A provider of L{IReactorTime} to track timeouts.
+
+    @cvar __streamCount: Global stream count for distinguishing streams.
     """
 
     timeout = None
     _reactor = None
+    __streamCount = 0
 
     logTraffic = False
 
@@ -195,20 +198,24 @@
         and call each handler's C{makeConnection} method with the L{XmlStream}
         instance.
         """
-        def logDataIn(buf):
-            log.msg("RECV: %r" % buf)
+        xs.serial = self.__streamCount
+        BaseStreamManager.__streamCount += 1
 
-        def logDataOut(buf):
-            log.msg("SEND: %r" % buf)
 
         if self.logTraffic:
-            xs.rawDataInFn = logDataIn
-            xs.rawDataOutFn = logDataOut
+            def logData(direction, data):
+                log.msg(format="%(direction)s (%(streamID)s): %(data)r",
+                        direction=direction, streamID=xs.serial, data=data)
+
+            log.msg(format="Connection %(streamID)s made", streamID=xs.serial)
+            xs.rawDataInFn = lambda data: logData("RECV", data)
+            xs.rawDataOutFn = lambda data: logData("SEND", data)
 
         xs.addObserver(xmlstream.STREAM_AUTHD_EVENT,
                        self.connectionInitialized)
         xs.addObserver(xmlstream.STREAM_END_EVENT,
                        self.connectionLost)
+
         self.xmlstream = xs
 
         for e in list(self):
@@ -222,6 +229,9 @@
         Send out cached stanzas and call each handler's
         C{connectionInitialized} method.
         """
+        if self.logTraffic:
+            log.msg(format="Connection %(streamID)s initialized",
+                    streamID=xs.serial)
 
         xs.addObserver('/iq[@type="result"]', self._onIQResponse)
         xs.addObserver('/iq[@type="error"]', self._onIQResponse)
@@ -247,6 +257,10 @@
         L{XmlStream} anymore and notifies each handler that the connection
         was lost by calling its C{connectionLost} method.
         """
+        if self.logTraffic:
+            log.msg(format="Connection %(streamID)s lost",
+                    streamID=self.xmlstream.serial)
+
         self.xmlstream = None
         self._initialized = False