Python moderator work and general clean up #203

servers/java/coweb-server/src/main/java/org/coweb/SessionHandler.java

@@ -102,18 +102,23 @@ public void configureChannel(ConfigurableServerChannel channel) {
 		sync = server.getChannel(this.syncEngineChannel);
 		sync.addListener(this);
 
-		this.sessionModerator = SessionModerator.getInstance(this, (String)config.get("sessionModerator"), confKey);
+		this.sessionModerator = SessionModerator.getInstance(this,
+                (String)config.get("sessionModerator"), confKey);
 		if (null == this.sessionModerator) {
 			config.put("moderatorIsUpdater", false);
-			/* Perhaps config.get("sessionModerator") had an exception or didn't exist,
-			   so either we try to create default implementation of moderator, or throw an exception. */
-			log.severe("SessionModerator.getInstance(" + config.get("sessionModerator") +
-					") failed, reverting to trying to create default implementation.");
-			this.sessionModerator = SessionModerator.getInstance(this, null, confKey);
+			/* Perhaps config.get("sessionModerator") had an exception or
+             * didn't exist, so either we try to create default implementation
+             * of moderator, or throw an exception. */
+			log.severe("SessionModerator.getInstance(" +
+                    config.get("sessionModerator") + ") failed, reverting to " +
+                    "trying to create default implementation.");
+			this.sessionModerator = SessionModerator.getInstance(this,
+                    null, confKey);
 			if (null == this.sessionModerator) {
 				throw new CowebException("Create SessionModerator", "");
 			}
-			log.severe("SessionModerator created default implementation, but moderator can no longer be updater.");
+			log.severe("SessionModerator created default implementation, but " +
+                    "moderator can no longer be updater.");
 		}
 
 		// create the late join handler. clients will be updaters by default.

servers/python/coweb/container.py

@@ -62,7 +62,7 @@ def __init__(self, options):
         # allow easy override of default settings without busting manager
         # creation
         self.on_configure()
-        
+
         # adjust all paths to make them absolute relative to container loc now
         self.httpStaticPath = self.get_absolute_path(self.httpStaticPath)
         for i, path in enumerate(self.cowebBotLocalPaths):

servers/python/coweb/moderator.py

@@ -7,25 +7,25 @@
 class SessionModerator:
     def __init__(self):
         print("hello from moderator")
-    def canClientJoinSession(self):
+    def canClientJoinSession(self, client, message):
         raise NotImplementedError()
-    def canClientMakeServiceRequest(self):
+    def canClientMakeServiceRequest(self, svcName, client, botMessage):
         raise NotImplementedError()
-    def canClientSubscribeService(self):
+    def getLateJoinState(self):
+        raise NotImplementedError()
+    def canClientSubscribeService(self, svcName, client, message):
         raise NotImplementedError()
-    def onClientJoinSession(self):
+    def onClientJoinSession(self, client, message):
         raise NotImplementedError()
-    def onClientLeaveSession(self):
+    def onClientLeaveSession(self, client):
         raise NotImplementedError()
-    def onServiceResponse(self):
+    def onServiceResponse(self, svcName, data, error, isPublic):
         raise NotImplementedError()
     def onSessionEnd(self):
         raise NotImplementedError()
     def onSessionReady(self):
         raise NotImplementedError()
-    def onSync(self):
-        raise NotImplementedError()
-    def getLateJoinState(self):
+    def onSync(self, client, data):
         raise NotImplementedError()
 
     @staticmethod
@@ -38,25 +38,25 @@ def getInstance(klass, key):
         return moderator
 
 class DefaultSessionModerator(SessionModerator):
-    def canClientJoinSession(self):
+    def canClientJoinSession(self, client, message):
         return True
-    def canClientMakeServiceRequest(self):
+    def canClientMakeServiceRequest(self, svcName, client, botMessage):
         return True
-    def canClientSubscribeService(self):
+    def getLateJoinState(self):
+        print("in late get state")
+        return {}
+    def canClientSubscribeService(self, svcName, client, message):
         return True
-    def onClientJoinSession(self):
+    def onClientJoinSession(self, client, message):
         pass
-    def onClientLeaveSession(self):
+    def onClientLeaveSession(self, client):
         pass
-    def onServiceResponse(self):
+    def onServiceResponse(self, svcName, data, error, isPublic):
         pass
     def onSessionEnd(self):
         pass
     def onSessionReady(self):
         pass
-    def onSync(self):
+    def onSync(self, client, data):
         pass
-    def getLateJoinState(self):
-        print("in late get state")
-        return {}
 

servers/python/coweb/session/__init__.py

@@ -11,3 +11,4 @@ def create_session(collab, *args, **kwargs):
     '''Builds a cooperative or non-cooperative session.'''
     cls = CollabSession if collab else Session
     return cls(*args, **kwargs)
+

servers/python/coweb/session/collab.py

@@ -12,14 +12,10 @@
 import re
 # coweb
 from . import session
-from .. import OEHandler
-from ..moderator import SessionModerator
 from coweb.session.late_join_handler import LateJoinHandler
 from coweb.session.moderator_late_join_handler import ModeratorLateJoinHandler
 
-OEHandler = OEHandler.OEHandler
-session_sync_regex = re.compile("/session/([A-z0-9]+)/sync(.*)");
-
+getServiceNameFromChannel = session.getServiceNameFromChannel
 log = logging.getLogger('coweb.session')
 
 class CollabSession(session.Session):
@@ -31,16 +27,6 @@ def __init__(self, *args, **kwargs):
         self.collab = True
         self._connectionClass = CollabSessionConnection
 
-        # operation total order
-        self._opOrder = -1
-
-        # TODO make this optional
-        self._opengine = OEHandler(self, 0)
-
-        # Moderator?
-        self._moderator = SessionModerator.getInstance(
-                self._container.moderatorClass, self.key)
-
         if self._container.moderatorIsUpdater:
             self._lateJoinHandler = ModeratorLateJoinHandler(self)
         else:
@@ -51,14 +37,6 @@ def get_order(self):
         self._opOrder += 1
         return self._opOrder
 
-    def sync_for_service(self, client, req):
-        '''
-        Forwards sync events to service bridge if at least one service has 
-        permission for sync events enabled.
-        '''
-        if self._services.needsSync:
-            self._services.on_user_sync(client, req)
-
     def on_purging_client(self, cid, client):
         '''Override to remove updater and end a session when none left.'''
         # notify all bots of unsubscribing user
@@ -75,66 +53,53 @@ class CollabSessionConnection(session.SessionConnection):
     '''Connection for collaborative sessions.'''
     def on_publish(self, cl, req, res):
         '''Override to handle updater and sync logic.'''
+        manager = self._manager
         channel = req['channel']
         if channel == '/service/session/updater':
             # handle updater response
             data = req.get('data', None)
             try:
-                self._manager.onUpdaterSendState(cl, data) 
+                manager.onUpdaterSendState(cl, data) 
             except Exception:
                 # bad updater, disconnect and assign a new one
-                self._manager.remove_bad_client(cl)
+                manager.remove_bad_client(cl)
                 return
-        else:
-            matches = session_sync_regex.match(channel);
-            if matches:
-                # handle sync events
-                try:
-                    # put siteId on message
-                    req['data']['siteId'] = cl.siteId
-                    self._manager._lateJoinHandler.ensureUpdater(cl)
-                except (KeyError, AttributeError):
-                    # not allowed to publish syncs, disconnect
-                    self._manager.remove_bad_client(cl)
-                    return
-                # last state no longer valid
-                self._manager._lateJoinHandler.clearLastState()
-                if '/app' == matches.group(2):
-                    # App sync.
-                    # Put total order sequence number on message
-                    req['data']['order'] = self._manager.get_order()
-                    # let manager deal with sync if forwarding it to services
-                    self._manager.sync_for_service(cl, req)
-                    if self._manager._opengine:
-                        # Push sync to op engine.
-                        self._manager._opengine.syncInbound(req['data'])
-                elif '/engine' == matches.group(2):
-                    # Engine sync.
-                    if self._manager._opengine:
-                        self._manager._opengine.engineSyncInbound(req['data'])
+        elif channel == manager.syncAppChannel:
+            manager._lateJoinHandler.clearLastState()
+            req['data']['siteId'] = cl.siteId
+            req['data']['order'] = manager.get_order()
+            # Push sync to op engine.
+            sync = manager._opengine.syncInbound(req['data'])
+            if sync:
+                manager._moderator.onSync(cl, sync)
+        elif channel == manager.engineSyncChannel:
+            manager._opengine.engineSyncInbound(req['data'])
         # delegate all other handling to base class
         super(CollabSessionConnection, self).on_publish(cl, req, res)
 
     def on_subscribe(self, cl, req, res):
         '''Override to handle late-joiner logic.'''
+        manager = self._manager
         sub = req['subscription']
-        didSub = True        
-        if sub.startswith('/service/bot'):
-            # handle private subscribe to bot
-            didSub = self._manager.subscribe_to_service(cl, req, res, False)
-        elif sub.startswith('/bot'):
-            # public subscribe to bot
-            didSub = self._manager.subscribe_to_service(cl, req, res, True)
+        didSub = True
+        pub = sub.startswith('/bot')
+        if pub or sub.startswith('/service/bot'):
+            # public subscribe to bot (/bot)
+            # handle private subscribe to bot (/service/bot)
+            svcName = getServiceNameFromChannel(sub, pub)
+            if manager._moderator.canClientSubscribeService(svcName, cl, req):
+                didSub = manager.subscribe_to_service(cl, req, res, pub)
         elif sub == '/service/session/join/*':
-            ext = req['ext']
-            coweb = ext['coweb']
-            updaterType = coweb['updaterType']
-            cl.updaterType = updaterType
-            self._manager._lateJoinHandler.onClientJoin(cl) 
-            didSub = True
+            if manager._moderator.canClientJoinSession(cl, req):
+                ext = req['ext']
+                coweb = ext['coweb']
+                updaterType = coweb['updaterType']
+                cl.updaterType = updaterType
+                if manager._lateJoinHandler.onClientJoin(cl):
+                    manager._moderator.onSessionReady()
         elif sub == '/service/session/updater':
-            self._manager._lateJoinHandler.addUpdater(cl)
-            didSub = True
+            manager._lateJoinHandler.addUpdater(cl)
+            manager._moderator.onClientJoinSession(cl, req)
         if didSub:
             # don't run default handling if sub failed
             super(CollabSessionConnection, self).on_subscribe(cl, req, res)

servers/python/coweb/session/late_join_handler.py

@@ -79,11 +79,6 @@ def addUpdater(self, client, notify=True):
         if notify:
             self._sendRosterAvailable(client)
 
-    def ensureUpdater(self, client):
-        '''Ensures a client is an updater. Exception if not.'''
-        # just try to access to ensure
-        self._updaters[client.clientId]
-
     def removeUpdater(self, client):
         '''Removes a client as an updater.'''
         # remove this client from the assigned siteids
@@ -187,6 +182,7 @@ def _sendRosterUnavailable(self, client):
         }
         self._session.publish(msg)
 
+    # Returns true iff this is the first client in the session.
     def onClientJoin(self, client):
         '''Queues a late-joiner to receive full state.'''
         clientId = client.clientId
@@ -200,12 +196,14 @@ def onClientJoin(self, client):
 
         roster = self._getRosterList(client)
 
+        isFirst = False
         sendState = False
         # First client in?
         if not len(self._updaters):
             self.addUpdater(client, False)
             data = []
             sendState = True
+            isFirst = True
         elif self._lastState is None:
             self.assignUpdater(client)
         else:
@@ -225,6 +223,7 @@ def onClientJoin(self, client):
                 'channel':'/service/session/join/state',
                 'data': data
             })
+        return isFirst
 
     def onUpdaterSendState(self, updater, data):
         '''Sends full state to a late-joiner waiting for it.'''

servers/python/coweb/session/moderator_late_join_handler.py

@@ -10,6 +10,7 @@ def __init__(self, session):
         super(ModeratorLateJoinHandler, self).__init__(session)
         self._moderator = session._moderator
 
+    # Returns true iff this is the first client in the session.
     def onClientJoin(self, client):
         '''Overrides to have the moderator immediatelty send late join state'''
         clientId = client.clientId
@@ -41,6 +42,8 @@ def onClientJoin(self, client):
 
         if 0 == self.getUpdaterCount():
             self.addUpdater(client, False)
+            return True
         else:
             self.addUpdater(client, True)
+            return False
 

servers/python/coweb/session/session.py

@@ -16,9 +16,25 @@
 # coweb
 from .. import bayeux
 from .. import service
+from .. import OEHandler
+from ..moderator import SessionModerator
 
+OEHandler = OEHandler.OEHandler
 log = logging.getLogger('coweb.session')
 
+def getServiceNameFromChannel(channel, pub):
+    # Public channels are of form /bot/<NAME>
+    # Private channels are of form /service/bot/<NAME>/(request|response)
+    parts = channel.split("/")
+    if pub:
+        if 3 == len(parts):
+            return parts[2]
+    else:
+        if 5 == len(parts) and \
+                ("request" == parts[4] or "response" == parts[4]):
+            return parts[3];
+    return ""
+
 class Session(bayeux.BayeuxManager):
     '''
     Manages a session instance that supports services but no user cooperative
@@ -36,13 +52,23 @@ def __init__(self, container, key, cacheState, *args, **kwargs):
                 self.sessionId
         self.rosterUnavailableChannel = '/session/%s/roster/unavailable' % \
                 self.sessionId
+        self.syncAppChannel = '/session/%s/sync/app' % self.sessionId
+        self.syncEngineChannel = '/session/%s/sync/engine' % self.sessionId
 
         self._connectionClass = SessionConnection
         self._container = container
         self._application = container.webApp
         # bridge between users and service bots
         self._services = service.ServiceSessionBridge(container, self)
 
+        # Operation total order, setup OP engine.
+        self._opOrder = -1
+        self._opengine = OEHandler(self, 0)
+
+        # Use moderator?
+        self._moderator = SessionModerator.getInstance(
+                self._container.moderatorClass, self.key)
+
     def build_connection(self, handler):
         '''Override to build proper connection.'''
         return self._connectionClass(handler, self)
@@ -168,9 +194,12 @@ def on_publish(self, cl, req, res):
         '''Overrides to handle bot requests.'''
         ch = req['channel']
         if ch.startswith('/service/bot'):
+            svcName = getServiceNameFromChannel(ch, False)
             # private bot message
-            if not self._manager.request_for_service(cl, req, res):
-                return
+            if self._manager._moderator.canClientMakeServiceRequest(svcName,
+                    cl, req):
+                if not self._manager.request_for_service(cl, req, res):
+                    return
         elif not self._manager.collab:
             # disallow posting to any other channel by clients
             res['error'] = '402:%s:not-allowed' % client.clientId