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) {