source: ralphm-patches/session_manager.patch

new file wokkel/ewokkel.py

@@ +1 @@
+# Copyright (c) Ralph Meijer.
+# See LICENSE for details.
+
+"""
+Exceptions for Wokkel.
+"""
+
+
+class WokkelError(Exception):
+    """
+    Base exception for Wokkel.
+    """
+
+class NoSuchContact(WokkelError):
+    """
+    Raised when the given contact is not present in the user's roster.
+    """
+
+class NotSubscribed(WokkelError):
+    """
+    Raised when the contact does not have a presence subscription to the user.
+    """
+
+class NoSuchResource(WokkelError):
+    """
+    Raised when the given resource is currently not connected.
+    """
+
+class NoSuchUser(WokkelError):
+    """
+    Raised when there is no user with the given name or JID.
+    """

wokkel/generic.py

@@ -11 +11 @@
            'DeferredXmlStreamFactory', 'prepareIDNName',
            'Stanza', 'Request', 'ErrorStanza']
 
+import copy
+
 from zope.interface import implements
 
 from twisted.internet import defer, protocol
@@ -53 +55 @@
     return results and results[0] or None
 
 
+def cloneElement(element):
+    """
+    Make a deep copy of a serialized element.
+
+    The returned element is an orphaned deep copy of the given original.
+
+    @note: Since the reference to the original parent, if any, is gone,
+    inherited attributes like C{xml:lang} are not preserved.
+
+    @type element: L{domish.Element}.
+    """
+    parent = element.parent
+    element.parent = None
+    clone = copy.deepcopy(element)
+    element.parent = parent
+    return clone
+
+
 
 class FallbackHandler(XMPPHandler):
     """

wokkel/iwokkel.py

@@ -996 +996 @@
         """)
 
 
+    interested = Attribute(
+        """
+        This session represents a I{interested resource}, i.e.  the user's
+        roster has been requested and roster pushes will be sent out to
+        this session.
+        """)
+
+
     def loggedIn(realm, mind):
         """
         Called by the realm when login occurs.

wokkel/test/test_generic.py

@@ -24 +24 @@
 
 NS_VERSION = 'jabber:iq:version'
 
+class CloneElementTest(unittest.TestCase):
+    """
+    Tests for L{xmppim.clonePresence}.
+    """
+
+    def test_rootElement(self):
+        """
+        The copied presence stanza is not identical, but renders identically.
+        """
+        parent = object()
+        originalElement = domish.Element((None, 'presence'))
+        originalElement.parent = parent
+        copyElement = generic.cloneElement(originalElement)
+
+        self.assertNotIdentical(copyElement, originalElement)
+        self.assertEqual(copyElement.toXml(), originalElement.toXml())
+        self.assertIdentical(None, copyElement.parent)
+        self.assertIdentical(parent, originalElement.parent)
+
+
+
 class VersionHandlerTest(unittest.TestCase):
     """
     Tests for L{wokkel.generic.VersionHandler}.
@@ -110 +131 @@
         <message type='chat' from='other@example.org' to='user@example.org'/>
         """
 
-        stanza = generic.Stanza.fromElement(generic.parseXml(xml))
+        element = generic.parseXml(xml)
+        stanza = generic.Stanza.fromElement(element)
         self.assertEqual('chat', stanza.stanzaType)
         self.assertEqual(JID('other@example.org'), stanza.sender)
         self.assertEqual(JID('user@example.org'), stanza.recipient)
+        self.assertIdentical(element, stanza.element)
+
+
+    def test_fromElementStanzaKind(self):
+        """
+        The stanza kind is also recorded in the stanza.
+        """
+        xml = """<presence/>"""
+        stanza = generic.Stanza.fromElement(generic.parseXml(xml))
+        self.assertEqual(u'presence', stanza.stanzaKind)
 
 
     def test_fromElementChildParser(self):

wokkel/test/test_xmppim.py

@@ -5 +5 @@
 Tests for L{wokkel.xmppim}.
 """
 
+from zope.interface import verify
+
+from twisted.cred import checkers, error as ecred
+from twisted.cred.portal import IRealm
 from twisted.internet import defer
 from twisted.trial import unittest
 from twisted.words.protocols.jabber import error
@@ -12 +16 @@
 from twisted.words.protocols.jabber.xmlstream import toResponse
 from twisted.words.xish import domish, utility
 
-from wokkel import xmppim
+from wokkel import ewokkel, component, xmppim
 from wokkel.generic import ErrorStanza, parseXml
 from wokkel.test.helpers import TestableRequestHandlerMixin, XmlStreamStub
+from wokkel.subprotocols import IQHandlerMixin
 
 NS_XML = 'http://www.w3.org/XML/1998/namespace'
 NS_ROSTER = 'jabber:iq:roster'
@@ -99 +104 @@
         self.assertEquals(50, presence.priority)
 
 
+
 class PresenceProtocolTest(unittest.TestCase):
     """
     Tests for L{xmppim.PresenceProtocol}
@@ -1418 +1424 @@
 
 
 
+class InMemoryRosterTest(unittest.TestCase):
+    """
+    Tests for L{xmppim.InMemoryRoster}.
+    """
+
+    def setUp(self):
+        contacts = [
+            xmppim.RosterItem(JID('contact1@example.org'),
+                              subscriptionFrom=True,
+                              subscriptionTo=False),
+            xmppim.RosterItem(JID('contact2@example.org'),
+                              subscriptionFrom=False,
+                              subscriptionTo=True),
+            ]
+        self.roster = xmppim.InMemoryRoster(contacts)
+
+
+    def test_getSubscribers(self):
+        def gotSubscribers(subscribers):
+            subscribers = list(subscribers)
+            self.assertIn(JID('contact1@example.org'), subscribers)
+            self.assertNotIn(JID('contact2@example.org'), subscribers)
+
+
+        d = self.roster.getSubscribers()
+        d.addCallback(gotSubscribers)
+        return d
+
+
+    def test_getSubscriptions(self):
+        def gotSubscriptions(subscriptions):
+            subscriptions = list(subscriptions)
+            self.assertNotIn(JID('contact1@example.org'), subscriptions)
+            self.assertIn(JID('contact2@example.org'), subscriptions)
+
+        d = self.roster.getSubscriptions()
+        d.addCallback(gotSubscriptions)
+        return d
+
+
+class UserRosterProtocolTest(unittest.TestCase):
+    """
+    Tests for L{xmppim.UserRosterProtocol}.
+    """
+
+    def setUp(self):
+        self.stub = XmlStreamStub()
+        self.service = xmppim.UserRosterProtocol()
+        self.service.makeConnection(self.stub.xmlstream)
+
+        entity = JID(u'user@example.org')
+        user = xmppim.User(entity)
+
+        contact = xmppim.RosterItem(JID(u'contact@example.org'))
+        user.roster = xmppim.InMemoryRoster([contact])
+
+        self.session = xmppim.UserSession(user)
+        self.stub.xmlstream.avatar = self.session
+
+
+    def test_getRosterReceived(self):
+        """
+        The returned roster is gotten from the session user.
+        """
+        def gotRoster(result):
+            self.assertIn(JID(u'contact@example.org'), result)
+
+        request = xmppim.RosterRequest()
+        d = self.service.getRosterReceived(request)
+        d.addCallback(gotRoster)
+        return d
+
+
+    def test_getRosterReceivedInterested(self):
+        """
+        Requesting the roster marks the session as interested in roster pushes.
+        """
+        def gotRoster(result):
+            self.assertTrue(self.session.interested)
+
+        request = xmppim.RosterRequest()
+        d = self.service.getRosterReceived(request)
+        d.addCallback(gotRoster)
+        return d
+
+
+    def test_handleRequestLocal(self):
+        """
+        Handle requests without recipient (= local server).
+        """
+        called = []
+        request = xmppim.RosterRequest(recipient=None)
+        self.patch(IQHandlerMixin, 'handleRequest', called.append)
+        self.service.handleRequest(request.toElement())
+        self.assertTrue(called)
+
+
+    def test_handleRequestOther(self):
+        """
+        If the request has a non-empty recipient, ignore this request.
+        """
+        request = xmppim.RosterRequest(recipient=JID('other.example.org'))
+        self.assertFalse(self.service.handleRequest(request.toElement()))
+
+
+
 class MessageTest(unittest.TestCase):
     """
     Tests for L{xmppim.Message}.
@@ -1625 +1737 @@
                          "was deprecated in Wokkel 0.8.0; "
                          "please use MessageProtocol.messageReceived instead.",
                          warnings[0]['message'])
+
+
+
+class UserSessionTest(unittest.TestCase):
+    """
+    Tests for L{xmppim.UserSession}.
+    """
+
+    def setUp(self):
+        self.session = xmppim.UserSession(None)
+
+
+    def test_interface(self):
+        """
+        UserSession implements IUserSession.
+        """
+        verify.verifyObject(xmppim.IUserSession, self.session)
+
+
+class UserTest(unittest.TestCase):
+    """
+    Tests for L{xmppim.User}.
+    """
+
+    def setUp(self):
+        self.user = xmppim.User(JID('user@example.org'),
+                                xmppim.InMemoryRoster([]))
+        self.session = xmppim.UserSession(self.user)
+        self.session.bindResource('Home')
+
+
+    @defer.inlineCallbacks
+    def test_getPresences(self):
+        """
+        A contact with a subscription to the user gets its presence.
+        """
+        contact = xmppim.RosterItem(JID('contact@example.org'),
+                                     subscriptionFrom=True)
+        self.user.roster.roster[contact.entity] = contact
+
+        presence = xmppim.AvailabilityPresence()
+        self.session.presence = presence
+
+        presences = yield self.user.getPresences(JID('contact@example.org'))
+        self.assertEqual([presence], list(presences))
+
+        defer.returnValue(None)
+
+
+    def test_getPresencesNoSubscription(self):
+        """
+        A contact without a subscription raises NotSubscribed.
+        """
+        contact = xmppim.RosterItem(JID('contact@example.org'),
+                                     subscriptionFrom=False)
+        self.user.roster.roster[contact.entity] = contact
+
+        presence = xmppim.AvailabilityPresence()
+        self.session.presence = presence
+
+        d = self.user.getPresences(JID('contact@example.org'))
+        self.assertFailure(d, ewokkel.NotSubscribed)
+        return d
+
+
+    def test_getPresencesUnknown(self):
+        """
+        A contact with a subscription to the user gets its presence.
+        """
+        presence = xmppim.AvailabilityPresence()
+        self.session.presence = presence
+        d = self.user.getPresences(JID('unknown@example.org'))
+        self.assertFailure(d, ewokkel.NoSuchContact)
+        return d
+
+
+class AnonymousRealmTest(unittest.TestCase):
+    """
+    Tests for L{xmppim.AnonymousRealm}.
+    """
+
+    def setUp(self):
+        self.realm = xmppim.AnonymousRealm('example.org')
+
+
+    def test_interface(self):
+        """
+        AnonymousRealm implements IRealm.
+        """
+        verify.verifyObject(IRealm, self.realm)
+
+
+    def test_requestAvatarAnonymous(self):
+        """
+        An anonymous avatar ID yields a generated JID.
+        """
+        def gotAvatar(result):
+            def gotUser(user):
+                self.assertIdentical(user, avatar.user)
+
+            iface, avatar, logout = result
+            self.assertNotIdentical(None, avatar.user)
+            self.assertNotIdentical(None, avatar.user.entity)
+            self.assertNotIdentical(None, avatar.user.entity.host)
+            self.assertEqual(u'example.org', avatar.user.entity.host)
+
+            d = self.realm.lookupUser(avatar.user.entity)
+            d.addCallback(gotUser)
+            return d
+
+        avatarID = checkers.ANONYMOUS
+        d = self.realm.requestAvatar(avatarID, None, xmppim.IUserSession)
+        d.addCallback(gotAvatar)
+        return d
+
+
+    def test_requestAvatarAnonymousDifferent(self):
+        """
+        Requesting two anonymous avatar IDs yields different JIDs.
+        """
+        def gotAvatar1(result):
+            iface, avatar1, logout = result
+
+            def gotAvatar2(result):
+                iface, avatar2, logout = result
+                self.assertNotIdentical(avatar1, avatar2)
+                self.assertNotIdentical(avatar1.user, avatar2.user)
+                self.assertNotEqual(avatar1.user.entity,
+                                    avatar2.user.entity)
+
+            d = self.realm.requestAvatar(avatarID, None, xmppim.IUserSession)
+            d.addCallback(gotAvatar2)
+            return d
+
+        avatarID = checkers.ANONYMOUS
+        d = self.realm.requestAvatar(avatarID, None, xmppim.IUserSession)
+        d.addCallback(gotAvatar1)
+        return d
+
+
+    def test_logout(self):
+        """
+        When the logout function is called, the user is removed from the realm.
+        """
+        def gotAvatar(result):
+            iface, avatar, logout = result
+
+            logout()
+
+            entity = avatar.user.entity
+            d = self.realm.lookupUser(entity)
+            self.assertFailure(d, ewokkel.NoSuchUser)
+            return d
+
+        avatarID = checkers.ANONYMOUS
+        d = self.realm.requestAvatar(avatarID, None, xmppim.IUserSession)
+        d.addCallback(gotAvatar)
+        return d
+
+
+
+class StaticRealmTest(unittest.TestCase):
+    """
+    Tests for L{xmppim.StaticRealmTest}.
+    """
+
+    def setUp(self):
+        entity = JID(u'\u00e9lise@example.org')
+        self.user = xmppim.User(entity)
+        users = {entity: self.user}
+        self.realm = xmppim.StaticRealm('example.org', users)
+
+
+    def test_interface(self):
+        """
+        StaticRealm implements IRealm.
+        """
+        verify.verifyObject(IRealm, self.realm)
+
+
+    def test_requestAvatar(self):
+        """
+        A UserSession is initialized and returned from requestAvatar.
+        """
+        def gotAvatar(result):
+            iface, avatar, logout = result
+            self.assertIdentical(self.user, avatar.user)
+
+        avatarID = u'\u00e9lise'.encode('utf-8')
+        d = self.realm.requestAvatar(avatarID, None, xmppim.IUserSession)
+        d.addCallback(gotAvatar)
+        return d
+
+
+    def test_requestAvatarUnknown(self):
+        """
+        A UserSession is initialized and returned from requestAvatar.
+        """
+        avatarID = u'nobody'.encode('utf-8')
+        d = self.realm.requestAvatar(avatarID, None, xmppim.IUserSession)
+        self.assertFailure(d, ecred.LoginDenied)
+        return d
+
+
+
+class TestRouter(component.Router):
+    """
+    Router that only records incoming traffic for testing.
+    """
+
+    def __init__(self):
+        super(TestRouter, self).__init__()
+        self.output = []
+
+
+    def route(self, stanza):
+        """
+        All routed stanzas are recorded.
+        """
+        self.output.append(stanza)
+
+
+
+class SessionManagerTest(unittest.TestCase):
+    """
+    Tests for L{xmppim.SessionManager}.
+    """
+
+    def setUp(self):
+        self.router = TestRouter()
+        self.sessionManager = xmppim.SessionManager(self.router,
+                                                    u'example.org')
+        self.sessionManager.startService()
+
+        self.input = []
+        def onElement(element):
+            self.input.append(element)
+
+        self.sessionManager.xmlstream.addObserver('/*', onElement)
+
+
+    def test_routeOrDeliverLocal(self):
+        """
+        Stanzas for local domains are reinjected in to the XML stream.
+        """
+        element = parseXml("""<presence from='test@example.org'
+                                        to='other@example.org'/>""")
+        self.sessionManager.xmlstream.send(element)
+
+        self.assertEqual(1, len(self.input))
+        self.assertEqual(0, len(self.router.output))
+
+
+    def test_routeOrDeliverRemote(self):
+        """
+        Stanzas for other domains are sent to the router.
+        """
+        element = parseXml("""<presence from='test@example.org'
+                                        to='other@example.com'/>""")
+        self.sessionManager.xmlstream.send(element)
+
+        self.assertEqual(0, len(self.input))
+        self.assertEqual(1, len(self.router.output))
+
+
+    def test_probePresence(self):
+        """
+        Presence probes are sent to contacts with a subscription.
+        """
+        def cb(result):
+            self.assertEqual(1, len(self.router.output))
+            element = self.router.output[-1]
+            self.assertEqual(u'presence',
+                             element.name)
+            self.assertEqual(u'probe',
+                             element.getAttribute(u'type'))
+            self.assertEqual(u'contact2@example.com',
+                             element.getAttribute(u'to'))
+
+        contacts = [
+            xmppim.RosterItem(JID(u'contact1@example.com'),
+                              subscriptionFrom=True,
+                              subscriptionTo=False),
+            xmppim.RosterItem(JID(u'contact2@example.com'),
+                              subscriptionFrom=False,
+                              subscriptionTo=True),
+            ]
+        roster = xmppim.InMemoryRoster(contacts)
+        entity = JID(u'user@example.org')
+        user = xmppim.User(entity, roster)
+
+        d = self.sessionManager.probePresence(user)
+        d.addCallback(cb)
+        return d

wokkel/xmppim.py

@@ -12 +12 @@
 
 import warnings
 
+from zope.interface import implementer
+
+from twisted.cred import error as ecred, portal
 from twisted.internet import defer
+from twisted.python import log, randbytes
 from twisted.words.protocols.jabber import error
 from twisted.words.protocols.jabber.jid import JID
 from twisted.words.xish import domish
 
-from wokkel.generic import ErrorStanza, Stanza, Request
+from wokkel.component import InternalComponent
+from wokkel.ewokkel import NoSuchContact
+from wokkel.ewokkel import NotSubscribed, NoSuchResource, NoSuchUser
+from wokkel.iwokkel import IUserSession
+from wokkel.generic import ErrorStanza, Stanza, Request, cloneElement
 from wokkel.subprotocols import IQHandlerMixin
 from wokkel.subprotocols import XMPPHandler
 from wokkel.subprotocols import asyncObserver
@@ -1062 +1070 @@
 
 
 
+class InMemoryRoster(object):
+
+    def __init__(self, items):
+        self.roster = Roster(((item.entity, item) for item in items))
+
+
+    def getRoster(self, version=None):
+        return defer.succeed(self.roster)
+
+
+    def getSubscribers(self):
+        subscribers = (entity for entity, item in self.roster.iteritems()
+                              if item.subscriptionFrom)
+        return defer.succeed(subscribers)
+
+
+    def getSubscriptions(self):
+        subscriptions = (entity for entity, item in self.roster.iteritems()
+                                if item.subscriptionTo)
+        return defer.succeed(subscriptions)
+
+
+    def getContact(self, entity):
+        try:
+            return defer.succeed(self.roster[entity])
+        except KeyError:
+            return defer.fail(NoSuchContact())
+
+
+
+
+
+class UserRosterProtocol(RosterServerProtocol):
+    """
+    Roster protocol handler for client connections.
+
+    This protocol is meant to be used with
+    L{wokkel.client.XMPPC2SServerFactory} to interact with L{UserSession} and
+    L{User}.
+    """
+
+    def handleRequest(self, iq):
+        """
+        Ignore roster requests for non-empty recipients.
+        """
+        if iq.getAttribute('to'):
+            return False
+        else:
+            return super(UserRosterProtocol, self).handleRequest(iq)
+
+
+    def getRosterReceived(self, request):
+        """
+        Return the roster of the user associated with this session.
+        """
+        session = self.xmlstream.avatar
+        session.interested = True
+        return session.user.roster.getRoster()
+
+
+
 class Message(Stanza):
     """
     A message stanza.
@@ -1175 +1244 @@
 
             self.onMessage(message.element)
             return getattr(message, "handled", False)
+
+
+
+@implementer(IUserSession)
+class UserSession(object):
+    """
+    An XMPP user session.
+
+    This represents the session for a connected client after authenticating
+    as L{user}.
+
+    @ivar user: The authenticated user for this session.
+    @type user: L{User}
+
+    @ivar mind: The protocol instance for this session.
+    @type mind: L{xmlstream.Xmlstream}
+
+    @ivar entity: The full JID of the entity after a resource has been
+       bound.
+    @type entity: L{JID}
+
+    @ivar connected: Flag that is C{True} while a resource is bound.
+    @type connected: L{boolean}
+
+    @ivar interested: Flag to record that the roster has been requested. When
+        L{True}, this session will receive roster pushes.
+    @type interested: L{boolean}
+
+    @ivar presence: Last broadcast presence from the client for this session.
+    @type presence: L{AvailabilityPresence}
+    """
+
+    user = None
+    mind = None
+    entity = None
+    connected = False
+    interested = False
+    presence = None
+
+    def __init__(self, user):
+        self.user = user
+
+
+    def loggedIn(self, realm, mind):
+        self.mind = mind
+        self.realm = realm
+
+
+    def bindResource(self, resource):
+        def cb(entity):
+            self.entity = entity
+            self.connected = True
+            return entity
+
+        d = self.user.bindResource(self, resource)
+        d.addCallback(cb)
+        return d
+
+
+    def logout(self):
+        self.connected = False
+
+        if self.entity:
+            self.user.unbindResource(self.entity.resource)
+
+
+    def send(self, element):
+        """
+        Called when the client sends a stanza.
+        """
+        self.realm.server.routeOrDeliver(element)
+
+
+    def receive(self, element, recipient=None):
+        """
+        Deliver a stanza to the client.
+        """
+        self.mind.send(element)
+
+
+    def broadcastPresence(self, presence):
+        doProbe = presence.available and not self.presence
+
+        if presence.available:
+            self.presence = presence
+        else:
+            self.presence = None
+
+        self.user.broadcastPresence(presence)
+
+        if doProbe:
+            self.user.probePresence(self)
+
+
+
+class User(object):
+    """
+    An XMPP user account.
+
+    @ivar entity: The JID of the user.
+    @type entity: L{JID}
+
+    @ivar roster: The user's roster.
+
+    @ivar sessions: The currently connected sessions for this user, indexed by
+        resource.
+    @type sessions: L{dict}
+
+    @ivar probeSent: Flag that is C{True} if presence probes have been sent
+        out for this user. This is only done on the initial presence broadcast
+        of the first available resource. Subsequent resources will get the
+        presences stored in L{contactPresences}.
+    @type probeSent: L{boolean}
+
+    @ivar contactPresences: Cached presences of contacts as a mapping of L{JID}
+        to L{AvailabilityPresence}.
+    @type contactPresences: L{dict}
+
+    @ivar realm: The realm that provided this user object.
+    @type realm: L{IRealm} provider
+    """
+
+    realm = None
+
+    def __init__(self, entity, roster=None):
+        self.entity = entity
+        self.roster = roster
+        self.sessions = {}
+        self.probeSent = False
+        self.contactPresences = {}
+
+
+    def bindResource(self, session, resource):
+        if not self.sessions:
+            self.probeSent = False
+
+        if resource is None:
+            resource = randbytes.secureRandom(8).encode('hex')
+        elif resource in self.sessions:
+            resource = resource + ' ' + randbytes.secureRandom(8).encode('hex')
+
+        entity = JID(tuple=(self.entity.user, self.entity.host, resource))
+        self.sessions[resource] = session
+
+        return defer.succeed(entity)
+
+
+    def unbindResource(self, resource):
+        del self.sessions[resource]
+        return defer.succeed(None)
+
+
+    def deliverIQ(self, stanza):
+        try:
+            session = self.sessions[stanza.recipient.resource]
+        except KeyError:
+            raise NoSuchResource()
+
+        session.receive(stanza.element)
+
+
+    def deliverMessage(self, stanza):
+        if stanza.recipient.resource:
+            try:
+                session = self.sessions[stanza.recipient.resource]
+            except KeyError:
+                if stanza.stanzaType in ('normal', 'chat', 'headline'):
+                    self.deliverMessageAnyResource(stanza)
+                else:
+                    raise NoSuchResource()
+            else:
+                session.receive(stanza.element)
+        else:
+            if stanza.stanzaType == 'groupchat':
+                raise NotImplementedError("Groupchat message to the bare JID")
+            else:
+                self.deliverMessageAnyResource(stanza)
+
+
+    def deliverMessageAnyResource(self, stanza):
+        if stanza.stanzaType == 'headline':
+            recipients = set()
+            for resource, session in self.sessions.iteritems():
+                if session.presence.priority >= 0:
+                    recipients.add(resource)
+        elif stanza.stanzaType in ('chat', 'normal'):
+            priorities = {}
+            for resource, session in self.sessions.iteritems():
+                if not session.presence or not session.presence.available:
+                    continue
+                priority = session.presence.priority
+                if priority >= 0:
+                    priorities.setdefault(priority, set()).add(resource)
+            if priorities:
+                maxPriority = max(priorities.keys())
+                recipients = priorities[maxPriority]
+            else:
+                # No available resource, offline storage not supported
+                raise NotImplementedError("Offline storage is not supported")
+        else:
+            recipients = set()
+
+        if recipients:
+            for resource in recipients:
+                session = self.sessions[resource]
+                session.receive(stanza.element)
+        else:
+            # silently discard
+            log.msg("Discarding message to %r" % stanza.recipient)
+
+
+    def deliverPresence(self, stanza):
+        if not stanza.recipient.resource:
+            # TODO: record presence in self.contactPresences for future probes
+
+            for session in self.sessions.itervalues():
+                if session.presence:
+                    session.receive(stanza.element)
+
+
+    def probePresence(self, session):
+        """
+        Probe presences for this user.
+
+        If this is the first session requesting presence probes, they are
+        sent out to the contacts via the realm. After that, the last received
+        presences are sent back to the session directly.
+        """
+        if not self.probeSent:
+            # send out probes
+            self.contactPresences = {}
+            self.realm.server.probePresence(self)
+            self.probeSent = True
+        else:
+            # deliver known contact presences
+            for presence in self.contactPresences.itervalues():
+                session.receive(presence.element)
+
+
+    @defer.inlineCallbacks
+    def broadcastPresence(self, presence):
+        """
+        Broadcast presence to all subscribed contacts and myself.
+        """
+        # TODO: set probeSent to False when last session becomes unavailable
+        # TODO: save last unavailable presence?
+        subscribers = yield self.roster.getSubscribers()
+        self.realm.server.multicast(presence, subscribers)
+
+
+    @defer.inlineCallbacks
+    def getPresences(self, entity):
+        """
+        Get presences on behalf of a contact.
+
+        @param entity: The contact requesting that initiated a presence probe.
+        @type entity: L{JID}
+
+        @return: Deferred that fires with an iterable of
+            L{AvailabilityPresence}.
+        @rtype: L{defer.Deferred}
+
+        @raise NotSubscribed: If the contact does not have a presence
+            subscription from this user.
+        @raise NoSuchContact: If the requestor is not a contact.
+        """
+        bareEntity = entity.userhostJID()
+        item = yield self.roster.getContact(bareEntity)
+
+        if not item.subscriptionFrom:
+            raise NotSubscribed()
+
+        presences = (session.presence for session in self.sessions.itervalues()
+                                      if session.presence)
+        defer.returnValue(presences)
+        # TODO: send last unavailable or unavailable presence?
+
+
+
+@implementer(portal.IRealm)
+class BaseRealm(object):
+    server = None
+
+    def __init__(self, domain):
+        self.domain = domain
+
+
+    def lookupUser(self, entity):
+        raise NotImplementedError()
+
+
+    def createUser(self, entity):
+        raise NotImplementedError()
+
+
+    def getUser(self, entity):
+        def trapNoSuchUser(failure):
+            failure.trap(NoSuchUser)
+            return self.createUser(entity)
+
+        d = self.lookupUser(entity)
+        d.addErrback(trapNoSuchUser)
+        return d
+
+
+    def logoutFactory(self, session):
+        return session.logout
+
+
+    def entityFromAvatarID(self, avatarId):
+        localpart = avatarId.decode('utf-8')
+        return JID(tuple=(localpart, self.domain, None))
+
+
+    def requestAvatar(self, avatarId, mind, *interfaces):
+        if IUserSession not in interfaces:
+            raise NotImplementedError(self, interfaces)
+
+        entity = self.entityFromAvatarID(avatarId)
+
+        def gotUser(user):
+            session = UserSession(user)
+            session.loggedIn(self, mind)
+            return IUserSession, session, self.logoutFactory(session)
+
+        d = self.getUser(entity)
+        d.addCallback(gotUser)
+        return d
+
+
+
+class AnonymousRealm(BaseRealm):
+
+    def __init__(self, domain):
+        BaseRealm.__init__(self, domain)
+        self.users = {}
+
+
+    def entityFromAvatarID(self, avatarId):
+        localpart = randbytes.secureRandom(8).encode('hex')
+        return JID(tuple=(localpart, self.domain, None))
+
+
+    def lookupUser(self, entity):
+        try:
+            user = self.users[entity]
+        except KeyError:
+            return defer.fail(NoSuchUser(entity))
+        return defer.succeed(user)
+
+
+    def createUser(self, entity):
+        user = User(entity, InMemoryRoster([]))
+        user.realm = self
+        self.users[entity] = user
+        return defer.succeed(user)
+
+
+    def logoutFactory(self, session):
+        def logout():
+            session.logout()
+            del self.users[session.user.entity]
+        return logout
+
+
+
+class StaticRealm(BaseRealm):
+
+    def __init__(self, domain, users):
+        BaseRealm.__init__(self, domain)
+        for user in users.itervalues():
+            user.realm = self
+        self.users = users
+
+
+    def lookupUser(self, entity):
+        try:
+            user = self.users[entity]
+        except KeyError:
+            return defer.fail(NoSuchUser(entity))
+        return defer.succeed(user)
+
+
+    def createUser(self, entity):
+        return defer.fail(ecred.LoginDenied("Can't create a new user"))
+
+
+
+class SessionManager(InternalComponent):
+    """
+    Session Manager.
+
+    @ivar xmlstream: XML Stream to inject incoming stanzas from client
+        connections into. Stanzas where the C{'to'} attribute is not set
+        or is directed at the local domain are injected as if received on
+        the XML Stream (using C{dispatch}), other stanzas are injected as if
+        they were sent from the XML Stream (using C{send}).
+    """
+
+    def startService(self):
+        InternalComponent.startService(self)
+        self.xmlstream.send = self.routeOrDeliver
+
+
+    def routeOrDeliver(self, element):
+        """
+        Deliver a stanza locally or pass on for routing.
+        """
+        if (JID(element['to']).host in self.domains):
+            # This stanza is for local delivery
+            log.msg("Delivering locally: %r" % element.toXml())
+            self._pipe.source.dispatch(element)
+        else:
+            # This stanza is for remote routing
+            log.msg("Routing remotely: %r" % element.toXml())
+            self._pipe.sink.dispatch(element)
+
+
+    def multicast(self, stanza, recipients):
+        """
+
+        @param stanza: The stanza to send. Its C{element} attribute should
+            already be set.
+        @type stanza: L{wokkel.generic.Stanza}.
+
+        @type recipients: iterable of L{JID}.
+        """
+        if not stanza.element:
+            stanza.toElement()
+
+        for recipient in recipients:
+            clone = cloneElement(stanza.element)
+            clone['to'] = recipient.full()
+            clone.handled = False
+            self.routeOrDeliver(clone)
+
+
+    @defer.inlineCallbacks
+    def probePresence(self, user):
+        """
+        Request the presences of all contacts the user has a subscription to.
+
+        This will send out presence probe stanzas, even to local contacts.
+        """
+        subscriptions = yield user.roster.getSubscriptions()
+        for entity in subscriptions:
+            presence = ProbePresence(recipient=entity,
+                                     sender=user.entity)
+            self.routeOrDeliver(presence.toElement())