source: ralphm-patches/s2s-controller.patch

wokkel/server.py

@@ -557 +557 @@
         else:
             self.service.dispatch(xs, element)
 
+class XMPPS2SClientFactory(BootstrapMixin, protocol.ClientFactory):
 
+    protocol = xmlstream.XmlStream
+
+    def __init__(self, thisHost, otherHost, secret)
+        BootstrapMixin.__init__(self)
+
+        self.thisHost = thisHost
+        self.otherHost = otherHost
+
+        self.authenticator = XMPPServerConnectAuthenticator(self.thisHost,
+                                                            self.otherHost,
+                                                            secret)
+
+
+    def buildProtocol(self, addr):
+        """
+        Create an instance of XmlStream.
+
+        A new authenticator instance will be created and passed to the new
+        XmlStream. Registered bootstrap event observers are installed as well.
+        """
+        xs = self.protocol(self.authenticator)
+        xs.factory = self
+        self.installBootstraps(xs)
+        return xs
+
+
+    def startedConnecting(self, connector):
+        """
+        Called when a connection has been started.
+        """
+
+
+    def clientConnectionFailed(self, connector, reason):
+        """
+        Called when a connection has failed to connect.
+        """
+
+
+    def clientConnectionLost(self, connector, reason):
+        """
+        Called when an established connection is lost.
+        """
+
+
+class StreamController(modal.Modal):
+
+    initialMode = 'disconnected'
+
+    def __init__(self, factory):
+        factory.addBootstrap(xmlstream.STREAM_AUTHD_EVENT,
+                             self.streamValid)
+        factory.addBootstrap(xmlstream.INIT_FAILED_EVENT,
+                             self.streamInvalid)
+
+
+
+        self.xmlstream = None
+
+
+    class connecting(modal.mode):
+        def __enter__(self):
+            self.pending = []
+
+        def send(self, obj):
+            self.pending.append(obj)
+
+
+    class connected(modal.mode):
+        def send(self, obj):
+            self.xmlstream.send(obj)
+
+
+    class disconnected(modal.mode):
+        pass
+
+
+    def streamValid(self, xs):
+        """
+        Called when the stream was validated.
+        """
+        connecting = False
+        self.xmlstream = xs
+
+
+    def streamInvalid(self, xs):
+        """
+        Called when the stream was not validated.
+        """
+        connecting = False
+
+
+    def send(self, obj):
+        """
+        Send or queue obj.
+        """
+        if self.xmlstream is None:
+            # queue up
+
+        else:
+            # send immediately
 
 class ServerService(object):
     """
@@ -601 +702 @@
         xs.addObserver(xmlstream.STREAM_END_EVENT,
                        lambda _: self.outgoingDisconnected(xs))
 
-        if (thisHost, otherHost) in self._outgoingQueues:
-            for element in self._outgoingQueues[thisHost, otherHost]:
-                xs.send(element)
-            del self._outgoingQueues[thisHost, otherHost]
-
-
     def outgoingDisconnected(self, xs):
         thisHost = xs.thisEntity.host
         otherHost = xs.otherEntity.host
@@ -622 +717 @@
         Initiate an outgoing XMPP server-to-server connection.
         """
 
-        def resetConnecting(_):
+        def resetConnecting(result):
             self._outgoingConnecting.remove((thisHost, otherHost))
+            return result
 
-        if (thisHost, otherHost) in self._outgoingConnecting:
-            return
+        #if (thisHost, otherHost) in self._outgoingConnecting:
+        #    return
 
         authenticator = XMPPServerConnectAuthenticator(thisHost,
                                                        otherHost,
                                                        self.secret)
         factory = DeferredS2SClientFactory(authenticator)
-        factory.addBootstrap(xmlstream.STREAM_AUTHD_EVENT,
-                             self.outgoingInitialized)
+        #factory.addBootstrap(xmlstream.STREAM_AUTHD_EVENT,
+        #                     self.outgoingInitialized)
         factory.logTraffic = self.logTraffic
 
         self._outgoingConnecting.add((thisHost, otherHost))
 
         d = initiateS2S(factory)
         d.addBoth(resetConnecting)
+        d.addCallback(self.outgoingInitialized)
         return d
 
 

wokkel/test/test_server.py

@@ -448 +448 @@
         self.service.dispatch(self.xmlstream, stanza)
 
         self.assertEqual(1, len(errors))
+
+
+    def test_send(self):
+        print
+        factories = []
+        inits = []
+
+        def initiateS2S(factory):
+            factories.append(factory)
+            return factory.deferred
+
+        self.patch(server, 'initiateS2S', initiateS2S)
+        #self.service.outgoingInitialized = lambda xs: inits.append(xs)
+        stanza = domish.Element((None, "presence"))
+        stanza['to'] = 'other@example.com'
+        stanza['from'] = 'user@example.org'
+
+        print "Sending first stanza"
+        self.service.send(stanza)
+        print "Output after first send", self.output
+        xs = factories[-1].buildProtocol(None)
+        xs.thisEntity = jid.JID('example.org')
+        xs.otherEntity = jid.JID('example.com')
+        xs.serial = 1
+        xs.send = self.output.append
+
+        print "Authenticating"
+        xs.dispatch(xs, xmlstream.STREAM_AUTHD_EVENT)
+        print "Output after authd", self.output
+
+        print "Sending second stanza"
+        self.service.send(stanza)
+        print "Output after second send", self.output
+
+