Changeset 3266

Timestamp:

01/12/09 08:49:32 (6 months ago)

Author:

jmoore

Message:

OmeroClustering : Moving away from jgroups to straight multicast/jdbc approach

The ReplicatedHashMap implementation used from the org.jgroups.blocks
package had issues with our usage of the map. Since we are already
guaranteed a share storage (the db) for all blitz instances within
a single cluster, we are now using Ice's multicast support (from 3.3)
as well as simpleJdbcTemplate to manage the same information that was
previously in jgroups.
(cherry-pick from commit e50eed4204c2a1d360c2b261b5e6a7618573ba65)

Location:

trunk

Files:

• components/blitz/resources/ome/services/blitz-config.xml (modified) (1 diff)
• components/blitz/resources/omero/Internal.ice (added)
• components/blitz/src/ome/services/blitz/fire/Discovery.java (added)
• components/blitz/src/ome/services/blitz/fire/Ring.java (modified) (15 diffs)
• components/blitz/src/ome/services/blitz/util/BlitzConfiguration.java (modified) (4 diffs)
• components/blitz/test/ome/services/blitz/test/ClusteredRingTest.java (added)
• components/blitz/test/ome/services/blitz/test/mock/MockFixture.java (modified) (6 diffs)
• components/blitz/test/ome/services/blitz/test/utests/RingTest.java (modified) (3 diffs)
• components/server/resources/ome/services/datalayer.xml (modified) (1 diff)
• components/server/resources/ome/services/hibernate.xml (modified) (3 diffs)
• components/server/src/ome/tools/spring/JotmFactoryBean.java (modified) (2 diffs)
• etc/internal.cfg (modified) (1 diff)

trunk/components/blitz/resources/ome/services/blitz-config.xml

@@ -30 +30 @@
     <constructor-arg ref="securitySystem"/>
     <constructor-arg ref="executor"/>
+    <constructor-arg ref="ring"/>
   </bean>
 
-  <bean id="ring" factory-bean="configuration" factory-method="getRing"/>
+  <bean id="ring" class="ome.services.blitz.fire.Ring">
+      <constructor-arg ref="simpleJdbcTemplate"/>
+  </bean>
+
   <bean id="Ice.Communicator" factory-bean="configuration" factory-method="getCommunicator"/>
   <bean id="BlitzAdapter" factory-bean="configuration" factory-method="getBlitzAdapter"/>

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

@@ -8 +8 @@
 package ome.services.blitz.fire;
 
+import java.sql.ResultSet;
+import java.sql.SQLException;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.HashSet;
 import java.util.List;
-import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
-import java.util.Vector;
 
 import ome.services.blitz.util.BlitzConfiguration;
 import ome.services.messages.CreateSessionMessage;
 import ome.services.messages.DestroySessionMessage;
+import ome.services.sessions.SessionManager;
+import omero.internal.ClusterPrx;
+import omero.internal.ClusterPrxHelper;
+import omero.internal.DiscoverCallbackPrx;
+import omero.internal._ClusterDisp;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-import org.jgroups.Address;
-import org.jgroups.ChannelFactory;
-import org.jgroups.JChannelFactory;
-import org.jgroups.View;
 import org.jgroups.blocks.ReplicatedHashMap;
 import org.springframework.context.ApplicationEvent;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.context.ApplicationListener;
 import org.springframework.context.event.ContextClosedEvent;
+import org.springframework.dao.DataIntegrityViolationException;
+import org.springframework.jdbc.BadSqlGrammarException;
+import org.springframework.jdbc.core.simple.ParameterizedRowMapper;
+import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
 
 import Glacier2.CannotCreateSessionException;
@@ -36 +42 @@
 import Glacier2.SessionManagerPrxHelper;
 import Glacier2.SessionPrx;
+import Ice.Current;
 
 /**
  * Distributed ring of {@link BlitzConfiguration} objects which manages lookups
  * of sessions and other resources from all the blitzes which take part in the
- * ring. Membership in the {@link Ring} is based on a single token --
+ * cluster. Membership in the {@link Ring} is based on a single token --
  * "omero.instance" -- retrieved from the current context, or if missing, a
  * calculated value which will prevent this instance from taking part in
@@ -49 +56 @@
  *@since Beta4
  */
-public class Ring implements ReplicatedHashMap.Notification<String, String>,
-        ApplicationListener {
+public class Ring extends _ClusterDisp implements ApplicationListener, ApplicationEventPublisherAware {
 
     private final static Log log = LogFactory.getLog(Ring.class);
@@ -60 +66 @@
     private final static String SESSIONS = "session-";
 
-    private final String uuid = UUID.randomUUID().toString();
-
+    /**
+     * UUID for this cluster node. Used to uniquely identify the session manager
+     * in this blitz instance.
+     */
+    public final String uuid = UUID.randomUUID().toString();
+
+    private final SimpleJdbcTemplate jdbc;
+
+    private/*final */ApplicationEventPublisher publisher;
+    
     private/* final */Ice.Communicator communicator;
 
+    /**
+     * Multicast-based adapter solely for cluster communication.
+     */
+    private/* final */Ice.ObjectAdapter clusterAdapter;
+
+    /**
+     * Standard blitz adapter which is used for the callback.
+     */
+    private/* final */Ice.ObjectAdapter adapter;
+
+    /**
+     * {@link Ice.ObjectPrx} for this {@link Ring} instance as added to the
+     * {@link #adapter} in the {@link #init(Ice.ObjectAdapter, String)} method.
+     */
+    private/* final */Ice.ObjectPrx ownProxy;
+
+    /**
+     * Multicast (datagram) proxy to all other cluster nodes.
+     */
+    private/* final */ClusterPrx cluster;
+    
+    /**
+     * Direct proxy value to the {@link SessionManager} in this blitz instance.
+     */
     private/* final */String directProxy;
 
-    private final ReplicatedHashMap<String, String> map;
-
-    private final String groupname;
-
-    public static String determineGroupName() {
-        String defaultValue = "Runtime" + Runtime.getRuntime().hashCode();
-        String environValue = System.getenv("OMERO_INSTANCE");
-        if (environValue == null || environValue.length() == 0) {
-            environValue = defaultValue;
-        }
-        return environValue;
-    }
-
-    public Ring(String props) {
-        this(determineGroupName(), props);
-    }
-
-    public Ring(String groupname, String props) {
-        this.groupname = groupname;
+    public Ring(SimpleJdbcTemplate jdbc) {
+        this.jdbc = jdbc;
         try {
-            ChannelFactory factory = new JChannelFactory();
-
-            map = new ReplicatedHashMap<String, String>(groupname, factory,
-                    props, 10000);
-            map.setBlockingUpdates(true);
-            map.addNotifier(this);
-            map.start(1000);
+            jdbc.update("create table session_ring "
+                    + "(key varchar unique, value varchar)");
+            log.info("Created OMERO.cluster table");
+        } catch (BadSqlGrammarException bsge) {
+            // Here we assume that this means that the table
+            // already exists.
+            log.info("session_ring already exists");
+        }
+    }
+
+    /**
+     * Passed to the {@link Discovery} instance for signalling completion from
+     * its background {@link Thread} 
+     */
+    public void setApplicationEventPublisher(ApplicationEventPublisher arg0) {
+        this.publisher = arg0;
+    }
+    
+    // Configuration and cluster usage
+    // =========================================================================
+
+    /**
+     * Typically called from within {@link BlitzConfiguration} after the
+     * communicator and adapter have been properly setup.
+     */
+    public void init(Ice.ObjectAdapter adapter, String directProxy) {
+        this.adapter = adapter;
+        this.communicator = adapter.getCommunicator();
+        this.directProxy = directProxy;
+
+        // The cluster we belong to.
+        Ice.ObjectPrx prx = this.communicator.propertyToProxy("ClusterProxy");
+        prx = prx.ice_datagram();
+        cluster = ClusterPrxHelper.uncheckedCast(prx);
+        
+        // Before we add our self we check the validity of the cluster.
+        checkClusterAndAddSelf();
+        put(MANAGERS + uuid, directProxy);
+        putIfAbsent(CONFIG + "redirect", uuid);
+        log.info("Current redirect: " + getRedirect());
+    }
+
+    /**
+     * Method called during initialization to get all the active uuids within
+     * the cluster, and remove any dead nodes. After this, we add this instance
+     * to the cluster.
+     */
+    protected void checkClusterAndAddSelf() {
+
+        // Now create the call back and start it
+        // The instance registers and removes itself.
+        Discovery discovery = new Discovery(adapter, this, this.publisher);
+        try {
+            cluster.discover(discovery.getProxy());
+            new Thread(discovery).start();
+        } catch (Ice.NoEndpointException nee) {
+            log.warn("No cluster endpoints found", nee);
+        }
+
+        // Now our checking is done, add ourselves.
+        this.clusterAdapter = this.communicator.createObjectAdapter("Cluster");
+        ownProxy = this.clusterAdapter.add(this, this.communicator
+                .stringToIdentity("Cluster"));
+        this.clusterAdapter.activate();
+    }
+
+    /**
+     * Returns the uuid for this manager instance. Other instances will use this
+     * value to keep the session_ring table in sync.
+     */
+    public void discover(DiscoverCallbackPrx cb, Current __current) {
+        try {
+            cb.clusterNodeUuid(uuid);
+            log.info("Sent cluster node uuid: " + uuid);
         } catch (Exception e) {
-            throw new RuntimeException("Could not start ring: ", e);
-        }
-        log.info("Initialized ring in group " + groupname);
-
-    }
-
-    public Ring() {
-        this("session_ring.xml");
-    }
-
-    /**
-     * 
-     * @param bc
-     */
-    public void init(Ice.Communicator communicator, String directProxy) {
-        this.communicator = communicator;
-        this.directProxy = directProxy;
-        map.put(MANAGERS + uuid, directProxy);
-        map.putIfAbsent(CONFIG + "redirect", directProxy);
-        log.info("Current redirect: " + getRedirect());
-    }
-
+            log.warn("Exception while sending cluster node uuid: "+uuid, e);
+        }
+    }
+
+    /**
+     * Called when any node goes down. First we try to remove any redirect for that
+     * instance. Then we try to install ourselves.
+     */
+    public void down(String downUuid, Current __current) {
+        removeIfEquals(CONFIG+"redirect", downUuid);
+        if (putIfAbsent(CONFIG+"redirect", this.uuid)) {
+            log.info("Installed self as new redirect: "+uuid);
+        }
+    }
+    
     public void destroy() {
         try {
-            log.info("Shutting down ring in group " + groupname);
-            map.remove(MANAGERS + uuid);
-            map.stop();
+            this.clusterAdapter.deactivate();
+            cluster.down(this.uuid);
+            remove(MANAGERS + uuid);
+            removeIfEquals(CONFIG+"redirect", uuid);
+            log.info("Disconnected from OMERO.cluster");
         } catch (Exception e) {
             log.error("Error stopping ring " + this, e);
@@ -126 +209 @@
     }
 
-    // Our usage
+    // Local usage
     // =========================================================================
 
     public boolean checkPassword(String userId) {
-        return map.get(SESSIONS + userId) != null;
+        boolean rv = get(SESSIONS + userId) != null;
+        log.info(String.format("Checking password: %s [%s]", userId, rv));
+        return rv;
     }
 
@@ -140 +225 @@
      */
     public String getRedirect() {
-        String redirect = map.get(CONFIG + "redirect");
+        String redirect = get(CONFIG + "redirect");
         return redirect;
     }
@@ -150 +235 @@
      * value is set. In either case, the previous value is returned.
      */
-    public String putRedirect(String uuidOrProxy) {
-        String oldValue;
+    public void putRedirect(String uuidOrProxy) {
         if (uuidOrProxy == null || uuidOrProxy.length() == 0) {
-            oldValue = map.remove(CONFIG + "redirect");
+            remove(CONFIG + "redirect");
         } else {
-            oldValue = map.put(CONFIG + "redirect", uuidOrProxy);
-        }
-        return oldValue;
+            put(CONFIG + "redirect", uuidOrProxy);
+        }
     }
 
@@ -177 +260 @@
                 proxyString = null;
             } else {
-                if (!map.containsKey(MANAGERS + redirect)) {
+                if (!containsKey(MANAGERS + redirect)) {
                     log.warn("No proxy found for manager: " + redirect);
                 } else {
-                    proxyString = map.get(MANAGERS + redirect);
+                    proxyString = get(MANAGERS + redirect);
                     log.info("Resolved redirect to: " + proxyString);
                 }
@@ -190 +273 @@
 
             // Check if the session is in ring
-            proxyString = map.get(SESSIONS + userId);
+            proxyString = get(SESSIONS + userId);
             if (proxyString != null && !proxyString.equals(directProxy)) {
                 log.info(String.format("Returning remote session on %s",
@@ -198 +281 @@
             // or needs to be load balanced
             else {
-                double IMPOSSIBLE = 309340.0;
+                double IMPOSSIBLE = 314159.0;
                 if (Math.random() > IMPOSSIBLE) {
                     Set<String> values = filter(MANAGERS);
@@ -209 +292 @@
                             List v = new ArrayList(values);
                             proxyString = (String) v.get(idx);
-                            proxyString = map.get(MANAGERS + proxyString);
+                            proxyString = get(MANAGERS + proxyString);
                             log.info(String.format("Load balancing to %s",
                                     proxyString));
@@ -238 +321 @@
     }
 
+    public Set<String> knownManagers() {
+        Set<String> managers = filter(MANAGERS);
+        Set<String> rv = new HashSet<String>();
+        for (String manager : managers) {
+            manager = manager.replaceFirst(MANAGERS, "");
+            rv.add(manager);
+        }
+        return rv;
+    }
+    
+    public void assertNodes(Set<String> nodeUuids) {
+        Set<String> managers = knownManagers();
+        for (String manager : managers) {
+            System.out.println(manager+"?="+nodeUuids);
+            if (!nodeUuids.contains(manager)) {
+                purgeNode(manager);
+            }
+        }
+    }
+
+    protected void purgeNode(String manager) {
+        log.info("Purging node: " + manager);
+        int count = jdbc.update("delete from session_ring where key like '"
+                + SESSIONS + "' and value = ?", manager);
+        log.info("Removed " + count + " sessions with value " + manager);
+        count = jdbc.update("delete from session_ring where key = ?", MANAGERS
+                + manager);
+        log.info("Removed " + MANAGERS + manager);
+        count = jdbc.update(
+                "delete from session_ring where key = ? and value = ?", CONFIG
+                        + "redirect", manager);
+        if (count != 0) {
+            log.info("Removed redirect to " + manager);
+        }
+        putRedirect(uuid);
+    }
+
     // Events
     // =========================================================================
@@ -243 +363 @@
     public void onApplicationEvent(ApplicationEvent arg0) {
         if (arg0 instanceof CreateSessionMessage) {
-            String uuid = ((CreateSessionMessage) arg0).getSessionId();
-            map.put(SESSIONS + uuid, directProxy);
+            String session = ((CreateSessionMessage) arg0).getSessionId();
+            put(SESSIONS + session, uuid); // Use our uuid rather than proxy
         } else if (arg0 instanceof DestroySessionMessage) {
-            String uuid = ((DestroySessionMessage) arg0).getSessionId();
-            map.remove(SESSIONS + uuid);
+            String session = ((DestroySessionMessage) arg0).getSessionId();
+            remove(SESSIONS + session);
         } else if (arg0 instanceof ContextClosedEvent) {
             // This happens 3 times for each nested context. Perhaps we
@@ -254 +374 @@
     }
 
-    // Notification interface
+    // Map interface
     // =========================================================================
-
-    public void contentsCleared() {
-        log.info("cleared");
-    }
-
-    public void contentsSet(Map<String, String> arg0) {
-        log.info("set contents:" + arg0);
-    }
-
-    public void entryRemoved(String arg0) {
-        log.info("remove:" + arg0);
-    }
-
-    public void entrySet(String arg0, String arg1) {
-        log.info("set:" + arg0 + "=" + arg1);
-    }
-
-    public void viewChange(View arg0, Vector<Address> arg1, Vector<Address> arg2) {
-        log.info("view change:" + arg0);
-    }
-
-    // Main
-    // =========================================================================
-
-    public static void main(String[] args) throws Exception {
-
-        Ring ring = new Ring();
+    // If the implementation of this class needs to be changed, these should be
+    // the only methods which need to be replaced. i.e. subclass implementors
+    // may want to start here.
+
+    public boolean containsKey(String key) {
+        int count = jdbc.queryForInt("select count(key) "
+                + "from session_ring " + "where key = ?", key);
+        return count > 0;
+    }
+
+    public String get(String key) {
+        String value = (String) jdbc.queryForObject("select value "
+                + "from session_ring " + "where key = ?", String.class, key);
+        return value;
+    }
+
+    public void put(String key, String value) {
+        boolean wasPut = putIfAbsent(key, value);
+        if (!wasPut) {
+            int count = jdbc.update(
+                    "update session_ring set value = ? where key = ?", value,
+                    key);
+            if (count == 0) {
+                log.info("Key not found for update: " + key);
+            } else {
+                log.info(String.format("Updated key %s with value %s", key,
+                        value));
+            }
+        }
+    }
+
+    public boolean putIfAbsent(String key, String value) {
         try {
-
-            if (args == null || args.length == 0) {
-                StringBuilder sb = new StringBuilder();
-                sb.append("java Ring [sessions | config | redirect [value]");
-                sb.append("\n");
-                sb.append("  print fqn        -    print all values at fqn\n");
-                sb.append("  redirect         -    print current redirect\n");
-                sb
-                        .append("  redirect [value] -    set current redirect. Empty removes\n");
-                sb.append(" \n");
-                System.out.println(sb.toString());
-            }
-
-            if (args[0].equals("print")) {
-                if (args.length > 1) {
-                    for (int i = 1; i < args.length; i++) {
-                        ring.printTree(args[i]);
+            int count = jdbc.update(
+                    "insert into session_ring (key, value) values (?, ?)", key,
+                    value);
+            if (count > 0) {
+                log.info(String.format("Put new key: %s with value: %s ", key,
+                        value));
+                return true;
+            }
+        } catch (DataIntegrityViolationException dive) {
+            // The key already exists in the table. This is therefore
+            // a no-op
+        }
+        return false;
+    }
+
+    /**
+     * Removes a key from the session_ring table if present.
+     * 
+     * @param key
+     */
+    public boolean remove(String key) {
+        int count = jdbc.update("delete from session_ring where key = ?", key);
+        if (count == 0) {
+            log.info("Key not found to remove: " + key);
+            return false;
+        } else {
+            log.info("Removed key: " + key);
+            return true;
+        }
+    }
+
+    /**
+     * Used to remove a key/value pair iff the value equals the value give here.
+     */
+    public boolean removeIfEquals(String key, String value) {
+        int count = jdbc
+                .update("delete from session_ring where key = ? and value = ?", key, value);
+        if (count == 0) {
+            log.info("Key and value do not match: " + key + "=" + value);
+            return false;
+        } else {
+            log.info("Removed key: " + key + " since value matched: " + value);
+            return true;
+        }
+    }
+
+    public List<String> keySet() {
+        return jdbc.query("select key from session_ring",
+                new ParameterizedRowMapper<String>() {
+                    public String mapRow(ResultSet arg0, int arg1)
+                            throws SQLException {
+                        return arg0.getString(1);
                     }
-                } else {
-                    ring.printAll();
-                }
-            } else if (args[0].equals("redirect")) {
-                if (args.length > 1) {
-                    String value = args[1];
-                    ring.putRedirect(value);
-                } else {
-                    ring.printRedirect();
-                }
-            } else if (args[0].equals("raw")) {
-                System.out.println(ring.map);
-            }
-
-        } finally {
-            ring.destroy();
-        }
-    }
-
-    public void printAll() {
-        for (String fqn : Arrays.asList(SESSIONS, MANAGERS, CONFIG)) {
-            printTree(fqn);
-        }
-    }
-
-    public void printTree(String prefix) {
-        System.out.println("===== " + prefix + " =====");
-        Set<String> keys = filter(prefix);
-        if (keys != null && keys.size() > 0) {
-            for (String key : keys) {
-                System.out.println(String.format("%s\t%s", key, map.get(key)));
-            }
-        } else {
-            System.out.println("(empty)");
-        }
-    }
-
-    public void printRedirect() {
-        Object uuidOrProxy = map.get(CONFIG + "redirect");
-        if (uuidOrProxy != null && uuidOrProxy.toString().length() > 0) {
-            System.out.println(uuidOrProxy.toString());
-        }
+                });
     }
 
@@ -350 +469 @@
 
     private Set<String> filter(String prefix) {
-        Set<String> values = new HashSet<String>(map.keySet());
+        Set<String> values = new HashSet<String>(keySet());
         Set<String> remove = new HashSet<String>();
         for (String value : values) {

trunk/components/blitz/src/ome/services/blitz/util/BlitzConfiguration.java

@@ -70 +70 @@
      * should be careful to check for nulls.
      */
-    public BlitzConfiguration(
+    public BlitzConfiguration(Ring ring,
             ome.services.sessions.SessionManager sessionManager,
             SecuritySystem securitySystem, Executor executor)
             throws RuntimeException {
-        this(createId(), sessionManager, securitySystem, executor);
+        this(createId(), ring, sessionManager, securitySystem, executor);
     }
 
@@ -89 +89 @@
      * @throws RuntimeException
      */
-    public BlitzConfiguration(Ice.InitializationData id,
+    public BlitzConfiguration(Ice.InitializationData id, Ring ring,
             ome.services.sessions.SessionManager sessionManager,
             SecuritySystem securitySystem, Executor executor)
@@ -97 +97 @@
 
         this.id = id;
-        this.blitzRing = new Ring();
+        this.blitzRing = ring;
         this.communicator = createCommunicator();
 
@@ -112 +112 @@
             managerDirectProxy = blitzAdapter.createDirectProxy(managerId());
 
-            // Must inject configuration before starting the adapter
-            blitzRing.init(communicator, communicator
+            blitzAdapter.activate();
+
+            // When using adapter methods from within the ring, it is necessary
+            // to start the adapter first.
+            blitzRing.init(blitzAdapter, communicator
                     .proxyToString(getDirectProxy()));
-            blitzAdapter.activate();
+
         } catch (RuntimeException e) {
             destroy();

trunk/components/blitz/test/ome/services/blitz/test/mock/MockFixture.java

@@ -27 +27 @@
 import omero.model.PermissionsI;
 import omero.util.ObjectFactoryRegistrar;
-
 import org.jmock.Mock;
 import org.jmock.MockObjectTestCase;
 import org.springframework.aop.target.HotSwappableTargetSource;
+import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
 
 import Ice.InitializationData;
@@ -44 +44 @@
     public final SessionManager mgr;
     public final SecuritySystem ss;
+    public final SimpleJdbcTemplate jdbc;
 
     public static OmeroContext basicContext() {
@@ -49 +50 @@
                 "classpath:ome/services/blitz-servantDefinitions.xml",
                 "classpath:ome/services/throttling/throttling.xml",
-                "classpath:ome/services/messaging.xml" });
+                "classpath:ome/services/messaging.xml",
+                "classpath:ome/services/datalayer.xml",
+                "classpath:ome/config.xml"});
     }
 
@@ -71 +74 @@
         this.ss = (SecuritySystem) ctx.getBean("securitySystem");
         this.mgr = (SessionManager) ctx.getBean("sessionManager");
-
+        this.jdbc = (SimpleJdbcTemplate) ctx.getBean("simpleJdbcTemplate");
+        
         // --------------------------------------------
 
@@ -93 +97 @@
         // Basic configuration
         id.properties.setProperty("BlitzAdapter.Endpoints","default -h 127.0.0.1");
-
-        blitz = new BlitzConfiguration(id, mgr, ss, ex);
+        // Cluster configuration from etc/internal.cfg
+        id.properties.setProperty("Cluster.Endpoints","udp -h 224.0.0.5 -p 10000");
+        id.properties.setProperty("ClusterProxy","Cluster:udp -h 224.0.0.5 -p 10000");
+        
+        
+        Ring ring = new Ring(jdbc);
+        ring.setApplicationEventPublisher(ctx);
+        
+        blitz = new BlitzConfiguration(id, ring, mgr, ss, ex);
         this.sm = (SessionManagerI) blitz.getBlitzManager();
         // The following is a bit of spring magic so that we can configure
@@ -105 +116 @@
 
     public void tearDown() {
-        this.blitz.getCommunicator().destroy();
-        this.ctx.closeAll();
+        this.blitz.destroy();
+        // this.ctx.closeAll();
     }
 

trunk/components/blitz/test/ome/services/blitz/test/utests/RingTest.java

@@ -7 +7 @@
 package ome.services.blitz.test.utests;
 
+import javax.sql.DataSource;
+
 import ome.services.blitz.fire.Ring;
+import ome.system.OmeroContext;
 
-import org.jgroups.blocks.ReplicatedTree;
 import org.jmock.Mock;
 import org.jmock.MockObjectTestCase;
-import org.testng.annotations.AfterClass;
-import org.testng.annotations.BeforeClass;
+import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
 import org.testng.annotations.BeforeTest;
 import org.testng.annotations.Test;
@@ -23 +24 @@
 public class RingTest extends MockObjectTestCase {
 
-    ReplicatedTree tree1, tree2;
+    OmeroContext ctx;
+    SimpleJdbcTemplate jdbc;
+    Ice.ObjectAdapter oa;
     Ice.Communicator ic;
-    Mock mockIc; 
-    
-    @BeforeClass
-    public void setup() throws Exception {
-        tree1 = new ReplicatedTree("test", "session_ring.xml", 1000);
-        tree2= new ReplicatedTree("test", "session_ring.xml", 1000);
-        tree1.start();
-        tree2.start();
-    }
-    
+    Mock mockIc, mockOa;
+
     @BeforeTest
     public void setupMethod() throws Exception {
         mockIc = mock(Ice.Communicator.class);
         ic = (Ice.Communicator) mockIc.proxy();
+        mockOa = mock(Ice.ObjectAdapter.class);
+        oa = (Ice.ObjectAdapter) mockOa.proxy();
+        ctx = new OmeroContext(new String[] {
+                "classpath:ome/config.xml",
+                "classpath:ome/services/datalayer.xml" });
+        DataSource dataSource = (DataSource) ctx.getBean("dataSource");
+        jdbc = new SimpleJdbcTemplate(dataSource);
+
     }
-    
-    @AfterClass
-    public void tearDown() {
-        tree1.stop();
-        tree2.stop();
-    }
-    
-    //@Test
-    public void testMain() throws Exception {
-        Ring.main(new String[]{});
-    }
-    
+
     @Test
     public void testFirstTakesOver() throws Exception {
-        Ring one = new Ring("takeover", "session_ring.xml");
-        one.init(ic, "one");
+        Ring one = new Ring(jdbc);
+        one.setApplicationEventPublisher(ctx);
+        one.init(oa, "one");
         assertEquals("one", one.getRedirect());
-        Ring two = new Ring("takeover", "session_ring.xml");
-        two.init(ic, "two");
+        Ring two = new Ring(jdbc);
+        two.setApplicationEventPublisher(ctx);
+        two.init(oa, "two");
         assertEquals("one", two.getRedirect());
     }
@@ -66 +60 @@
         fail();
     }
-    
+
     @Test
     public void testHandlesMissingServers() throws Exception {
         fail();
     }
-    
+
     @Test
     public void testRemovesUnreachable() throws Exception {
         fail();
     }
-    
+
     @Test
     public void testReaddsSelfIfTemporarilyUnreachable() throws Exception {
         fail();
     }
-    
+
     @Test
-    public void testLotsOfCalls() throws Exception {
-        int i = 0;
-        int j = 0;
-        long start = System.currentTimeMillis();
-        for (int k = 0; k < 100; k++) {
-            i++; j++;
-            tree1.put("/1", i+"", i+"");
-            tree2.put("/2", j+"", j+"");
-        }
-
-
-        assertEquals(100, tree2.getKeys("/1").size());
-        assertEquals(100, tree1.getKeys("/2").size());
-    
-    }
-    
-    @Test
-    public void testPrintSessions() throws Exception {
-        Ring ring = new Ring("test", "session_ring.xml");
-        ring.printTree("/SESSIONS");
-        ring.destroy();
+    public void testAllSessionRemovedIfDiscoveryFails() throws Exception {
+        fail();
     }
 
+    @Test
+    public void testAllSessionsReassertedIfSessionComesBackOnline()
+            throws Exception {
+        fail();
+    }
 }

trunk/components/server/resources/ome/services/datalayer.xml

@@ -69 +69 @@
     </property>
   </bean>
-
+  
+  <bean id="transactionTemplate" class="org.springframework.transaction.support.TransactionTemplate">
+    <constructor-arg ref="transactionManager"/>
+  </bean>
+  
+  <bean id="simpleJdbcTemplate" class="org.springframework.jdbc.core.simple.SimpleJdbcTemplate">
+    <description>
+    Scope: private
+    
+    This template uses session.connection() for data access. See javadocs for
+    more info.
+    </description>
+    <constructor-arg ref="dataSource"/>
+  </bean>
+  
   <bean id="unsafeJdbcTemplate" class="org.springframework.jdbc.core.simple.SimpleJdbcTemplate">
     <description>

trunk/components/server/resources/ome/services/hibernate.xml

@@ -75 +75 @@
   </bean>
   
-  <bean id="transactionTemplate" class="org.springframework.transaction.support.TransactionTemplate">
-    <constructor-arg ref="transactionManager"/>
-  </bean>
-  
   <bean id="proxyHandler" class="ome.tools.hibernate.ProxyCleanupFilter$Interceptor">
     <description>
@@ -111 +107 @@
     <property name="allowCreate" value="false"/>
   </bean>
-
-  <bean id="simpleJdbcTemplate" class="org.springframework.jdbc.core.simple.SimpleJdbcTemplate">
-    <description>
-    Scope: private
-    
-    This template uses session.connection() for data access. See javadocs for
-    more info.
-    </description>
-    <constructor-arg ref="dataSource"/>
-  </bean>
   
   <bean id="sessionFactory" class="org.springframework.orm.hibernate3.annotation.AnnotationSessionFactoryBean">
@@ -141 +127 @@
         <bean id="transactionManager.transactionManager" class="org.springframework.beans.factory.config.PropertyPathFactoryBean"/>
     </property>
+    <!-- http://coffeedrivenjava.blogspot.com/2006/08/distributed-xa-transactions-w-hibernate.html -->
+    <property name="useTransactionAwareDataSource" value="true"/>
   </bean>
 

trunk/components/server/src/ome/tools/spring/JotmFactoryBean.java

@@ -10 +10 @@
 import javax.naming.Context;
 import javax.naming.InitialContext;
+import javax.naming.NameAlreadyBoundException;
 import javax.naming.NamingException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 
 /**
@@ -20 +24 @@
         org.springframework.transaction.jta.JotmFactoryBean {
 
+    private final static Log log = LogFactory.getLog(JotmFactoryBean.class);
+    
     public JotmFactoryBean() throws NamingException {
         super();
-        Context context = new InitialContext();
-        context.bind("java:comp/UserTransaction", getObject());
+        try {
+            Context context = new InitialContext();
+            context.bind("java:comp/UserTransaction", getObject());
+            log.info("Bound UserTransaction in JNDI");
+        } catch (NameAlreadyBoundException nabe) {
+            log.info("UserTransaction already bound in JNDI");
+        }
     }
 

trunk/etc/internal.cfg

@@ -6 +6 @@
 IceGridAdmin.Username=root
 IceGridAdmin.Password=ome
+
+#
+# Used to communicate between instances
+# See: omero/Internal.ice
+#
+Cluster.Endpoints=udp -h 239.0.0.5 -p 22222
+ClusterProxy=Cluster:udp -h 239.0.0.5 -p 22222