source: ralphm-patches/server-stream-manager.patch @ 74:355afce3af27

# HG changeset patch
# Parent 71082c22f73200a0b8b4b5c6e5d2773f9f68dd4d
Generalize StreamManager and add ServerStreamManager.

This generalizes `StreamManager` to `BaseStreamManager` to take the
common functionality and reuse it for `ServerStreamManager`. Where
`StreamManager` is used for initiating connections, with a factory as
a parameter, `ServerStreamManager` works for receiving connections. It
must be created in a protocol factory when a connection is established,
and then its `makeConnection` should be called to hook it up to the
`XmlStream` instance.

diff --git a/wokkel/client.py b/wokkel/client.py
--- a/wokkel/client.py
+++ b/wokkel/client.py
@@ -109,7 +109,7 @@
         self._connection.disconnect()
 
 
-    def _authd(self, xs):
+    def connectionInitialized(self, xs):
         """
         Called when the stream has been initialized.
 
@@ -118,7 +118,7 @@
         by its constituent initializers.
         """
         self.jid = self.factory.authenticator.jid
-        StreamManager._authd(self, xs)
+        StreamManager.connectionInitialized(self, xs)
 
 
     def initializationFailed(self, reason):
diff --git a/wokkel/component.py b/wokkel/component.py
--- a/wokkel/component.py
+++ b/wokkel/component.py
@@ -36,7 +36,7 @@
         StreamManager.__init__(self, factory)
 
 
-    def _authd(self, xs):
+    def connectionInitialized(self, xs):
         """
         Called when stream initialization has completed.
 
@@ -53,7 +53,7 @@
             old_send(obj)
 
         xs.send = send
-        StreamManager._authd(self, xs)
+        StreamManager.connectionInitialized(self, xs)
 
 
     def initializationFailed(self, reason):
diff --git a/wokkel/subprotocols.py b/wokkel/subprotocols.py
--- a/wokkel/subprotocols.py
+++ b/wokkel/subprotocols.py
@@ -120,7 +120,7 @@
 
 
 
-class StreamManager(XMPPHandlerCollection):
+class BaseStreamManager(XMPPHandlerCollection):
     """
     Business logic representing a managed XMPP connection.
 
@@ -131,43 +131,39 @@
 
     @ivar xmlstream: currently managed XML stream
     @type xmlstream: L{XmlStream}
+
     @ivar logTraffic: if true, log all traffic.
     @type logTraffic: C{bool}
+
     @ivar _initialized: Whether the stream represented by L{xmlstream} has
                         been initialized. This is used when caching outgoing
                         stanzas.
     @type _initialized: C{bool}
-    @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, reactor=None):
+    def __init__(self, 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
         self._packetQueue = []
         self._initialized = False
 
-        factory.addBootstrap(xmlstream.STREAM_CONNECTED_EVENT, self._connected)
-        factory.addBootstrap(xmlstream.STREAM_AUTHD_EVENT, self._authd)
-        factory.addBootstrap(xmlstream.INIT_FAILED_EVENT,
-                             self.initializationFailed)
-        factory.addBootstrap(xmlstream.STREAM_END_EVENT, self._disconnected)
-        self.factory = factory
-
         if reactor is None:
             from twisted.internet import reactor
         self._reactor = reactor
@@ -193,7 +189,7 @@
             handler.connectionInitialized()
 
 
-    def _connected(self, xs):
+    def makeConnection(self, xs):
         """
         Called when the transport connection has been established.
 
@@ -211,13 +207,17 @@
             xs.rawDataInFn = logDataIn
             xs.rawDataOutFn = logDataOut
 
+        xs.addObserver(xmlstream.STREAM_AUTHD_EVENT,
+                       self.connectionInitialized)
+        xs.addObserver(xmlstream.STREAM_END_EVENT,
+                       self.connectionLost)
         self.xmlstream = xs
 
         for e in list(self):
             e.makeConnection(xs)
 
 
-    def _authd(self, xs):
+    def connectionInitialized(self, xs):
         """
         Called when the stream has been initialized.
 
@@ -240,21 +240,8 @@
             e.connectionInitialized()
 
 
-    def initializationFailed(self, reason):
-        """
-        Called when stream initialization has failed.
 
-        Stream initialization has halted, with the reason indicated by
-        C{reason}. It may be retried by calling the authenticator's
-        C{initializeStream}. See the respective authenticators for details.
-
-        @param reason: A failure instance indicating why stream initialization
-                       failed.
-        @type reason: L{failure.Failure}
-        """
-
-
-    def _disconnected(self, reason):
+    def connectionLost(self, reason):
         """
         Called when the stream has been closed.
 
@@ -464,6 +451,60 @@
 
 
 
+class StreamManager(BaseStreamManager):
+    """
+    Business logic representing a managed XMPP client connection.
+
+    @ivar _packetQueue: internal buffer of unsent data. See L{send} for details.
+    @type _packetQueue: L{list}
+    """
+
+    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.
+        """
+        BaseStreamManager.__init__(self, reactor=reactor)
+
+        factory.addBootstrap(xmlstream.STREAM_CONNECTED_EVENT,
+                             self.makeConnection)
+        self.factory = factory
+
+
+    def makeConnection(self, xs):
+        xs.addObserver(xmlstream.INIT_FAILED_EVENT,
+                       self.initializationFailed)
+        BaseStreamManager.makeConnection(self, xs)
+
+
+    def initializationFailed(self, reason):
+        """
+        Called when stream initialization has failed.
+
+        Stream initialization has halted, with the reason indicated by
+        C{reason}. It may be retried by calling the authenticator's
+        C{initializeStream}. See the respective authenticators for details.
+
+        @param reason: A failure instance indicating why stream initialization
+                       failed.
+        @type reason: L{failure.Failure}
+        """
+
+
+
+class ServerStreamManager(BaseStreamManager):
+    """
+    Business logic representing a managed server-side XMPP connection.
+    """
+
+    def __init__(self, reactor=None):
+        BaseStreamManager.__init__(self, reactor=reactor)
+
+
+
 class IQHandlerMixin(object):
     """
     XMPP subprotocol mixin for handle incoming IQ stanzas.
diff --git a/wokkel/test/helpers.py b/wokkel/test/helpers.py
--- a/wokkel/test/helpers.py
+++ b/wokkel/test/helpers.py
@@ -108,5 +108,5 @@
         factory = DummyFactory()
         StreamManager.__init__(self, factory, reactor)
         self.stub = XmlStreamStub()
-        self._connected(self.stub.xmlstream)
-        self._authd(self.stub.xmlstream)
+        self.makeConnection(self.stub.xmlstream)
+        self.connectionInitialized(self.stub.xmlstream)
diff --git a/wokkel/test/test_client.py b/wokkel/test/test_client.py
--- a/wokkel/test/test_client.py
+++ b/wokkel/test/test_client.py
@@ -38,7 +38,7 @@
         """
         xs = self.client.factory.buildProtocol(None)
         self.client.factory.authenticator.jid = JID('user@example.org/test')
-        xs.dispatch(xs, xmlstream.STREAM_AUTHD_EVENT)
+        self.client.connectionInitialized(xs)
         self.assertEquals(JID('user@example.org/test'), self.client.jid)
 
 
diff --git a/wokkel/test/test_component.py b/wokkel/test/test_component.py
--- a/wokkel/test/test_component.py
+++ b/wokkel/test/test_component.py
@@ -49,11 +49,18 @@
     def __init__(self, *args, **kwargs):
         component.Component.__init__(self, *args, **kwargs)
         self.factory.clock = Clock()
+        self.output = []
 
 
     def _getConnection(self):
         c = FakeConnector(self.factory, None, None)
         c.connect()
+        xs = self.factory.buildProtocol(None)
+        xs.send = self.output.append
+        xs.connectionMade()
+        self.makeConnection(xs)
+        xs.thisEntity = xs.otherEntity
+        xs.dispatch(xs, xmlstream.STREAM_AUTHD_EVENT)
         return c
 
 
@@ -104,6 +111,21 @@
         self.assertEqual(1, connector.connects)
 
 
+    def test_stampFrom(self):
+        """
+        Outgoing elements with missing sender address get component JID.
+        """
+        comp = TestableComponent('example.org', 5347,
+                                 'test.example.org', 'secret')
+        comp.startService()
+
+        element = domish.Element((component.NS_COMPONENT_ACCEPT, "message"))
+        element["to"] = "test@example.org"
+        comp.xmlstream.send(element)
+
+        self.assertEqual('test.example.org', element.getAttribute("from"))
+
+
 
 class InternalComponentTest(unittest.TestCase):
     """
diff --git a/wokkel/test/test_subprotocols.py b/wokkel/test/test_subprotocols.py
--- a/wokkel/test/test_subprotocols.py
+++ b/wokkel/test/test_subprotocols.py
@@ -189,20 +189,7 @@
 
 
 
-class StreamManagerTest(unittest.TestCase):
-    """
-    Tests for L{subprotocols.StreamManager}.
-    """
-
-    def setUp(self):
-        factory = xmlstream.XmlStreamFactory(xmlstream.Authenticator())
-        self.clock = task.Clock()
-        self.streamManager = subprotocols.StreamManager(factory, self.clock)
-        self.xmlstream = factory.buildProtocol(None)
-        self.transport = proto_helpers.StringTransport()
-        self.xmlstream.transport = self.transport
-
-        self.request = IQGetStanza()
+class BaseStreamManagerTestsMixin(object):
 
     def _streamStarted(self):
         """
@@ -216,25 +203,14 @@
         self.xmlstream.dispatch(self.xmlstream, "//event/stream/authd")
 
 
-    def test_basic(self):
+    def test_interface(self):
         """
-        Test correct initialization and setup of factory observers.
+        ServerStreamManager implements IXMPPHandlerCollection.
         """
-        factory = DummyFactory()
-        sm = subprotocols.StreamManager(factory)
-        self.assertIdentical(None, sm.xmlstream)
-        self.assertEquals([], sm.handlers)
-        self.assertEquals(sm._connected,
-                          sm.factory.callbacks['//event/stream/connected'])
-        self.assertEquals(sm._authd,
-                          sm.factory.callbacks['//event/stream/authd'])
-        self.assertEquals(sm._disconnected,
-                          sm.factory.callbacks['//event/stream/end'])
-        self.assertEquals(sm.initializationFailed,
-                          sm.factory.callbacks['//event/xmpp/initfailed'])
+        verifyObject(ijabber.IXMPPHandlerCollection, self.streamManager)
 
 
-    def test_connected(self):
+    def test_makeConnection(self):
         """
         Test that protocol handlers have their connectionMade method called
         when the XML stream is connected.
@@ -242,27 +218,27 @@
         sm = self.streamManager
         handler = DummyXMPPHandler()
         handler.setHandlerParent(sm)
-        xs = xmlstream.XmlStream(xmlstream.Authenticator())
-        sm._connected(xs)
+        xs = self.xmlstream
+        sm.makeConnection(xs)
         self.assertEquals(1, handler.doneMade)
         self.assertEquals(0, handler.doneInitialized)
         self.assertEquals(0, handler.doneLost)
 
 
-    def test_connectedLogTrafficFalse(self):
+    def test_makeConnectionLogTrafficFalse(self):
         """
         Test raw data functions unset when logTraffic is set to False.
         """
         sm = self.streamManager
         handler = DummyXMPPHandler()
         handler.setHandlerParent(sm)
-        xs = xmlstream.XmlStream(xmlstream.Authenticator())
-        sm._connected(xs)
+        xs = self.xmlstream
+        sm.makeConnection(xs)
         self.assertIdentical(None, xs.rawDataInFn)
         self.assertIdentical(None, xs.rawDataOutFn)
 
 
-    def test_connectedLogTrafficTrue(self):
+    def test_makeConnectionLogTrafficTrue(self):
         """
         Test raw data functions set when logTraffic is set to True.
         """
@@ -270,13 +246,13 @@
         sm.logTraffic = True
         handler = DummyXMPPHandler()
         handler.setHandlerParent(sm)
-        xs = xmlstream.XmlStream(xmlstream.Authenticator())
-        sm._connected(xs)
+        xs = self.xmlstream
+        sm.makeConnection(xs)
         self.assertNotIdentical(None, xs.rawDataInFn)
         self.assertNotIdentical(None, xs.rawDataOutFn)
 
 
-    def test_authd(self):
+    def test_connectionInitialized(self):
         """
         Test that protocol handlers have their connectionInitialized method
         called when the XML stream is initialized.
@@ -284,27 +260,27 @@
         sm = self.streamManager
         handler = DummyXMPPHandler()
         handler.setHandlerParent(sm)
-        xs = xmlstream.XmlStream(xmlstream.Authenticator())
-        sm._authd(xs)
+        xs = self.xmlstream
+        sm.connectionInitialized(xs)
         self.assertEquals(0, handler.doneMade)
         self.assertEquals(1, handler.doneInitialized)
         self.assertEquals(0, handler.doneLost)
 
 
-    def test_disconnected(self):
+    def test_connectionLost(self):
         """
         Protocol handlers have connectionLost called on stream disconnect.
         """
         sm = self.streamManager
         handler = DummyXMPPHandler()
         handler.setHandlerParent(sm)
-        sm._disconnected(None)
+        sm.connectionLost(None)
         self.assertEquals(0, handler.doneMade)
         self.assertEquals(0, handler.doneInitialized)
         self.assertEquals(1, handler.doneLost)
 
 
-    def test_disconnectedReason(self):
+    def test_connectionLostReason(self):
         """
         A L{STREAM_END_EVENT} results in L{StreamManager} firing the handlers
         L{connectionLost} methods, passing a L{failure.Failure} reason.
@@ -313,7 +289,7 @@
         handler = FailureReasonXMPPHandler()
         handler.setHandlerParent(sm)
         xmlstream.XmlStream(xmlstream.Authenticator())
-        sm._disconnected(failure.Failure(Exception("no reason")))
+        sm.connectionLost(failure.Failure(Exception("no reason")))
         self.assertEquals(True, handler.gotFailureReason)
 
 
@@ -335,8 +311,8 @@
         Adding a handler when connected doesn't call connectionInitialized.
         """
         sm = self.streamManager
-        xs = xmlstream.XmlStream(xmlstream.Authenticator())
-        sm._connected(xs)
+        xs = self.xmlstream
+        sm.makeConnection(xs)
         handler = DummyXMPPHandler()
         handler.setHandlerParent(sm)
 
@@ -358,10 +334,10 @@
                 self.nestedHandler.setHandlerParent(self.parent)
 
         sm = self.streamManager
-        xs = xmlstream.XmlStream(xmlstream.Authenticator())
+        xs = self.xmlstream
         handler = NestingHandler()
         handler.setHandlerParent(sm)
-        sm._connected(xs)
+        sm.makeConnection(xs)
 
         self.assertEquals(1, handler.doneMade)
         self.assertEquals(0, handler.doneInitialized)
@@ -383,9 +359,9 @@
         called.
         """
         sm = self.streamManager
-        xs = xmlstream.XmlStream(xmlstream.Authenticator())
-        sm._connected(xs)
-        sm._authd(xs)
+        xs = self.xmlstream
+        sm.makeConnection(xs)
+        sm.connectionInitialized(xs)
         handler = DummyXMPPHandler()
         handler.setHandlerParent(sm)
 
@@ -407,11 +383,11 @@
                 self.nestedHandler.setHandlerParent(self.parent)
 
         sm = self.streamManager
-        xs = xmlstream.XmlStream(xmlstream.Authenticator())
+        xs = self.xmlstream
         handler = NestingHandler()
         handler.setHandlerParent(sm)
-        sm._connected(xs)
-        sm._authd(xs)
+        sm.makeConnection(xs)
+        sm.connectionInitialized(xs)
 
         self.assertEquals(1, handler.doneMade)
         self.assertEquals(1, handler.doneInitialized)
@@ -435,12 +411,12 @@
                 self.nestedHandler.setHandlerParent(self.parent)
 
         sm = self.streamManager
-        xs = xmlstream.XmlStream(xmlstream.Authenticator())
+        xs = self.xmlstream
         handler = NestingHandler()
         handler.setHandlerParent(sm)
-        sm._connected(xs)
-        sm._authd(xs)
-        sm._disconnected(xs)
+        sm.makeConnection(xs)
+        sm.connectionInitialized(xs)
+        sm.connectionLost(xs)
 
         self.assertEquals(1, handler.doneMade)
         self.assertEquals(1, handler.doneInitialized)
@@ -470,16 +446,13 @@
 
         The data should be sent directly over the XML stream.
         """
-        factory = xmlstream.XmlStreamFactory(xmlstream.Authenticator())
-        sm = subprotocols.StreamManager(factory)
-        xs = factory.buildProtocol(None)
-        xs.transport = proto_helpers.StringTransport()
+        xs = self.xmlstream
         xs.connectionMade()
         xs.dataReceived("<stream:stream xmlns='jabber:client' "
                         "xmlns:stream='http://etherx.jabber.org/streams' "
                         "from='example.com' id='12345'>")
         xs.dispatch(xs, "//event/stream/authd")
-        sm.send("<presence/>")
+        self.streamManager.send("<presence/>")
         self.assertEquals("<presence/>", xs.transport.value())
 
 
@@ -490,12 +463,11 @@
         The data should be cached until an XML stream has been established and
         initialized.
         """
-        factory = xmlstream.XmlStreamFactory(xmlstream.Authenticator())
-        sm = subprotocols.StreamManager(factory)
+        sm = self.streamManager
         handler = DummyXMPPHandler()
         sm.addHandler(handler)
 
-        xs = factory.buildProtocol(None)
+        xs = self.xmlstream
         xs.transport = proto_helpers.StringTransport()
         sm.send("<presence/>")
         self.assertEquals("", xs.transport.value())
@@ -522,7 +494,7 @@
         """
         factory = xmlstream.XmlStreamFactory(xmlstream.Authenticator())
         sm = subprotocols.StreamManager(factory)
-        xs = factory.buildProtocol(None)
+        xs = self.xmlstream
         xs.transport = proto_helpers.StringTransport()
         xs.connectionMade()
         xs.dataReceived("<stream:stream xmlns='jabber:client' "
@@ -545,7 +517,7 @@
         handler = DummyXMPPHandler()
         sm.addHandler(handler)
 
-        xs = factory.buildProtocol(None)
+        xs = self.xmlstream
         xs.connectionMade()
         xs.transport = proto_helpers.StringTransport()
         xs.connectionLost(None)
@@ -677,6 +649,7 @@
         """
         d = self.streamManager.request(self.request)
         xs = self.xmlstream
+        xs.connectionMade()
         xs.connectionLost(failure.Failure(ConnectionDone()))
         self.assertFailure(d, ConnectionDone)
         return d
@@ -692,6 +665,7 @@
             d = xmlstream.IQ(self.xmlstream).send()
             d.addErrback(eb)
 
+        self.xmlstream.connectionMade()
         d = self.streamManager.request(self.request)
         d.addErrback(eb)
         self.xmlstream.connectionLost(failure.Failure(ConnectionDone()))
@@ -736,6 +710,7 @@
         self.request.timeout = 60
         d = self.streamManager.request(self.request)
 
+        self.xmlstream.connectionMade()
         self.xmlstream.connectionLost(failure.Failure(ConnectionDone()))
         self.assertFailure(d, ConnectionDone)
         self.assertFalse(self.clock.calls)
@@ -778,6 +753,56 @@
 
 
 
+class StreamManagerTest(unittest.TestCase, BaseStreamManagerTestsMixin):
+    """
+    Tests for L{subprotocols.StreamManager}.
+    """
+
+    def setUp(self):
+        factory = xmlstream.XmlStreamFactory(xmlstream.Authenticator())
+        self.clock = task.Clock()
+        self.streamManager = subprotocols.StreamManager(factory, self.clock)
+        self.xmlstream = factory.buildProtocol(None)
+        self.transport = proto_helpers.StringTransport()
+        self.xmlstream.transport = self.transport
+
+        self.request = IQGetStanza()
+
+
+    def test_basic(self):
+        """
+        Test correct initialization and setup of factory observers.
+        """
+        factory = DummyFactory()
+        sm = subprotocols.StreamManager(factory)
+        self.assertIdentical(None, sm.xmlstream)
+        self.assertEquals([], sm.handlers)
+        self.assertEquals(sm.makeConnection,
+                          sm.factory.callbacks['//event/stream/connected'])
+
+
+
+class ServerStreamManagerTest(unittest.TestCase, BaseStreamManagerTestsMixin):
+    """
+    Tests for L{subprotocols.StreamManager}.
+    """
+
+    def setUp(self):
+        self.clock = task.Clock()
+        self.streamManager = subprotocols.ServerStreamManager(
+            reactor=self.clock)
+
+        self.xmlstream = xmlstream.XmlStream(xmlstream.Authenticator())
+        self.transport = proto_helpers.StringTransport()
+        self.xmlstream.transport = self.transport
+
+        self.xmlstream.addObserver(xmlstream.STREAM_CONNECTED_EVENT,
+                                   self.streamManager.makeConnection)
+
+        self.request = IQGetStanza()
+
+
+
 class DummyIQHandler(subprotocols.IQHandlerMixin):
     iqHandlers = {'/iq[@type="get"]': 'onGet'}