Changeset 2596

Timestamp:

07/05/08 22:16:06 (2 months ago)

Author:

jmoore

Message:

ticket:1018 - Working solution (needs concurrency testing)

Each ServiceFactory instance now keeps track of its clientId.

This solution uses a managed map in SessionManagerI to
make up for the missing ObjectAdapter.findAll() method.

Location:

trunk/components/blitz/src/ome/services/blitz

Files:

• fire/SessionManagerI.java (modified) (7 diffs)
• impl/ServiceFactoryI.java (modified) (10 diffs)

trunk/components/blitz/src/ome/services/blitz/fire/SessionManagerI.java

@@ -11 +11 @@
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 
 import ome.conditions.InternalException;
@@ -25 +27 @@
 import ome.system.Principal;
 import ome.system.Roles;
+import omero.ApiUsageException;
 import omero.constants.EVENT;
 import omero.constants.GROUP;
@@ -71 +74 @@
     protected final Set<String> sessionsForReaping = new HashSet<String>();
 
+    /**
+     * An internal mapping to all {@link ServiceFactoryI} instances for a given
+     * session since there is no method on {@link Ice.ObjectAdapter} to retrieve
+     * all servants.
+     */
+    protected final Map<String, Set<String>> sessionToClientIds = new ConcurrentHashMap<String, Set<String>>();
+
     public SessionManagerI(SecuritySystem secSys,
             SessionManager sessionManager, Executor executor) {
@@ -102 +112 @@
 
         try {
-            Ice.ObjectPrx _prx;
+
+            // Create the session for this ServiceFactory
             Principal p = new Principal(userId, group, event);
             ome.model.meta.Session s = sessionManager.create(p);
             Principal sp = new Principal(s.getUuid(), group, event);
 
-            Ice.Identity id = ServiceFactoryI.sessionId(s.getUuid());
-            Ice.Object servant = current.adapter.find(id);
-            if (servant == null) {
-                ServiceFactoryI session = new ServiceFactoryI(context,
-                        sessionManager, executor, sp, CPTORS);
-
-                _prx = current.adapter.add(session, id);
-                if (log.isDebugEnabled()) {
-                    log.debug(String.format("Created session %s for user %s",
-                            id.name, userId));
-                }
+            // Create the ServiceFactory
+            ServiceFactoryI session = new ServiceFactoryI(current, context,
+                    sessionManager, executor, sp, CPTORS);
+
+            Ice.Identity id = session.sessionId(s.getUuid());
+            Ice.ObjectPrx _prx = current.adapter.add(session, id);
+            if (!sessionToClientIds.containsKey(s.getUuid())) {
+                sessionToClientIds.put(s.getUuid(), new HashSet<String>());
+                log.debug(String.format("Created session %s for user %s",
+                        id.name, userId));
             } else {
+                sessionToClientIds.get(s.getUuid()).add(session.clientId);
                 if (log.isDebugEnabled()) {
                     log.debug(String.format("Rejoined session %s for user %s",
                             id.name, userId));
                 }
-                _prx = current.adapter.createProxy(id);
+
             }
             return Glacier2.SessionPrxHelper.uncheckedCast(_prx);
 
-        } catch (RuntimeException t) {
+        } catch (Exception t) {
             ConvertToBlitzExceptionMessage convert = new ConvertToBlitzExceptionMessage(
                     this, t);
@@ -139 +150 @@
             if (convert.to instanceof CannotCreateSessionException) {
                 throw (CannotCreateSessionException) convert.to;
+            } else if (convert.to instanceof ApiUsageException) {
+                ApiUsageException aue = (ApiUsageException) convert.to;
+                throw new CannotCreateSessionException(aue.message);
             }
 
@@ -158 +172 @@
 
             // And unregister the service if possible
-            Ice.Identity id = ServiceFactoryI.sessionId(curr.id.category);
+            Ice.Identity id;
+            try {
+                id = ServiceFactoryI.sessionId(curr, curr.id.category);
+            } catch (ApiUsageException e) {
+                throw new RuntimeException(
+                        "Could not unregister servant: could not create session id");
+            }
             Ice.Object obj = curr.adapter.find(id);
             if (obj instanceof ServiceFactoryI) {
                 ServiceFactoryI sf = (ServiceFactoryI) obj;
-                sf.unregisterServant(Ice.Util.stringToIdentity(key), curr);
+                sf.unregisterServant(Ice.Util.stringToIdentity(key),
+                        curr.adapter);
             }
         } else if (event instanceof DestroySessionMessage) {
@@ -180 +201 @@
      * Unfortunately, that message does not have an {@link Ice.Current} instance
      * and so reaping must happen asynchronously.
+     * 
+     * @param cantUseThisCurrent
+     *            a current from another method invocation which is not usable
+     *            for reaping the given sessions except to get the current
+     *            {@link Ice.ObjectAdapter}
      */
-    private void reapSessions(Ice.Current current) {
+    private void reapSessions(Ice.Current cantUseThisCurrent) {
+        Ice.ObjectAdapter adapter = cantUseThisCurrent.adapter;
         synchronized (sessionsForReaping) {
             List<String> ids = new ArrayList<String>(sessionsForReaping);
             for (String id : ids) {
-                try {
-                    Ice.Identity iid = ServiceFactoryI.sessionId(id);
-                    Ice.Object obj = current.adapter.find(iid);
-                    if (obj == null) {
-                        log.debug(id + " already removed.");
-                    } else {
-                        ServiceFactoryI sf = (ServiceFactoryI) obj;
-                        sf.close(current);
+                for (String clientId : sessionToClientIds.get(id)) {
+                    try {
+                        Ice.Identity iid = ServiceFactoryI.sessionId(clientId,
+                                id);
+                        Ice.Object obj = adapter.find(iid);
+                        if (obj == null) {
+                            log.debug(id + " already removed.");
+                        } else {
+                            ServiceFactoryI sf = (ServiceFactoryI) obj;
+                            sf.doDestroy(adapter);
+                        }
+                        sessionsForReaping.remove(id);
+                    } catch (Exception e) {
+                        log.error("Error reaping session " + id
+                                + " from client " + clientId, e);
                     }
-                    sessionsForReaping.remove(id);
-                } catch (Exception e) {
-                    log.error("Error reaping session " + id, e);
                 }
             }

trunk/components/blitz/src/ome/services/blitz/impl/ServiceFactoryI.java

@@ -126 +126 @@
 public final class ServiceFactoryI extends _ServiceFactoryDisp {
 
+    // STATIC
+
     private final static Log log = LogFactory.getLog(ServiceFactoryI.class);
+
+    // PRIVATE STATE
+
+    boolean doClose = false;
+
+    public final String clientId;
+
+    // SHARED STATE
 
     final SessionManager sessionManager;
@@ -145 +155 @@
     final Map<String, Ice.Object> interactiveSlots;
 
-    /**
-     * The CLIENT_UUIDs for which a close should be invoked on this session.
-     */
-    final Map<String, Boolean> closeUuids;
-
     final Principal principal;
 
@@ -163 +168 @@
     // =========================================================================
 
-    public ServiceFactoryI(OmeroContext context, SessionManager manager,
-            Executor executor, Principal p,
-            List<HardWiredInterceptor> interceptors) {
+    public ServiceFactoryI(Ice.Current current, OmeroContext context,
+            SessionManager manager, Executor executor, Principal p,
+            List<HardWiredInterceptor> interceptors) throws ApiUsageException {
+        this.clientId = clientId(current);
         this.context = context;
         this.sessionManager = manager;
@@ -180 +186 @@
         activeServants = getMap(ACTIVE_SERVANTS, cache);
         interactiveSlots = getMap(INTERACTIVE_SLOTS, cache);
-        closeUuids = getMap(CLOSE_UUIDS, cache);
     }
 
     static String ACTIVE_SERVANTS = "activeServants";
     static String INTERACTIVE_SLOTS = "interactiveSlots";
-    static String CLOSE_UUIDS = "closeUuids";
 
     @SuppressWarnings("unchecked")
@@ -474 +478 @@
 
     public void detachOnDestroy(Ice.Current current) {
-        String key = current.ctx.get(omero.constants.CLIENTUUID.value);
-        closeUuids.remove(key);
+        doClose = false;
     }
 
     @Deprecated
     public void close(Ice.Current current) {
-        closeOnDestroy(current);
+        doClose = false;
     }
 
     public void closeOnDestroy(Ice.Current current) {
-        String key = current.ctx.get(omero.constants.CLIENTUUID.value);
-        closeUuids.put(key, Boolean.TRUE);
+        doClose = true;
     }
 
@@ -501 +503 @@
         // If we are supposed to close, do only so if the ref count
         // is < 1.
-        String key = current.ctx.get(omero.constants.CLIENTUUID.value);
-        if (key != null && ref < 1) {
+        if (doClose && ref < 1) {
 
             ref = sessionManager.close(this.principal.getName());
@@ -512 +513 @@
             }
 
-            if (log.isInfoEnabled()) {
-                log.info(String.format("Closing %s session", this));
+            doDestroy(current.adapter);
+
+        }
+    }
+
+    /**
+     * Performs the actual cleanup operation. Since {@link #destroy()} is called
+     * non-discriminantly by the router, even when a client has just died, we
+     * have this internal method for handling the actual closing of resources.
+     */
+    public void doDestroy(Ice.ObjectAdapter adapter) {
+
+        if (log.isInfoEnabled()) {
+            log.info(String.format("Closing %s session", this));
+        }
+
+        // Cleaning up resources
+
+        // INTERACTIVE
+        Set<String> ipIds = interactiveSlots.keySet();
+        for (String id : ipIds) {
+            InteractiveProcessorI ip = (InteractiveProcessorI) interactiveSlots
+                    .get(id);
+            ip.stop();
+            adapter.remove(Ice.Util.stringToIdentity(id));
+            interactiveSlots.remove(id);
+        }
+
+        // SERVANTS
+        // Here we call the "close()" method on all methods which require
+        // that
+        // logic allowing the IceMethodInvoker to raise the
+        // UnregisterServantEvent, otherwise there is a recursive call back
+        // to close.
+        List<String> ids = activeServices();
+        for (String idString : ids) {
+            Ice.Identity id = Ice.Util.stringToIdentity(idString);
+            Ice.Object obj = adapter.find(id);
+            try {
+                if (obj instanceof StatefulServiceInterface) {
+                    Method m = obj.getClass().getMethod("close",
+                            Ice.Current.class);
+                    Ice.Current __curr = new Ice.Current();
+                    __curr.id = id;
+                    __curr.adapter = adapter;
+                    __curr.operation = "close";
+                    m.invoke(obj, __curr);
+                } else {
+                    unregisterServant(id, adapter);
+                }
+            } catch (Exception e) {
+                log.error("Failure to close: " + idString + "=" + obj, e);
             }
-
-            // Cleaning up resources
-
-            // INTERACTIVE
-            Set<String> ipIds = interactiveSlots.keySet();
-            for (String id : ipIds) {
-                InteractiveProcessorI ip = (InteractiveProcessorI) interactiveSlots
-                        .get(id);
-                ip.stop();
-                current.adapter.remove(Ice.Util.stringToIdentity(id));
-                interactiveSlots.remove(id);
-            }
-
-            // SERVANTS
-            // Here we call the "close()" method on all methods which require
-            // that
-            // logic allowing the IceMethodInvoker to raise the
-            // UnregisterServantEvent, otherwise there is a recursive call back
-            // to close.
-            List<String> ids = activeServices(current);
-            for (String idString : ids) {
-                Ice.Identity id = Ice.Util.stringToIdentity(idString);
-                Ice.Object obj = current.adapter.find(id);
-                try {
-                    if (obj instanceof StatefulServiceInterface) {
-                        Method m = obj.getClass().getMethod("close",
-                                Ice.Current.class);
-                        Ice.Current __curr = new Ice.Current();
-                        __curr.id = id;
-                        __curr.adapter = current.adapter;
-                        __curr.operation = "close";
-                        __curr.mode = current.mode; // FIXME due to bug
-                        // http://www.zeroc.com/forums/bug-reports/3348-ice-current-hashcode-can-throw-npe-null-enum.html
-                        m.invoke(obj, __curr);
-                    } else {
-                        unregisterServant(id, current);
-                    }
-                } catch (Exception e) {
-                    log.error("Failure to close: " + idString + "=" + obj, e);
-                }
-            }
-
-            // All resources cleaned up to the best of our ability,
-            // now we can remove the current session. If an exception if thrown,
-            // there's not much we can do.
-            try {
-                current.adapter.remove(ServiceFactoryI.sessionId(principal
-                        .getName()));
-            } catch (Throwable t) {
-                // FIXME
-                InternalException ie = new InternalException(t.getMessage());
-                ie.setStackTrace(t.getStackTrace());
-            }
+        }
+
+        // All resources cleaned up to the best of our ability,
+        // now we can remove the current session. If an exception if thrown,
+        // there's not much we can do.
+        try {
+            adapter.remove(sessionId(principal.getName()));
+        } catch (Throwable t) {
+            // FIXME
+            InternalException ie = new InternalException(t.getMessage());
+            ie.setStackTrace(t.getStackTrace());
         }
     }
@@ -723 +732 @@
      * to an {@link UnregisterServantMessage}
      */
-    public void unregisterServant(Ice.Identity id, Ice.Current current) {
+    public void unregisterServant(Ice.Identity id, Ice.ObjectAdapter adapter) {
 
         // Here we assume that if the "close()" call is required, that it has
@@ -730 +739 @@
         // onApplicationEvent().
         // Otherwise, it is being called directly by SF.close().
-        Ice.Object obj = current.adapter.remove(id);
+        Ice.Object obj = adapter.remove(id);
         Object removed = activeServants.remove(Ice.Util.identityToString(id));
         if (removed == null) {
@@ -748 +757 @@
     }
 
-    public static Ice.Identity sessionId(String uuid) {
+    // Id Helpers
+    // =========================================================================
+    // Used for naming service factory instances and creating Ice.Identities
+    // from Ice.Currents, etc.
+
+    public static Ice.Identity sessionId(Ice.Current current, String uuid)
+            throws ApiUsageException {
+        String clientId = clientId(current);
+        return sessionId(clientId, uuid);
+    }
+
+    public static Ice.Identity sessionId(String clientId, String uuid) {
         Ice.Identity id = new Ice.Identity();
-        id.category = "session";
+        id.category = "session-" + clientId;
         id.name = uuid;
         return id;
+
+    }
+
+    public Ice.Identity sessionId(String uuid) {
+        return sessionId(clientId, uuid);
+    }
+
+    public static String clientId(Ice.Current current) throws ApiUsageException {
+        String clientId = null;
+        if (current.ctx != null) {
+            clientId = current.ctx.get(omero.constants.CLIENTUUID.value);
+        }
+        if (clientId == null) {
+            throw new ApiUsageException(null, null, "No "
+                    + omero.constants.CLIENTUUID.value
+                    + " key provided in context.");
+        }
+        return clientId;
     }