[hornetq-commits] JBoss hornetq SVN: r12037 - in branches/Branch_2_2_AS7/src/main/org/hornetq: ra and 1 other directory.

[do-not-reply at jboss.org]

Author: ataylor
Date: 2012-01-21 05:16:47 -0500 (Sat, 21 Jan 2012)
New Revision: 12037

Modified:
   branches/Branch_2_2_AS7/src/main/org/hornetq/core/protocol/core/impl/ChannelImpl.java
   branches/Branch_2_2_AS7/src/main/org/hornetq/ra/HornetQRAManagedConnectionFactory.java
   branches/Branch_2_2_AS7/src/main/org/hornetq/ra/HornetQResourceAdapter.java
Log:
https://issues.jboss.org/browse/HORNETQ-828

Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/protocol/core/impl/ChannelImpl.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/protocol/core/impl/ChannelImpl.java	2012-01-19 11:41:39 UTC (rev 12036)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/protocol/core/impl/ChannelImpl.java	2012-01-21 10:16:47 UTC (rev 12037)
@@ -349,6 +349,11 @@
          throw new IllegalArgumentException("Cannot find channel with id " + id + " to close");
       }
 
+      if(failingOver)
+      {
+         failingOver = false;
+         failoverCondition.notifyAll();
+      }
       closed = true;
    }
 

Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/ra/HornetQRAManagedConnectionFactory.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/ra/HornetQRAManagedConnectionFactory.java	2012-01-19 11:41:39 UTC (rev 12036)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/ra/HornetQRAManagedConnectionFactory.java	2012-01-21 10:16:47 UTC (rev 12037)
@@ -77,6 +77,12 @@
     */
    private HornetQConnectionFactory connectionFactory;
 
+
+   /**
+    * Connection Factory used if properties are set
+    */
+   private HornetQConnectionFactory recoveryConnectionFactory;
+
    /*
    * The resource recovery if there is one
    * */
@@ -141,7 +147,8 @@
       if (connectionFactory == null)
       {
          connectionFactory = ra.createHornetQConnectionFactory(mcfProperties);
-         resourceRecovery = ra.getRecoveryManager().register(connectionFactory, null, null);
+         recoveryConnectionFactory = ra.createRecoveryHornetQConnectionFactory(mcfProperties);
+         resourceRecovery = ra.getRecoveryManager().register(recoveryConnectionFactory, null, null);
       }
       return cf;
    }
@@ -758,7 +765,9 @@
       if (connectionFactory == null)
       {
          connectionFactory = ra.createHornetQConnectionFactory(mcfProperties);
-         resourceRecovery = ra.getRecoveryManager().register(connectionFactory, null, null);
+         recoveryConnectionFactory = ra.createRecoveryHornetQConnectionFactory(mcfProperties);
+         resourceRecovery = ra.getRecoveryManager().register(recoveryConnectionFactory, null, null);
+         resourceRecovery = ra.getRecoveryManager().register(recoveryConnectionFactory, null, null);
       }
       return connectionFactory;
    }
@@ -810,5 +819,15 @@
       {
          ra.getRecoveryManager().unRegister(resourceRecovery);
       }
+
+      if(connectionFactory != null)
+      {
+         connectionFactory.close();
+      }
+
+      if(recoveryConnectionFactory != null)
+      {
+         recoveryConnectionFactory.close();
+      }
    }
 }

Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/ra/HornetQResourceAdapter.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/ra/HornetQResourceAdapter.java	2012-01-19 11:41:39 UTC (rev 12036)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/ra/HornetQResourceAdapter.java	2012-01-21 10:16:47 UTC (rev 12037)
@@ -106,6 +106,8 @@
    private final Map<ActivationSpec, HornetQActivation> activations;
 
    private HornetQConnectionFactory defaultHornetQConnectionFactory;
+
+   private HornetQConnectionFactory recoveryHornetQConnectionFactory;
    
    private TransactionManager tm;
 
@@ -241,10 +243,10 @@
     */
    public void stop()
    {
-      if (HornetQResourceAdapter.trace)
-      {
-         HornetQResourceAdapter.log.trace("stop()");
-      }
+      //if (HornetQResourceAdapter.trace)
+     // {
+         HornetQResourceAdapter.log.info("stop()*******************************************************************");
+     // }
 
       for (Map.Entry<ActivationSpec, HornetQActivation> entry : activations.entrySet())
       {
@@ -263,8 +265,11 @@
       if (defaultHornetQConnectionFactory != null)
       {
          defaultHornetQConnectionFactory.close();
+      }
 
-         XARecoveryConfig xaRecoveryConfig = new XARecoveryConfig(defaultHornetQConnectionFactory, raProperties.getUserName(), raProperties.getPassword());
+      if(recoveryHornetQConnectionFactory != null)
+      {
+         recoveryHornetQConnectionFactory.close();
       }
 
       recoveryManager.stop();
@@ -1371,7 +1376,8 @@
    protected void setup() throws HornetQException
    {
       defaultHornetQConnectionFactory = createHornetQConnectionFactory(raProperties);
-      recoveryManager.register(defaultHornetQConnectionFactory, raProperties.getUserName(), raProperties.getPassword());
+      recoveryHornetQConnectionFactory = createRecoveryHornetQConnectionFactory(raProperties);
+      recoveryManager.register(recoveryHornetQConnectionFactory, raProperties.getUserName(), raProperties.getPassword());
    }
 
    public Map<ActivationSpec, HornetQActivation> getActivations()
@@ -1545,6 +1551,106 @@
       return cf;
    }
 
+   public HornetQConnectionFactory createRecoveryHornetQConnectionFactory(final ConnectionFactoryProperties overrideProperties)
+   {
+      HornetQConnectionFactory cf;
+      List<String> connectorClassName = overrideProperties.getParsedConnectorClassNames() != null ? overrideProperties.getParsedConnectorClassNames()
+                                                                                    : raProperties.getParsedConnectorClassNames();
+
+      String discoveryAddress = overrideProperties.getDiscoveryAddress() != null ? overrideProperties.getDiscoveryAddress()
+                                                                                : getDiscoveryAddress();
+
+      if (discoveryAddress != null)
+      {
+         Integer discoveryPort = overrideProperties.getDiscoveryPort() != null ? overrideProperties.getDiscoveryPort()
+                                                                              : getDiscoveryPort();
+
+         if(discoveryPort == null)
+         {
+            discoveryPort = HornetQClient.DEFAULT_DISCOVERY_PORT;
+         }
+
+         DiscoveryGroupConfiguration groupConfiguration = new DiscoveryGroupConfiguration(discoveryAddress, discoveryPort);
+
+         if (log.isDebugEnabled())
+         {
+            log.debug("Creating Recovery Connection Factory on the resource adapter for discovery=" + groupConfiguration);
+         }
+
+         Long refreshTimeout = overrideProperties.getDiscoveryRefreshTimeout() != null ? overrideProperties.getDiscoveryRefreshTimeout()
+                                                                    : raProperties.getDiscoveryRefreshTimeout();
+         if (refreshTimeout == null)
+         {
+            refreshTimeout = HornetQClient.DEFAULT_DISCOVERY_REFRESH_TIMEOUT;
+         }
+
+         Long initialTimeout = overrideProperties.getDiscoveryInitialWaitTimeout() != null ? overrideProperties.getDiscoveryInitialWaitTimeout()
+                                                                        : raProperties.getDiscoveryInitialWaitTimeout();
+
+         if(initialTimeout == null)
+         {
+            initialTimeout = HornetQClient.DEFAULT_DISCOVERY_INITIAL_WAIT_TIMEOUT;
+         }
+
+         groupConfiguration.setDiscoveryInitialWaitTimeout(initialTimeout);
+
+         groupConfiguration.setRefreshTimeout(refreshTimeout);
+
+         cf = HornetQJMSClient.createConnectionFactoryWithoutHA(groupConfiguration, JMSFactoryType.XA_CF);
+      }
+      else
+      if (connectorClassName != null)
+      {
+         TransportConfiguration[] transportConfigurations = new TransportConfiguration[connectorClassName.size()];
+
+         List<Map<String, Object>> connectionParams;
+         if(overrideProperties.getParsedConnectorClassNames() != null)
+         {
+            connectionParams = overrideProperties.getParsedConnectionParameters();
+         }
+         else
+         {
+            connectionParams = raProperties.getParsedConnectionParameters();
+         }
+
+         for (int i = 0; i < connectorClassName.size(); i++)
+         {
+            TransportConfiguration tc;
+            if(connectionParams == null || i >= connectionParams.size())
+            {
+               tc = new TransportConfiguration(connectorClassName.get(i));
+               log.debug("No connector params provided using default");
+            }
+            else
+            {
+               tc = new TransportConfiguration(connectorClassName.get(i), connectionParams.get(i));
+            }
+
+            transportConfigurations[i] = tc;
+         }
+
+
+         if (log.isDebugEnabled())
+         {
+            log.debug("Creating Recovery Connection Factory on the resource adapter for transport=" + transportConfigurations);
+         }
+
+         cf = HornetQJMSClient.createConnectionFactoryWithoutHA(JMSFactoryType.XA_CF, transportConfigurations);
+
+      }
+      else
+      {
+         throw new IllegalArgumentException("must provide either TransportType or DiscoveryGroupAddress and DiscoveryGroupPort for HornetQ ResourceAdapter Connection Factory");
+      }
+      setParams(cf, overrideProperties);
+
+      //now make sure we are HA in any way
+
+      cf.setReconnectAttempts(0);
+      cf.setInitialConnectAttempts(0);
+      return cf;
+   }
+
    public Map<String, Object> overrideConnectionParameters(final Map<String, Object> connectionParams,
                                                            final Map<String, Object> overrideConnectionParams)
    {