source: ralphm-patches/xmpp_client_service.patch @ 37:03cd0cb8548c

new file doc/examples/client_service.tac

@@ +1 @@
+from twisted.application import service, strports
+from twisted.internet import defer
+
+from wokkel import client
+from wokkel.generic import FallbackHandler
+from wokkel.subprotocols import XMPPHandler
+from wokkel.xmppim import RosterItem, RosterServerProtocol
+
+from twisted.words.protocols.jabber.jid import internJID as JID
+
+import socket
+domain = socket.gethostname()
+
+class StaticRoster(RosterServerProtocol):
+
+    def __init__(self):
+        RosterServerProtocol.__init__(self)
+        self.roster = {'ralphm': [
+                           RosterItem(JID('intosi@'+domain),
+                                      subscriptionTo=True,
+                                      subscriptionFrom=True,
+                                      name='Intosi',
+                                      groups=set(['Friends'])),
+                           RosterItem(JID('termie@'+domain),
+                                      subscriptionTo=True,
+                                      subscriptionFrom=True,
+                                      name='termie'),
+                           ],
+                       'test': [],
+                       }
+
+    def getRoster(self, entity):
+        return defer.succeed(self.roster[entity.user])
+
+
+class Hello(XMPPHandler):
+
+    def q(self):
+        from wokkel.xmppim import AvailabilityPresence
+        p = AvailabilityPresence(JID('ralphm@'+domain+'/default'), JID('termie@'+domain+'/Home'), show='chat')
+        self.parent.send(p.toElement())
+
+
+    def connectionInitialized(self):
+        from twisted.internet import reactor
+        reactor.callLater(5, self.q)
+
+
+
+application = service.Application("Jabber server")
+
+sessionManager = client.SessionManager()
+FallbackHandler().setHandlerParent(sessionManager)
+StaticRoster().setHandlerParent(sessionManager)
+Hello().setHandlerParent(sessionManager)
+
+clientService = client.ClientService(sessionManager, domain)
+clientService.logTraffic = True
+
+
+c2sFactory = client.XMPPC2SServerFactory(clientService)
+c2sFactory.logTraffic = True
+c2sService = strports.service('5224', c2sFactory)
+c2sService.setServiceParent(application)

wokkel/client.py

@@ -17 +17 @@
 from twisted.names.srvconnect import SRVConnector
 from twisted.python import log
 from twisted.words.protocols.jabber import client, error, sasl, xmlstream
-from twisted.words.protocols.jabber.jid import internJID as JID
+from twisted.words.protocols.jabber.jid import JID, internJID
 from twisted.words.xish import domish
 
 from wokkel import generic
-from wokkel.subprotocols import StreamManager
+from wokkel.compat import XmlStreamServerFactory
+from wokkel.subprotocols import StreamManager, XMPPHandlerCollection
 
 NS_CLIENT = 'jabber:client'
 
@@ -311 +312 @@
 
         # TODO: check for resource conflicts
 
+        print self.username, self.domain, self.resource
         newJID = JID(tuple=(self.username, self.domain, self.resource))
 
         reply = domish.Element((None, 'iq'))
@@ -350 +352 @@
             reply['id'] = iq['id']
         reply.addElement((client.NS_XMPP_SESSION, 'session'))
         self.xmlstream.send(reply)
+
+
+
+class XMPPC2SServerFactory(XmlStreamServerFactory):
+
+    def __init__(self, service):
+        self.service = service
+
+        def authenticatorFactory():
+            return XMPPClientListenAuthenticator(service.domain)
+
+        XmlStreamServerFactory.__init__(self, authenticatorFactory)
+        self.addBootstrap(xmlstream.STREAM_CONNECTED_EVENT,
+                          self.onConnectionMade)
+        self.addBootstrap(xmlstream.STREAM_AUTHD_EVENT,
+                          self.onAuthenticated)
+
+        self.serial = 0
+
+
+    def onConnectionMade(self, xs):
+        """
+        Called when a client-to-server connection was made.
+
+        This enables traffic debugging on incoming streams.
+        """
+        xs.serial = self.serial
+        self.serial += 1
+
+        log.msg("Client connection %d made" % xs.serial)
+
+        def logDataIn(buf):
+            log.msg("RECV (%d): %r" % (xs.serial, buf))
+
+        def logDataOut(buf):
+            log.msg("SEND (%d): %r" % (xs.serial, buf))
+
+        if self.logTraffic:
+            xs.rawDataInFn = logDataIn
+            xs.rawDataOutFn = logDataOut
+
+        xs.addObserver(xmlstream.STREAM_ERROR_EVENT, self.onError)
+
+
+    def onAuthenticated(self, xs):
+        log.msg("Client connection %d authenticated" % xs.serial)
+
+        xs.addObserver(xmlstream.STREAM_END_EVENT, self.onConnectionLost,
+                                                   0, xs)
+        xs.addObserver('/*', self.onElement, 0, xs)
+
+        self.service.connectionInitialized(xs)
+
+
+    def onConnectionLost(self, xs, reason):
+        log.msg("Client connection %d disconnected" % xs.serial)
+
+        self.service.connectionLost(xs, reason)
+
+
+    def onError(self, reason):
+        log.err(reason, "Stream Error")
+
+
+    def onElement(self, xs, element):
+        """
+        Called when an element was received from one of the connected streams.
+
+        """
+        if element.handled:
+            return
+        else:
+            self.service.dispatch(xs, element)
+
+
+
+class SessionManager(XMPPHandlerCollection):
+
+    def __init__(self):
+        XMPPHandlerCollection.__init__(self)
+        self.xmlstream = None
+
+
+    def makeConnection(self, xs):
+        self.xmlstream = xs
+
+        for handler in self:
+            handler.makeConnection(xs)
+            handler.connectionInitialized()
+
+
+    def send(self, obj):
+        if self.xmlstream:
+            self.xmlstream.send(obj)
+
+
+class ClientService(object):
+    """
+    Service for accepting XMPP client connections.
+
+    Incoming client connections are first authenticated using the
+    L{XMPPClientListenAuthenticator}, and then kept in a dictionary that is
+    keyed with the JID that was bound to that connection.
+
+    Where L{StreamManager} manages one connection at a time, this
+    service demultiplexes incoming connections. For the subprotocol handlers
+    (objects providing {ijabber.IXMPPHandler}), their stream is always
+    connected and initialized, but they only have to deal with one stream. This
+    makes it easier to create adapters.
+
+    As an L{xmlstream.XMPPHandlerCollection}, this service creates a fake XML
+    stream that is passed to the XMPP subprotocol handlers. The received
+    stanzas from the incoming client connections are passed on to the handlers
+    using this fake XML, while having their C{from} attribute set to the JID of
+    the client stream. Stanzas sent by the handlers are forwarded to the
+    matching client stream, selected by the stanzas C{to} attribute.
+    """
+
+    logTraffic = False
+
+    def __init__(self, sessionManager, domain):
+        self.sessionManager = sessionManager
+        self.domain = domain
+
+        self.streams = {}
+
+        pipe = generic.XmlPipe()
+        self.xmlstream = pipe.source
+        self.sessionManager.makeConnection(pipe.sink)
+        self.xmlstream.addObserver('/*', self.send)
+
+
+    def connectionInitialized(self, xs):
+        self.streams[xs.otherEntity] = xs
+
+
+    def connectionLost(self, xs, reason):
+        if xs.otherEntity in self.streams:
+            del self.streams[xs.otherEntity]
+
+
+    def send(self, stanza):
+        """
+        Send stanza to the proper XML Stream.
+
+        This uses addressing embedded in the element to find the correct stream
+        to forward the element to.
+        """
+        destination = internJID(stanza["to"])
+
+        if destination not in self.streams:
+            log.msg("Euh")
+            raise Exception("Destination unreachable")
+
+        self.streams[destination].send(stanza)
+
+
+    def dispatch(self, xs, stanza):
+        """
+        Called when an element was received from one of the connected streams.
+        """
+        stanza["from"] = xs.otherEntity.full()
+        self.xmlstream.send(stanza)