5:13 a.m.
Author: ataylor
Date: 2012-02-03 06:13:56 -0500 (Fri, 03 Feb 2012)
New Revision: 12071
Modified:
trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/ChannelImpl.java
trunk/hornetq-ra/hornetq-ra-jar/src/main/java/org/hornetq/ra/HornetQRAManagedConnectionFactory.java
trunk/hornetq-ra/hornetq-ra-jar/src/main/java/org/hornetq/ra/HornetQResourceAdapter.java
Log:
https://issues.jboss.org/browse/HORNETQ-828 - fix blocking issue with ra and recovery
Modified:
trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/ChannelImpl.java
===================================================================
---
trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/ChannelImpl.java 2012-02-03
10:55:59 UTC (rev 12070)
+++
trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/ChannelImpl.java 2012-02-03
11:13:56 UTC (rev 12071)
@@ -385,6 +385,10 @@
throw new IllegalArgumentException("Cannot find channel with id " + id
+ " to close");
}
+ if(failingOver)
+ {
+ unlock();
+ }
closed = true;
}
Modified:
trunk/hornetq-ra/hornetq-ra-jar/src/main/java/org/hornetq/ra/HornetQRAManagedConnectionFactory.java
===================================================================
---
trunk/hornetq-ra/hornetq-ra-jar/src/main/java/org/hornetq/ra/HornetQRAManagedConnectionFactory.java 2012-02-03
10:55:59 UTC (rev 12070)
+++
trunk/hornetq-ra/hornetq-ra-jar/src/main/java/org/hornetq/ra/HornetQRAManagedConnectionFactory.java 2012-02-03
11:13:56 UTC (rev 12071)
@@ -76,6 +76,12 @@
*/
private HornetQConnectionFactory connectionFactory;
+
+ /**
+ * Connection Factory used if properties are set
+ */
+ private HornetQConnectionFactory recoveryConnectionFactory;
+
/*
* The resource recovery if there is one
* */
@@ -137,6 +143,12 @@
cm);
}
+ if (connectionFactory == null)
+ {
+ connectionFactory = ra.createHornetQConnectionFactory(mcfProperties);
+ recoveryConnectionFactory =
ra.createRecoveryHornetQConnectionFactory(mcfProperties);
+ resourceRecovery = ra.getRecoveryManager().register(recoveryConnectionFactory,
null, null);
+ }
return cf;
}
@@ -308,6 +320,7 @@
}
this.ra = (HornetQResourceAdapter)ra;
+ this.ra.setManagedConnectionFactory(this);
}
/**
@@ -752,7 +765,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 connectionFactory;
}
@@ -804,5 +818,17 @@
{
ra.getRecoveryManager().unRegister(resourceRecovery);
}
+
+ if(connectionFactory != null)
+ {
+ connectionFactory.close();
+ connectionFactory = null;
+ }
+
+ if(recoveryConnectionFactory != null)
+ {
+ recoveryConnectionFactory.close();
+ recoveryConnectionFactory = null;
+ }
}
}
Modified:
trunk/hornetq-ra/hornetq-ra-jar/src/main/java/org/hornetq/ra/HornetQResourceAdapter.java
===================================================================
---
trunk/hornetq-ra/hornetq-ra-jar/src/main/java/org/hornetq/ra/HornetQResourceAdapter.java 2012-02-03
10:55:59 UTC (rev 12070)
+++
trunk/hornetq-ra/hornetq-ra-jar/src/main/java/org/hornetq/ra/HornetQResourceAdapter.java 2012-02-03
11:13:56 UTC (rev 12071)
@@ -13,10 +13,7 @@
package org.hornetq.ra;
import java.io.Serializable;
-import java.util.HashMap;
-import java.util.Hashtable;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -103,13 +100,17 @@
private final Map<ActivationSpec, HornetQActivation> activations;
private HornetQConnectionFactory defaultHornetQConnectionFactory;
-
+
+ private HornetQConnectionFactory recoveryHornetQConnectionFactory;
+
private TransactionManager tm;
private String unparsedJndiParams;
- RecoveryManager recoveryManager;
+ private RecoveryManager recoveryManager;
+ private final List<HornetQRAManagedConnectionFactory> managedConnectionFactories
= new ArrayList<HornetQRAManagedConnectionFactory>();
+
/**
* Constructor
*/
@@ -245,11 +246,21 @@
activations.clear();
+ for (HornetQRAManagedConnectionFactory managedConnectionFactory :
managedConnectionFactories)
+ {
+ managedConnectionFactory.stop();
+ }
+
+ managedConnectionFactories.clear();
+
if (defaultHornetQConnectionFactory != null)
{
defaultHornetQConnectionFactory.close();
+ }
- XARecoveryConfig xaRecoveryConfig = new
XARecoveryConfig(defaultHornetQConnectionFactory, raProperties.getUserName(),
raProperties.getPassword());
+ if(recoveryHornetQConnectionFactory != null)
+ {
+ recoveryHornetQConnectionFactory.close();
}
recoveryManager.stop();
@@ -1356,7 +1367,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()
@@ -1530,6 +1542,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)
{
@@ -1723,4 +1835,10 @@
cf.setConnectionLoadBalancingPolicyClassName(val5);
}
}
+
+
+ public void setManagedConnectionFactory(HornetQRAManagedConnectionFactory
hornetQRAManagedConnectionFactory)
+ {
+ managedConnectionFactories.add(hornetQRAManagedConnectionFactory);
+ }
}