[hornetq-commits] JBoss hornetq SVN: r9569 - in branches/2_2_0_HA_Improvements/src/main/org/hornetq/core: server/cluster/impl and 1 other directory.

[do-not-reply at jboss.org]

Author: jmesnil
Date: 2010-08-20 06:18:05 -0400 (Fri, 20 Aug 2010)
New Revision: 9569

Modified:
   branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/ServerLocatorImpl.java
   branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java
   branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionBridge.java
Log:
HA refactoring

* moved code to reconnect from ServerLocator to ClusterConnectionBridge

Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/ServerLocatorImpl.java
===================================================================
--- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/ServerLocatorImpl.java	2010-08-20 08:17:24 UTC (rev 9568)
+++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/ServerLocatorImpl.java	2010-08-20 10:18:05 UTC (rev 9569)
@@ -1125,14 +1125,13 @@
          return;
       }
 
-      final TopologyMember member = topology.getMember(nodeID);
       removed = topology.removeMember(nodeID);
 
       if (!topology.isEmpty())
       {
          updateArraysAndPairs();
 
-         if (topology.size() == 1 && topology.getMember(nodeID) != null)
+         if (topology.size() == 1 && topology.getMember(this.nodeID) != null)
          {
             receivedTopology = false;
          }
@@ -1146,39 +1145,6 @@
          receivedTopology = false;
       }
 
-      if (ha && discoveryAddress == null && removed)
-      {
-         threadPool.execute(new Runnable()
-         {
-            public void run()
-            {
-               System.out.println(ServerLocatorImpl.this.nodeID + " will try to connect to " + nodeID);
-               ClientSessionFactory sf = null;
-               do
-               {
-                  try
-                  {
-                     Pair<TransportConfiguration,TransportConfiguration> pair = member.getConnector();
-                     TransportConfiguration tc = (pair.a != null) ? pair.a : pair.b;
-                     sf = createSessionFactory(tc);
-                  }
-                  catch (HornetQException e)
-                  {
-                     if (e.getCode() == HornetQException.NOT_CONNECTED)
-                     {
-                        continue;
-                     }
-                  }
-                  catch (Exception e)
-                  {
-                     break;
-                  }
-               }
-               while (sf == null);
-            }
-         });
-      }
-
       if (removed)
       {
          for (ClusterTopologyListener listener : topologyListeners)
@@ -1245,7 +1211,7 @@
          this.initialConnectors[count++] = entry.getConnector();
       }
       
-      if (ha && clusterConnection && !receivedTopology)
+      if (ha && clusterConnection && !receivedTopology && initialConnectors.length > 0)
       {
          // FIXME the node is alone in the cluster. We create a connection to the new node
          // to trigger the node notification to form the cluster.

Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java
===================================================================
--- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java	2010-08-20 08:17:24 UTC (rev 9568)
+++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java	2010-08-20 10:18:05 UTC (rev 9569)
@@ -74,7 +74,7 @@
 
    private Queue queue;
 
-   private final Executor executor;
+   protected final Executor executor;
 
    private final Filter filter;
 

Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionBridge.java
===================================================================
--- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionBridge.java	2010-08-20 08:17:24 UTC (rev 9568)
+++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionBridge.java	2010-08-20 10:18:05 UTC (rev 9569)
@@ -231,15 +231,48 @@
    @Override
    public void connectionFailed(HornetQException me)
    {
-      try
+      if (!session.isClosed())
       {
-         session.cleanUp(false);
+         try
+         {
+            session.cleanUp(false);
+         }
+         catch (Exception e)
+         {
+            log.warn("Unable to clean up the session after a connection failure", e);
+         }
+         serverLocator.notifyNodeDown(targetNodeID);
+         if (serverLocator.getDiscoveryAddress() == null)
+         {
+            executor.execute(new Runnable()
+            {
+               
+               public void run()
+               {
+                  ClientSessionFactory sf = null;
+                  do
+                  {
+                     try
+                     {
+                        sf = serverLocator.createSessionFactory(connector);
+                     }
+                     catch (HornetQException e)
+                     {
+                        if (e.getCode() == HornetQException.NOT_CONNECTED)
+                        {
+                           continue;
+                        }
+                     }
+                     catch (Exception e)
+                     {
+                        break;
+                     }
+                  }
+                  while (sf == null);
+               }
+            });
+         }
       }
-      catch (Exception e)
-      {
-    	  log.warn("Unable to clean up the session after a connection failure", e);
-      }
-      serverLocator.notifyNodeDown(targetNodeID);
       super.connectionFailed(me);
    }