Author: igarashitm Date: 2011-09-22 16:11:24 -0400 (Thu, 22 Sep 2011) New Revision: 11398 Modified: branches/HORNETQ-316_for_2_2_EAP/src/main/org/hornetq/ra/ConnectionFactoryProperties.java branches/HORNETQ-316_for_2_2_EAP/src/main/org/hornetq/ra/HornetQResourceAdapter.java branches/HORNETQ-316_for_2_2_EAP/src/main/org/hornetq/ra/Util.java Log: Adjust ResourceAdapter for discovery changes Modified: branches/HORNETQ-316_for_2_2_EAP/src/main/org/hornetq/ra/ConnectionFactoryProperties.java =================================================================== --- branches/HORNETQ-316_for_2_2_EAP/src/main/org/hornetq/ra/ConnectionFactoryProperties.java 2011-09-22 19:24:22 UTC (rev 11397) +++ branches/HORNETQ-316_for_2_2_EAP/src/main/org/hornetq/ra/ConnectionFactoryProperties.java 2011-09-22 20:11:24 UTC (rev 11398) @@ -100,6 +100,8 @@ private Integer threadPoolMaxSize; + private Map<String, Object> discoveryPluginParameters; + /** * @return the transportType */ @@ -119,6 +121,18 @@ hasBeenUpdated = true; } + public Map<String, Object> getParsedDiscoveryPluginParameters() + { + return discoveryPluginParameters; + } + + public void setParsedDiscoveryPluginParameters(final Map<String, Object> discoveryPluginParameters) + { + this.discoveryPluginParameters = discoveryPluginParameters; + hasBeenUpdated = true; + } + + public void setParsedConnectorClassNames(final List<String> value) { connectorClassName = value; Modified: branches/HORNETQ-316_for_2_2_EAP/src/main/org/hornetq/ra/HornetQResourceAdapter.java =================================================================== --- branches/HORNETQ-316_for_2_2_EAP/src/main/org/hornetq/ra/HornetQResourceAdapter.java 2011-09-22 19:24:22 UTC (rev 11397) +++ branches/HORNETQ-316_for_2_2_EAP/src/main/org/hornetq/ra/HornetQResourceAdapter.java 2011-09-22 20:11:24 UTC (rev 11398) @@ -32,6 +32,7 @@ import javax.transaction.xa.XAResource; import org.hornetq.api.core.DiscoveryGroupConfiguration; +import org.hornetq.api.core.DiscoveryGroupConstants; import org.hornetq.api.core.HornetQException; import org.hornetq.api.core.TransportConfiguration; import org.hornetq.api.core.client.ClientSession; @@ -39,6 +40,7 @@ import org.hornetq.api.core.client.HornetQClient; import org.hornetq.api.jms.HornetQJMSClient; import org.hornetq.api.jms.JMSFactoryType; +import org.hornetq.core.client.impl.SimpleUDPServerLocatorImpl; import org.hornetq.core.logging.Logger; import org.hornetq.jms.client.HornetQConnectionFactory; import org.hornetq.ra.inflow.HornetQActivation; @@ -91,6 +93,11 @@ private String unparsedConnectors; /** + * The discovery plugin properties for resource adapter before parsing + */ + private String unparsedDiscoveryPluginProperties; + + /** * Have the factory been configured */ private final AtomicBoolean configured; @@ -287,6 +294,20 @@ this.raProperties.setHA(ha); } + public String getDiscoveryPluginParameters() + { + return unparsedDiscoveryPluginProperties; + } + + public void setDiscoveryPluginProperties(final String config) + { + if(config != null) + { + this.unparsedDiscoveryPluginProperties = config; + raProperties.setParsedDiscoveryPluginParameters(Util.parseDiscoveryPluginConfig(config)); + } + } + /** * Get the discovery group name * @@ -1417,6 +1438,8 @@ } else if (discoveryAddress != null) { + Map<String,Object> params = new HashMap<String,Object>(); + Integer discoveryPort = overrideProperties.getDiscoveryPort() != null ? overrideProperties.getDiscoveryPort() : getDiscoveryPort(); @@ -1425,8 +1448,6 @@ discoveryPort = HornetQClient.DEFAULT_DISCOVERY_PORT; } - DiscoveryGroupConfiguration groupConfiguration = new DiscoveryGroupConfiguration(discoveryAddress, discoveryPort); - Long refreshTimeout = overrideProperties.getDiscoveryRefreshTimeout() != null ? overrideProperties.getDiscoveryRefreshTimeout() : raProperties.getDiscoveryRefreshTimeout(); if (refreshTimeout == null) @@ -1442,10 +1463,13 @@ initialTimeout = HornetQClient.DEFAULT_DISCOVERY_INITIAL_WAIT_TIMEOUT; } - groupConfiguration.setDiscoveryInitialWaitTimeout(initialTimeout); + params.put(DiscoveryGroupConstants.GROUP_ADDRESS_NAME, discoveryAddress); + params.put(DiscoveryGroupConstants.GROUP_PORT_NAME, discoveryPort); + params.put(DiscoveryGroupConstants.REFRESH_TIMEOUT_NAME, initialTimeout); + params.put(DiscoveryGroupConstants.INITIAL_WAIT_TIMEOUT_NAME, initialTimeout); + + DiscoveryGroupConfiguration groupConfiguration = new DiscoveryGroupConfiguration(SimpleUDPServerLocatorImpl.class.getName(), params, null); - groupConfiguration.setRefreshTimeout(refreshTimeout); - if (ha) { cf = HornetQJMSClient.createConnectionFactoryWithHA(groupConfiguration, JMSFactoryType.XA_CF); @@ -1455,6 +1479,25 @@ cf = HornetQJMSClient.createConnectionFactoryWithoutHA(groupConfiguration, JMSFactoryType.XA_CF); } } + else if (this.unparsedDiscoveryPluginProperties != null) + { + // for another discovery strategy + Map<String, Object> discoveryPluginParams = + overrideConnectionParameters(overrideProperties.getParsedDiscoveryPluginParameters(),raProperties.getParsedDiscoveryPluginParameters()); + + String serverLocatorClassName = (String)discoveryPluginParams.get("server-locator-class"); + + DiscoveryGroupConfiguration groupConfiguration = new DiscoveryGroupConfiguration(serverLocatorClassName, discoveryPluginParams, null); + + if (ha) + { + cf = HornetQJMSClient.createConnectionFactoryWithHA(groupConfiguration, JMSFactoryType.XA_CF); + } + else + { + cf = HornetQJMSClient.createConnectionFactoryWithoutHA(groupConfiguration, JMSFactoryType.XA_CF); + } + } else { throw new IllegalArgumentException("must provide either TransportType or DiscoveryGroupAddress and DiscoveryGroupPort for HornetQ ResourceAdapter Connection Factory"); Modified: branches/HORNETQ-316_for_2_2_EAP/src/main/org/hornetq/ra/Util.java =================================================================== --- branches/HORNETQ-316_for_2_2_EAP/src/main/org/hornetq/ra/Util.java 2011-09-22 19:24:22 UTC (rev 11397) +++ branches/HORNETQ-316_for_2_2_EAP/src/main/org/hornetq/ra/Util.java 2011-09-22 20:11:24 UTC (rev 11398) @@ -199,6 +199,27 @@ return hashtable; } + public static Map<String, Object> parseDiscoveryPluginConfig(final String config) + { + HashMap<String, Object> result = new HashMap<String, Object>(); + + String elements[] = config.split(";"); + + for (String element : elements) + { + String expression[] = element.split("="); + + if (expression.length != 2) + { + throw new IllegalArgumentException("Invalid expression " + element + " at " + config); + } + + result.put(expression[0].trim(), expression[1].trim()); + } + + return result; + } + public static List<Map<String, Object>> parseConfig(final String config) { List<Map<String, Object>> result =new ArrayList<Map<String, Object>>();