[hornetq-commits] JBoss hornetq SVN: r8446 - trunk/src/main/org/hornetq/core/server/cluster/impl.

[do-not-reply at jboss.org]

Author: jmesnil
Date: 2009-11-28 05:37:27 -0500 (Sat, 28 Nov 2009)
New Revision: 8446

Modified:
   trunk/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java
Log:
bridge reconnection if server is started but not accepting session creation

* if the bridge creates its session while its destination is started but not
  accepting session (SESSION_CREATION_REJECTED), retry to create the session
  until it succeeds

Modified: trunk/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java	2009-11-28 10:15:45 UTC (rev 8445)
+++ trunk/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java	2009-11-28 10:37:27 UTC (rev 8446)
@@ -638,59 +638,85 @@
       {
          return false;
       }
+      
+      boolean retry = false;
 
-      try
+      do
       {
-         if (discoveryAddress != null)
+         try
          {
-            csf = new ClientSessionFactoryImpl(discoveryAddress, discoveryPort);
-         }
-         else
-         {
-            csf = new ClientSessionFactoryImpl(connectorPair.a, connectorPair.b);
-         }
+            if (discoveryAddress != null)
+            {
+               csf = new ClientSessionFactoryImpl(discoveryAddress, discoveryPort);
+            }
+            else
+            {
+               csf = new ClientSessionFactoryImpl(connectorPair.a, connectorPair.b);
+            }
 
-         csf.setFailoverOnServerShutdown(failoverOnServerShutdown);
-         csf.setRetryInterval(retryInterval);
-         csf.setRetryIntervalMultiplier(retryIntervalMultiplier);
-         csf.setReconnectAttempts(reconnectAttempts);
-         csf.setBlockOnPersistentSend(false);
+            csf.setFailoverOnServerShutdown(failoverOnServerShutdown);
+            csf.setRetryInterval(retryInterval);
+            csf.setRetryIntervalMultiplier(retryIntervalMultiplier);
+            csf.setReconnectAttempts(reconnectAttempts);
+            csf.setBlockOnPersistentSend(false);
 
-         // Must have confirmations enabled so we get send acks
+            // Must have confirmations enabled so we get send acks
 
-         csf.setConfirmationWindowSize(confirmationWindowSize);
+            csf.setConfirmationWindowSize(confirmationWindowSize);
 
-         // Session is pre-acknowledge
-         session = (ClientSessionInternal)csf.createSession(clusterUser, clusterPassword, false, true, true, true, 1);
+            // Session is pre-acknowledge
+            session = (ClientSessionInternal)csf.createSession(clusterUser, clusterPassword, false, true, true, true, 1);
 
-         if (session == null)
-         {
-            // This can happen if the bridge is shutdown
-            return false;
-         }
+            if (session == null)
+            {
+               // This can happen if the bridge is shutdown
+               return false;
+            }
 
-         producer = session.createProducer();
+            producer = session.createProducer();
 
-         session.addFailureListener(BridgeImpl.this);
+            session.addFailureListener(BridgeImpl.this);
 
-         session.setSendAcknowledgementHandler(BridgeImpl.this);
+            session.setSendAcknowledgementHandler(BridgeImpl.this);
 
-         setupNotificationConsumer();
+            setupNotificationConsumer();
 
-         active = true;
+            active = true;
 
-         queue.addConsumer(BridgeImpl.this);
+            queue.addConsumer(BridgeImpl.this);
 
-         queue.deliverAsync(executor);
+            queue.deliverAsync(executor);
 
-         return true;
-      }
-      catch (Exception e)
-      {
-         log.warn("Bridge " + name + " is unable to connect to destination. It will be disabled.", e);
+            log.info("Bridge " + name + " is connected to its destination");
 
-         return false;
-      }
+            return true;
+         }
+         catch (HornetQException e)
+         {
+            // the session was created while its server was starting, retry it:
+            if (e.getCode() == HornetQException.SESSION_CREATION_REJECTED)
+            {
+               log.warn("Server is starting, retry to create the session for bridge " + name);
+
+               retry = true;
+               continue;
+            }
+            else
+            {
+               log.warn("Bridge " + name + " is unable to connect to destination. It will be disabled.", e);
+
+               return false;
+            }
+         }      
+         catch (Exception e)
+         {
+            log.warn("Bridge " + name + " is unable to connect to destination. It will be disabled.", e);
+
+            return false;
+         }
+      } while(retry);
+      
+      return false;
    }
 
    // Inner classes -------------------------------------------------