February 2011 - hornetq-commits - Jboss List Archives

JBoss hornetq SVN: r10199 - in branches/Branch_2_2_EAP: tests/src/org/hornetq/tests/integration and 1 other directory.

by do-not-reply＠jboss.org

Author: clebert.suconic(a)jboss.com Date: 2011-02-11 17:06:46 -0500 (Fri, 11 Feb 2011) New Revision: 10199 Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/postoffice/impl/DuplicateIDCacheImpl.java branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/DuplicateDetectionTest.java Log: JBPAPP-5888 - fixing DuplicateCache bug Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/postoffice/impl/DuplicateIDCacheImpl.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/postoffice/impl/DuplicateIDCacheImpl.java 2011-02-11 03:10:27 UTC (rev 10198) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/postoffice/impl/DuplicateIDCacheImpl.java 2011-02-11 22:06:46 UTC (rev 10199) @@ -292,7 +292,6 @@ public void afterPrepare(final Transaction tx) { - process(); } public void afterRollback(final Transaction tx) @@ -303,14 +302,6 @@ { return null; } - - /* (non-Javadoc) - * @see org.hornetq.core.transaction.TransactionOperation#getDistinctQueues() - */ - public Collection<Queue> getDistinctQueues() - { - return Collections.emptySet(); - } } Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/DuplicateDetectionTest.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/DuplicateDetectionTest.java 2011-02-11 03:10:27 UTC (rev 10198) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/DuplicateDetectionTest.java 2011-02-11 22:06:46 UTC (rev 10199) @@ -52,7 +52,7 @@ public void testSimpleDuplicateDetecion() throws Exception { ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(ServiceTestBase.INVM_CONNECTOR_FACTORY)); - + ClientSessionFactory sf = locator.createSessionFactory(); ClientSession session = sf.createSession(false, true, true); @@ -122,7 +122,7 @@ public void testSimpleDuplicateDetectionWithString() throws Exception { ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(ServiceTestBase.INVM_CONNECTOR_FACTORY)); - + ClientSessionFactory sf = locator.createSessionFactory(); ClientSession session = sf.createSession(false, true, true); @@ -185,14 +185,14 @@ session.close(); sf.close(); - + locator.close(); } public void testCacheSize() throws Exception { ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(ServiceTestBase.INVM_CONNECTOR_FACTORY)); - + ClientSessionFactory sf = locator.createSessionFactory(); ClientSession session = sf.createSession(false, true, true); @@ -351,7 +351,7 @@ public void testTransactedDuplicateDetection1() throws Exception { ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(ServiceTestBase.INVM_CONNECTOR_FACTORY)); - + ClientSessionFactory sf = locator.createSessionFactory(); ClientSession session = sf.createSession(false, false, false); @@ -403,7 +403,7 @@ public void testTransactedDuplicateDetection2() throws Exception { ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(ServiceTestBase.INVM_CONNECTOR_FACTORY)); - + ClientSessionFactory sf = locator.createSessionFactory(); ClientSession session = sf.createSession(false, false, false); @@ -449,7 +449,7 @@ public void testTransactedDuplicateDetection3() throws Exception { ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(ServiceTestBase.INVM_CONNECTOR_FACTORY)); - + ClientSessionFactory sf = locator.createSessionFactory(); ClientSession session = sf.createSession(false, false, false); @@ -511,7 +511,7 @@ public void testEntireTransactionRejected() throws Exception { ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(ServiceTestBase.INVM_CONNECTOR_FACTORY)); - + ClientSessionFactory sf = locator.createSessionFactory(); ClientSession session = sf.createSession(false, false, false); @@ -572,7 +572,7 @@ public void testXADuplicateDetection1() throws Exception { ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(ServiceTestBase.INVM_CONNECTOR_FACTORY)); - + ClientSessionFactory sf = locator.createSessionFactory(); ClientSession session = sf.createSession(true, false, false); @@ -651,7 +651,7 @@ public void testXADuplicateDetection2() throws Exception { ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(ServiceTestBase.INVM_CONNECTOR_FACTORY)); - + ClientSessionFactory sf = locator.createSessionFactory(); ClientSession session = sf.createSession(true, false, false); @@ -732,7 +732,7 @@ public void testXADuplicateDetection3() throws Exception { ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(ServiceTestBase.INVM_CONNECTOR_FACTORY)); - + ClientSessionFactory sf = locator.createSessionFactory(); ClientSession session = sf.createSession(true, false, false); @@ -809,10 +809,82 @@ locator.close(); } + public void testXADuplicateDetectionPrepareAndRollback() throws Exception + { + ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(ServiceTestBase.INVM_CONNECTOR_FACTORY)); + + ClientSessionFactory sf = locator.createSessionFactory(); + + ClientSession session = sf.createSession(true, false, false); + + Xid xid = new XidImpl("xa1".getBytes(), 1, UUIDGenerator.getInstance().generateStringUUID().getBytes()); + + session.start(xid, XAResource.TMNOFLAGS); + + session.start(); + + final SimpleString queueName = new SimpleString("DuplicateDetectionTestQueue"); + + session.createQueue(queueName, queueName, null, false); + + ClientProducer producer = session.createProducer(queueName); + + ClientMessage message = createMessage(session, 0); + SimpleString dupID = new SimpleString("abcdefg"); + message.putBytesProperty(Message.HDR_DUPLICATE_DETECTION_ID, dupID.getData()); + producer.send(message); + + session.end(xid, XAResource.TMSUCCESS); + + session.prepare(xid); + + session.rollback(xid); + + session.close(); + + Xid xid2 = new XidImpl("xa2".getBytes(), 1, UUIDGenerator.getInstance().generateStringUUID().getBytes()); + + session = sf.createSession(true, false, false); + + session.start(xid2, XAResource.TMNOFLAGS); + + session.start(); + + producer = session.createProducer(queueName); + + producer.send(message); + + session.end(xid2, XAResource.TMSUCCESS); + + session.prepare(xid2); + + session.commit(xid2, false); + + session.close(); + + session = sf.createSession(false, false, false); + + session.start(); + + ClientConsumer consumer = session.createConsumer(queueName); + + ClientMessage msgRec = consumer.receive(5000); + assertNotNull(msgRec); + msgRec.acknowledge(); + + session.commit(); + + session.close(); + + sf.close(); + + locator.close(); + } + public void testXADuplicateDetection4() throws Exception { ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(ServiceTestBase.INVM_CONNECTOR_FACTORY)); - + ClientSessionFactory sf = locator.createSessionFactory(); ClientSession session = sf.createSession(true, false, false); @@ -913,7 +985,7 @@ messagingService2.start(); ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(ServiceTestBase.INVM_CONNECTOR_FACTORY)); - + ClientSessionFactory sf = locator.createSessionFactory(); ClientSession session = sf.createSession(false, true, true); @@ -1000,7 +1072,7 @@ messagingService2.start(); ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(ServiceTestBase.INVM_CONNECTOR_FACTORY)); - + ClientSessionFactory sf = locator.createSessionFactory(); ClientSession session = sf.createSession(false, true, true); @@ -1082,7 +1154,7 @@ messagingService2.start(); ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(ServiceTestBase.INVM_CONNECTOR_FACTORY)); - + ClientSessionFactory sf = locator.createSessionFactory(); ClientSession session = sf.createSession(false, true, true); @@ -1175,7 +1247,7 @@ messagingService2.start(); ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(ServiceTestBase.INVM_CONNECTOR_FACTORY)); - + ClientSessionFactory sf = locator.createSessionFactory(); ClientSession session = sf.createSession(false, true, true); @@ -1277,7 +1349,7 @@ messagingService2.start(); ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(ServiceTestBase.INVM_CONNECTOR_FACTORY)); - + ClientSessionFactory sf = locator.createSessionFactory(); ClientSession session = sf.createSession(false, true, true); @@ -1364,7 +1436,7 @@ messagingService2.start(); ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(ServiceTestBase.INVM_CONNECTOR_FACTORY)); - + ClientSessionFactory sf = locator.createSessionFactory(); ClientSession session = sf.createSession(false, false, false); @@ -1453,7 +1525,7 @@ messagingService2.start(); ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(ServiceTestBase.INVM_CONNECTOR_FACTORY)); - + ClientSessionFactory sf = locator.createSessionFactory(); ClientSession session = sf.createSession(false, false, false); @@ -1548,7 +1620,7 @@ messagingService2.start(); ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(ServiceTestBase.INVM_CONNECTOR_FACTORY)); - + ClientSessionFactory sf = locator.createSessionFactory(); ClientSession session = sf.createSession(true, false, false); @@ -1651,7 +1723,7 @@ messagingService2.start(); ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(ServiceTestBase.INVM_CONNECTOR_FACTORY)); - + ClientSessionFactory sf = locator.createSessionFactory(); ClientSession session = sf.createSession(true, false, false); @@ -1752,7 +1824,7 @@ messagingService2.start(); ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(ServiceTestBase.INVM_CONNECTOR_FACTORY)); - + ClientSessionFactory sf = locator.createSessionFactory(); ClientSession session = sf.createSession(true, false, false); @@ -1855,7 +1927,7 @@ messagingService2.start(); ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(ServiceTestBase.INVM_CONNECTOR_FACTORY)); - + ClientSessionFactory sf = locator.createSessionFactory(); ClientSession session = sf.createSession(true, false, false);

14 years, 1 month

1
0
0 / 0

JBoss hornetq SVN: r10198 - in branches/Branch_2_2_EAP: hornetq-rest and 1 other directories.

by do-not-reply＠jboss.org

Author: clebert.suconic(a)jboss.com Date: 2011-02-10 22:10:27 -0500 (Thu, 10 Feb 2011) New Revision: 10198 Modified: branches/Branch_2_2_EAP/build-maven.xml branches/Branch_2_2_EAP/hornetq-rest/pom.xml branches/Branch_2_2_EAP/src/config/common/hornetq-version.properties Log: new release Modified: branches/Branch_2_2_EAP/build-maven.xml =================================================================== --- branches/Branch_2_2_EAP/build-maven.xml 2011-02-10 19:30:34 UTC (rev 10197) +++ branches/Branch_2_2_EAP/build-maven.xml 2011-02-11 03:10:27 UTC (rev 10198) @@ -13,7 +13,7 @@ --> <project default="upload" name="HornetQ"> - <property name="hornetq.version" value="2.2.0.EAP-QA-10185"/> + <property name="hornetq.version" value="2.2.0.EAP-QA-10198"/> <property name="build.dir" value="build"/> <property name="jars.dir" value="${build.dir}/jars"/> Modified: branches/Branch_2_2_EAP/hornetq-rest/pom.xml =================================================================== --- branches/Branch_2_2_EAP/hornetq-rest/pom.xml 2011-02-10 19:30:34 UTC (rev 10197) +++ branches/Branch_2_2_EAP/hornetq-rest/pom.xml 2011-02-11 03:10:27 UTC (rev 10198) @@ -10,7 +10,7 @@ <properties> <resteasy.version>2.0.1.GA</resteasy.version> - <hornetq.version>2.2.0.EAP-QA-10185</hornetq.version> + <hornetq.version>2.2.0.EAP-QA-10198</hornetq.version> </properties> <licenses> Modified: branches/Branch_2_2_EAP/src/config/common/hornetq-version.properties =================================================================== --- branches/Branch_2_2_EAP/src/config/common/hornetq-version.properties 2011-02-10 19:30:34 UTC (rev 10197) +++ branches/Branch_2_2_EAP/src/config/common/hornetq-version.properties 2011-02-11 03:10:27 UTC (rev 10198) @@ -1,4 +1,4 @@ -hornetq.version.versionName=QA_10185 +hornetq.version.versionName=QA_10198 hornetq.version.majorVersion=2 hornetq.version.minorVersion=2 hornetq.version.microVersion=0

14 years, 1 month

1
0
0 / 0

JBoss hornetq SVN: r10197 - branches/Branch_2_2_EAP/src/main/org/hornetq/jms/server/impl.

by do-not-reply＠jboss.org

Author: clebert.suconic(a)jboss.com Date: 2011-02-10 14:30:34 -0500 (Thu, 10 Feb 2011) New Revision: 10197 Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java Log: JBPAPP-5895 - Caching object creating for after the server was activated Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java 2011-02-10 10:25:47 UTC (rev 10196) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java 2011-02-10 19:30:34 UTC (rev 10197) @@ -111,6 +111,9 @@ private final Map<String, List<String>> connectionFactoryJNDI = new HashMap<String, List<String>>(); + // We keep things cached if objects are created while the JMS is not active + private final List<Runnable> cachedCommands = new ArrayList<Runnable>(); + private final HornetQServer server; private JMSManagementService jmsManagementService; @@ -211,6 +214,14 @@ { deploy(); } + + for (Runnable run: cachedCommands) + { + log.info("Running cached command for " + run); + run.run(); + } + + cachedCommands.clear(); } catch (Exception e) @@ -230,7 +241,8 @@ if (registry == null) { - if (!contextSet) context = new InitialContext(); + if (!contextSet) + context = new InitialContext(); registry = new JndiBindingRegistry(context); } @@ -355,7 +367,7 @@ public synchronized void setContext(final Context context) { this.context = context; - + if (registry != null && registry instanceof JndiBindingRegistry) { registry.setContext(context); @@ -377,85 +389,106 @@ final boolean durable, final String... jndi) throws Exception { - checkInitialised(); - boolean added = internalCreateQueue(queueName, selectorString, durable); - - if (!added) + if (active && queues.get(queueName) != null) { return false; } - HornetQDestination destination = queues.get(queueName); - if (destination == null) + runAfterActive(new RunnableException() { - // sanity check. internalCreateQueue should already have done this check - throw new IllegalArgumentException("Queue does not exist"); - } - - ArrayList<String> bindings = new ArrayList<String>(); - - for (String jndiItem : jndi) - { - if (bindToJndi(jndiItem, destination)) + public String toString() { - bindings.add(jndiItem); + return "createQueue for " + queueName; } - } + + public void runException() throws Exception + { + if (internalCreateQueue(queueName, selectorString, durable)) + { - String[] usedJNDI = bindings.toArray(new String[bindings.size()]); - addToBindings(queueJNDI, queueName, usedJNDI); + HornetQDestination destination = queues.get(queueName); + if (destination == null) + { + // sanity check. internalCreateQueue should already have done this check + throw new IllegalArgumentException("Queue does not exist"); + } - if (storeConfig && durable) - { - storage.storeDestination(new PersistedDestination(PersistedType.Queue, queueName, selectorString, durable)); - storage.addJNDI(PersistedType.Queue, queueName, usedJNDI); - } + ArrayList<String> bindings = new ArrayList<String>(); - return added; + for (String jndiItem : jndi) + { + if (bindToJndi(jndiItem, destination)) + { + bindings.add(jndiItem); + } + } + + String[] usedJNDI = bindings.toArray(new String[bindings.size()]); + addToBindings(queueJNDI, queueName, usedJNDI); + + if (storeConfig && durable) + { + storage.storeDestination(new PersistedDestination(PersistedType.Queue, queueName, selectorString, durable)); + storage.addJNDI(PersistedType.Queue, queueName, usedJNDI); + } + } + } + }); + + return true; } public synchronized boolean createTopic(final boolean storeConfig, final String topicName, final String... jndi) throws Exception { - checkInitialised(); - - boolean added = internalCreateTopic(topicName); - - if (!added) + if (active && topics.get(topicName) != null) { return false; } - else + + runAfterActive(new RunnableException() { - HornetQDestination destination = topics.get(topicName); - - if (destination == null) + public String toString() { - // sanity check. internalCreateQueue should already have done this check - throw new IllegalArgumentException("Queue does not exist"); + return "createTopic for " + topicName; } + + public void runException() throws Exception + { + if (internalCreateTopic(topicName)) + { + HornetQDestination destination = topics.get(topicName); - ArrayList<String> bindings = new ArrayList<String>(); + if (destination == null) + { + // sanity check. internalCreateQueue should already have done this check + throw new IllegalArgumentException("Queue does not exist"); + } - for (String jndiItem : jndi) - { - if (bindToJndi(jndiItem, destination)) - { - bindings.add(jndiItem); + ArrayList<String> bindings = new ArrayList<String>(); + + for (String jndiItem : jndi) + { + if (bindToJndi(jndiItem, destination)) + { + bindings.add(jndiItem); + } + } + + String[] usedJNDI = bindings.toArray(new String[bindings.size()]); + addToBindings(topicJNDI, topicName, usedJNDI); + + if (storeConfig) + { + storage.storeDestination(new PersistedDestination(PersistedType.Topic, topicName)); + storage.addJNDI(PersistedType.Topic, topicName, usedJNDI); + } } } + }); - String[] usedJNDI = bindings.toArray(new String[bindings.size()]); - addToBindings(topicJNDI, topicName, usedJNDI); + return true; - if (storeConfig) - { - storage.storeDestination(new PersistedDestination(PersistedType.Topic, topicName)); - storage.addJNDI(PersistedType.Topic, topicName, usedJNDI); - } - - return true; - } } public boolean addTopicToJndi(final String topicName, final String jndiBinding) throws Exception @@ -529,7 +562,8 @@ } if (registry.lookup(jndiBinding) != null) { - throw new HornetQException(HornetQException.ADDRESS_EXISTS, "JNDI " + name + " is already being used by another connection factory"); + throw new HornetQException(HornetQException.ADDRESS_EXISTS, "JNDI " + name + + " is already being used by another connection factory"); } boolean added = bindToJndi(jndiBinding, factory); if (added) @@ -710,9 +744,7 @@ HornetQConnectionFactory cf = connectionFactories.get(name); if (cf == null) { - ConnectionFactoryConfiguration configuration = new ConnectionFactoryConfigurationImpl(name, - ha, - connectorNames); + ConnectionFactoryConfiguration configuration = new ConnectionFactoryConfigurationImpl(name, ha, connectorNames); configuration.setFactoryType(cfType); createConnectionFactory(true, configuration, jndiBindings); } @@ -720,7 +752,7 @@ public synchronized void createConnectionFactory(final String name, final boolean ha, - JMSFactoryType cfType, + JMSFactoryType cfType, final List<String> connectorNames, final String clientID, final long clientFailureCheckPeriod, @@ -757,9 +789,7 @@ HornetQConnectionFactory cf = connectionFactories.get(name); if (cf == null) { - ConnectionFactoryConfiguration configuration = new ConnectionFactoryConfigurationImpl(name, - ha, - connectorNames); + ConnectionFactoryConfiguration configuration = new ConnectionFactoryConfigurationImpl(name, ha, connectorNames); configuration.setClientID(clientID); configuration.setClientFailureCheckPeriod(clientFailureCheckPeriod); configuration.setConnectionTTL(connectionTTL); @@ -790,7 +820,7 @@ configuration.setReconnectAttempts(reconnectAttempts); configuration.setFailoverOnInitialConnection(failoverOnInitialConnection); configuration.setGroupID(groupId); - createConnectionFactory(true, configuration, jndiBindings); + createConnectionFactory(true, configuration, jndiBindings); } } @@ -833,9 +863,7 @@ HornetQConnectionFactory cf = connectionFactories.get(name); if (cf == null) { - ConnectionFactoryConfiguration configuration = new ConnectionFactoryConfigurationImpl(name, - ha, - jndiBindings); + ConnectionFactoryConfiguration configuration = new ConnectionFactoryConfigurationImpl(name, ha, jndiBindings); configuration.setDiscoveryGroupName(discoveryGroupName); configuration.setFactoryType(cfType); configuration.setClientID(clientID); @@ -880,9 +908,7 @@ HornetQConnectionFactory cf = connectionFactories.get(name); if (cf == null) { - ConnectionFactoryConfiguration configuration = new ConnectionFactoryConfigurationImpl(name, - ha, - jndiBindings); + ConnectionFactoryConfiguration configuration = new ConnectionFactoryConfigurationImpl(name, ha, jndiBindings); configuration.setDiscoveryGroupName(discoveryGroupName); createConnectionFactory(true, configuration, jndiBindings); } @@ -890,35 +916,50 @@ public synchronized void createConnectionFactory(final boolean storeConfig, final ConnectionFactoryConfiguration cfConfig, - String... jndi) throws Exception + final String... jndi) throws Exception { - HornetQConnectionFactory cf = internalCreateCF(cfConfig); - - ArrayList<String> bindings = new ArrayList<String>(); - - for (String jndiItem : jndi) + runAfterActive(new RunnableException() { - if (bindToJndi(jndiItem, cf)) + + public String toString() { - bindings.add(jndiItem); + return "createConnectionFactory for " + cfConfig.getName(); } - } + - String[] usedJNDI = bindings.toArray(new String[bindings.size()]); - addToBindings(connectionFactoryJNDI, cfConfig.getName(), usedJNDI); + public void runException() throws Exception + { - if (storeConfig) - { - storage.storeConnectionFactory(new PersistedConnectionFactory(cfConfig)); - storage.addJNDI(PersistedType.ConnectionFactory, cfConfig.getName(), usedJNDI); - } + HornetQConnectionFactory cf = internalCreateCF(cfConfig); + + ArrayList<String> bindings = new ArrayList<String>(); + + for (String jndiItem : jndi) + { + if (bindToJndi(jndiItem, cf)) + { + bindings.add(jndiItem); + } + } + + String[] usedJNDI = bindings.toArray(new String[bindings.size()]); + addToBindings(connectionFactoryJNDI, cfConfig.getName(), usedJNDI); + + if (storeConfig) + { + storage.storeConnectionFactory(new PersistedConnectionFactory(cfConfig)); + storage.addJNDI(PersistedType.ConnectionFactory, cfConfig.getName(), usedJNDI); + } + + } + }); } - + public JMSStorageManager getJMSStorageManager() { return storage; } - + // used on tests only public void replaceStorageManager(JMSStorageManager newStorage) { @@ -1023,11 +1064,15 @@ { if (cfConfig.getDiscoveryGroupName() != null) { - DiscoveryGroupConfiguration groupConfig = server.getConfiguration().getDiscoveryGroupConfigurations().get(cfConfig.getDiscoveryGroupName()); - + DiscoveryGroupConfiguration groupConfig = server.getConfiguration() + .getDiscoveryGroupConfigurations() + .get(cfConfig.getDiscoveryGroupName()); + if (groupConfig == null) { - throw new HornetQException(HornetQException.ILLEGAL_STATE, "Discovery Group '" + cfConfig.getDiscoveryGroupName() + "' doesn't exist on maing config"); + throw new HornetQException(HornetQException.ILLEGAL_STATE, + "Discovery Group '" + cfConfig.getDiscoveryGroupName() + + "' doesn't exist on maing config"); } if (cfConfig.isHA()) @@ -1043,22 +1088,24 @@ { if (cfConfig.getConnectorNames() == null || cfConfig.getConnectorNames().size() == 0) { - throw new HornetQException(HornetQException.ILLEGAL_STATE, "Null Connector name passed to create ConnectionFactory"); + throw new HornetQException(HornetQException.ILLEGAL_STATE, + "Null Connector name passed to create ConnectionFactory"); } - + TransportConfiguration[] configs = new TransportConfiguration[cfConfig.getConnectorNames().size()]; - + int count = 0; for (String name : cfConfig.getConnectorNames()) { - TransportConfiguration connector = server.getConfiguration().getConnectorConfigurations().get(name); + TransportConfiguration connector = server.getConfiguration().getConnectorConfigurations().get(name); if (connector == null) { - throw new HornetQException(HornetQException.ILLEGAL_STATE, "Connector '" + name + "' not found on the main configuration file"); + throw new HornetQException(HornetQException.ILLEGAL_STATE, "Connector '" + name + + "' not found on the main configuration file"); } configs[count++] = connector; } - + if (cfConfig.isHA()) { cf = HornetQJMSClient.createConnectionFactoryWithHA(cfConfig.getFactoryType(), configs); @@ -1099,7 +1146,7 @@ cf.setCompressLargeMessage(cfConfig.isCompressLargeMessages()); cf.setGroupID(cfConfig.getGroupID()); } - + connectionFactories.put(cfConfig.getName(), cf); jmsManagementService.registerConnectionFactory(cfConfig.getName(), cf); @@ -1131,7 +1178,7 @@ { return false; } - + if (registry != null) { for (String jndiBinding : jndiBindings) @@ -1144,7 +1191,7 @@ connectionFactories.remove(name); jmsManagementService.unregisterConnectionFactory(name); - + return true; } @@ -1176,17 +1223,16 @@ checkInitialised(); return server.getHornetQServerControl().listSessions(connectionID); } - - + public String listPreparedTransactionDetailsAsJSON() throws Exception { ResourceManager resourceManager = server.getResourceManager(); Map<Xid, Long> xids = resourceManager.getPreparedTransactionsWithCreationTime(); - if(xids == null || xids.size()==0) + if (xids == null || xids.size() == 0) { return ""; } - + ArrayList<Entry<Xid, Long>> xidsSortedByCreationTime = new ArrayList<Map.Entry<Xid, Long>>(xids.entrySet()); Collections.sort(xidsSortedByCreationTime, new Comparator<Entry<Xid, Long>>() { @@ -1196,24 +1242,22 @@ return (int)(entry1.getValue() - entry2.getValue()); } }); - + JSONArray txDetailListJson = new JSONArray(); for (Map.Entry<Xid, Long> entry : xidsSortedByCreationTime) { Xid xid = entry.getKey(); - TransactionDetail detail = new JMSTransactionDetail(xid, - resourceManager.getTransaction(xid), - entry.getValue()); + TransactionDetail detail = new JMSTransactionDetail(xid, resourceManager.getTransaction(xid), entry.getValue()); txDetailListJson.put(detail.toJSON()); } return txDetailListJson.toString(); } - + public String listPreparedTransactionDetailsAsHTML() throws Exception { ResourceManager resourceManager = server.getResourceManager(); Map<Xid, Long> xids = resourceManager.getPreparedTransactionsWithCreationTime(); - if(xids == null || xids.size() == 0) + if (xids == null || xids.size() == 0) { return "<h3>*** Prepared Transaction Details ***</h3><p>No entry.</p>"; } @@ -1227,18 +1271,16 @@ return (int)(entry1.getValue() - entry2.getValue()); } }); - + StringBuilder html = new StringBuilder(); html.append("<h3>*** Prepared Transaction Details ***</h3>"); - + for (Map.Entry<Xid, Long> entry : xidsSortedByCreationTime) { Xid xid = entry.getKey(); - TransactionDetail detail = new JMSTransactionDetail(xid, - resourceManager.getTransaction(xid), - entry.getValue()); + TransactionDetail detail = new JMSTransactionDetail(xid, resourceManager.getTransaction(xid), entry.getValue()); JSONObject txJson = detail.toJSON(); - + html.append("<table border=\"1\">"); html.append("<tr><th>creation_time</th>"); html.append("<td>" + txJson.get(TransactionDetail.KEY_CREATION_TIME) + "</td>"); @@ -1256,14 +1298,14 @@ html.append("<table border=\"1\" cellspacing=\"0\" cellpadding=\"0\">"); JSONArray msgs = txJson.getJSONArray(TransactionDetail.KEY_TX_RELATED_MESSAGES); - for(int i=0; i<msgs.length(); i++) - { + for (int i = 0; i < msgs.length(); i++) + { JSONObject msgJson = msgs.getJSONObject(i); JSONObject props = msgJson.getJSONObject(TransactionDetail.KEY_MSG_PROPERTIES); StringBuilder propstr = new StringBuilder(); @SuppressWarnings("unchecked") Iterator<String> propkeys = props.keys(); - while(propkeys.hasNext()) + while (propkeys.hasNext()) { String key = propkeys.next(); propstr.append(key); @@ -1271,7 +1313,7 @@ propstr.append(props.get(key)); propstr.append(", "); } - + html.append("<th>operation_type</th>"); html.append("<td>" + msgJson.get(TransactionDetail.KEY_MSG_OP_TYPE) + "</th>"); html.append("<th>message_type</th>"); @@ -1284,12 +1326,10 @@ html.append("</table></td></tr>"); html.append("</table><br/>"); } - + return html.toString(); } - - // Public -------------------------------------------------------- // Private ------------------------------------------------------- @@ -1536,5 +1576,39 @@ return false; } } + + private boolean runAfterActive(RunnableException runnable) throws Exception + { + if (active) + { + runnable.runException(); + return true; + } + else + { + log.info("Caching command for " + runnable + " since the JMS Server is not active yet"); + cachedCommands.add(runnable); + return false; + } + } + + + private abstract class RunnableException implements Runnable + { + public void run() + { + try + { + runException(); + } + catch (Exception e) + { + log.warn(e.getMessage(), e); + } + } + + public abstract void runException() throws Exception; + } + }

14 years, 1 month

1
0
0 / 0

JBoss hornetq SVN: r10196 - in branches/HORNETQ-316: src/main/org/hornetq/integration and 3 other directories.

by do-not-reply＠jboss.org

Author: igarashitm Date: 2011-02-10 05:25:47 -0500 (Thu, 10 Feb 2011) New Revision: 10196 Added: branches/HORNETQ-316/src/main/org/hornetq/integration/discovery/ branches/HORNETQ-316/src/main/org/hornetq/integration/discovery/jgroups/ branches/HORNETQ-316/src/main/org/hornetq/integration/discovery/jgroups/BroadcastGroupConstants.java branches/HORNETQ-316/src/main/org/hornetq/integration/discovery/jgroups/DiscoveryGroupConstants.java branches/HORNETQ-316/src/main/org/hornetq/integration/discovery/jgroups/JGroupsBroadcastGroupImpl.java branches/HORNETQ-316/src/main/org/hornetq/integration/discovery/jgroups/JGroupsDiscoveryGroupImpl.java branches/HORNETQ-316/src/main/org/hornetq/integration/discovery/jgroups/JGroupsServerLocatorImpl.java Modified: branches/HORNETQ-316/build-hornetq.xml branches/HORNETQ-316/pom.xml branches/HORNETQ-316/src/main/org/hornetq/ra/ConnectionFactoryProperties.java branches/HORNETQ-316/src/main/org/hornetq/ra/HornetQResourceAdapter.java Log: added jgroups discovery plugin Modified: branches/HORNETQ-316/build-hornetq.xml =================================================================== --- branches/HORNETQ-316/build-hornetq.xml 2011-02-10 02:23:25 UTC (rev 10195) +++ branches/HORNETQ-316/build-hornetq.xml 2011-02-10 10:25:47 UTC (rev 10196) @@ -69,6 +69,8 @@ <property name="spring.integration.sources.jar.name" value="hornetq-spring-integration-sources.jar"/> <property name="twitter.integration.jar.name" value="hornetq-twitter-integration.jar"/> <property name="twitter.integration.sources.jar.name" value="hornetq-twitter-integration-sources.jar"/> + <property name="jgroups.discovery.jar.name" value="hornetq-jgroups-discovery.jar"/> + <property name="jgroups.discovery.sources.jar.name" value="hornetq-jgroups-discovery.jar"/> <property name="bootstrap.jar.name" value="hornetq-bootstrap.jar"/> <property name="bootstrap.sources.jar.name" value="hornetq-bootstrap-sources.jar"/> <property name="logging.jar.name" value="hornetq-logging.jar"/> @@ -87,6 +89,7 @@ <property name="resources.jar.name" value="hornetq-resources.jar"/> <property name="resources.sources.jar.name" value="hornetq-resources-sources.jar"/> <property name="twitter4j.jar.name" value="twitter4j-core.jar"/> + <property name="jgroups.jar.name" value="jgroups.jar"/> <property name="eap.examples.zip.name" value="hornetq-eap-examples.zip"/>  @@ -99,6 +102,7 @@ <property name="build.jboss.integration.classes.dir" value="${build.dir}/classes/jboss-integration"/> <property name="build.spring.integration.classes.dir" value="${build.dir}/classes/spring-integration"/> <property name="build.twitter.integration.classes.dir" value="${build.dir}/classes/twitter-integration"/> + <property name="build.jgroups.discovery.classes.dir" value="${build.dir}/classes/jgroups-discovery"/> <property name="build.service.classes.dir" value="${build.dir}/classes/service"/> <property name="build.bootstrap.classes.dir" value="${build.dir}/classes/bootstrap"/> <property name="build.logging.classes.dir" value="${build.dir}/classes/logging"/> @@ -223,6 +227,11 @@ <path refid="org.twitter4j.classpath"/> </path> + <path id="jgroups.discovery.compilation.classpath"> + <path location="${build.core.classes.dir}"/> + <path refid="jgroups.jgroups.classpath"/> + </path> + <path id="spring.integration.compilation.classpath"> <path location="${build.core.classes.dir}"/> <path location="${build.jms.classes.dir}"/> @@ -259,6 +268,7 @@ <path refid="bootstrap.compilation.classpath"/> <path refid="junit.junit.classpath"/> <path refid="org.twitter4j.classpath"/> + <path refid="jgroups.jgroups.classpath"/> <path refid="org.springframework.classpath"/> <path location="${build.jars.dir}/${ra.jar.name}"/> <path location="${build.jars.dir}/${jms.jar.name}"/> @@ -267,6 +277,7 @@ <path location="${build.jars.dir}/${logging.jar.name}"/> <path location="${build.jars.dir}/${spring.integration.jar.name}"/> <path location="${build.jars.dir}/${twitter.integration.jar.name}"/> + <path location="${build.jars.dir}/${jgroups.discovery.jar.name}"/> </path> <path id="jms.test.compilation.classpath"> @@ -321,6 +332,7 @@ <path refid="apache.logging.classpath"/> <path refid="org.springframework.classpath"/> <path refid="org.twitter4j.classpath"/> + <path refid="jgroups.jgroups.classpath"/> </path> <path id="emma.unit.test.execution.classpath"> @@ -403,6 +415,7 @@ <mkdir dir="${build.jboss.integration.classes.dir}"/> <mkdir dir="${build.spring.integration.classes.dir}"/> <mkdir dir="${build.twitter.integration.classes.dir}"/> + <mkdir dir="${build.jgroups.discovery.classes.dir}"/> <mkdir dir="${build.service.classes.dir}"/> <mkdir dir="${build.bootstrap.classes.dir}"/> <mkdir dir="${build.logging.classes.dir}"/> @@ -606,6 +619,27 @@ </javac> </target> + <target name="compile-jgroups-discovery" depends="compile-core"> + <javac destdir="${build.jgroups.discovery.classes.dir}" + target="${javac.target}" + source="${javac.source}" + optimize="${javac.optimize}" + debug="${javac.debug}" + depend="${javac.depend}" + verbose="${javac.verbose}" + deprecation="${javac.deprecation}" + includeAntRuntime="${javac.include.ant.runtime}" + includeJavaRuntime="${javac.include.java.runtime}" + encoding="${javac.encoding}" + failonerror="${javac.fail.onerror}"> + <src> + <pathelement path="${src.main.dir}"/> + </src> + <include name="org/hornetq/integration/discovery/jgroups/**/*.java"/> + <classpath refid="jgroups.discovery.compilation.classpath"/> + </javac> + </target> + <target name="compile-spring-integration" depends="compile-core"> <javac destdir="${build.spring.integration.classes.dir}" target="${javac.target}" @@ -771,11 +805,11 @@  <target name="sources-jar" description="create jar files containing source code" - depends="jar-core-sources, jar-core-client-sources, jar-core-client-java5-sources, jar-jms-sources, jar-jms-client-sources, jar-jms-client-java5-sources, jar-jboss-integration-sources, jar-jboss-service-sources, jar-bootstrap-sources, jar-logging-sources, jar-ra-sources, jar-resources-sources, jar-twitter-integration-sources, jar-spring-integration-sources"> + depends="jar-core-sources, jar-core-client-sources, jar-core-client-java5-sources, jar-jms-sources, jar-jms-client-sources, jar-jms-client-java5-sources, jar-jboss-integration-sources, jar-jboss-service-sources, jar-bootstrap-sources, jar-logging-sources, jar-ra-sources, jar-resources-sources, jar-twitter-integration-sources, jar-jgroups-discovery-sources, jar-spring-integration-sources"> </target> <target name="jar" - depends="jar-core, jar-core-client, jar-core-client-java5, jar-jms, jar-jms-client, jar-jms-client-java5, jar-jboss-integration, jar-jboss-service, jar-bootstrap, jar-logging, jar-ra, jar-mc, jar-jnp-client, jar-resources, sources-jar, jar-twitter-integration, jar-spring-integration, jar-rest"> + depends="jar-core, jar-core-client, jar-core-client-java5, jar-jms, jar-jms-client, jar-jms-client-java5, jar-jboss-integration, jar-jboss-service, jar-bootstrap, jar-logging, jar-ra, jar-mc, jar-jnp-client, jar-resources, sources-jar, jar-twitter-integration, jar-jgroups-discovery, jar-spring-integration, jar-rest"> </target> <target name="jar-jnp-client" depends="init"> @@ -935,6 +969,24 @@ </jar> </target> + <target name="jar-jgroups-discovery" depends="compile-jgroups-discovery"> + + <jar jarfile="${build.jars.dir}/${jgroups.discovery.jar.name}"> + <fileset dir="${build.jgroups.discovery.classes.dir}" includes="**"/> + </jar> + + </target> + + <target name="jar-jgroups-discovery-sources"> + + <jar jarfile="${build.jars.dir}/${jgroups.discovery.sources.jar.name}"> + <fileset dir="${src.main.dir}"> + <include name="org/hornetq/integration/discovery/jgroups/**/*.java"/> + </fileset> + </jar> + + </target> + <target name="jar-spring-integration" depends="compile-spring-integration"> <jar jarfile="${build.jars.dir}/${spring.integration.jar.name}"> @@ -1267,6 +1319,7 @@ <include name="${jnp.client.jar.name}"/> <include name="${spring.integration.jar.name}"/> <include name="${twitter.integration.jar.name}"/> + <include name="${jgroups.discovery.jar.name}"/> </fileset> <fileset dir="${org.jboss.naming.lib}"> <include name="jnpserver.jar"/> @@ -1277,6 +1330,7 @@ </copy> <copy file="${org.jboss.netty.lib}/${netty.jar.name}" tofile="${build.distro.lib.dir}/netty.jar"/> <copy file="${org.twitter4j.lib}/${twitter4j.jar.name}" tofile="${build.distro.lib.dir}/${twitter4j.jar.name}"/> + <copy file="${jgroups.jgroups.lib}/${jgroups.jar.name}" tofile="${build.distro.lib.dir}/${jgroups.jar.name}"/> <copy todir="${build.distro.config.dir}"> <fileset dir="${src.config.dir}"> <include name="*.xml"/> Modified: branches/HORNETQ-316/pom.xml =================================================================== --- branches/HORNETQ-316/pom.xml 2011-02-10 02:23:25 UTC (rev 10195) +++ branches/HORNETQ-316/pom.xml 2011-02-10 10:25:47 UTC (rev 10196) @@ -255,6 +255,12 @@ <artifactId>twitter4j-core</artifactId> <version>2.1.6</version> </dependency> +  + <dependency> + <groupId>jgroups</groupId> + <artifactId>jgroups</artifactId> + <version>2.3</version> + </dependency>  <dependency> <groupId>junit</groupId> Added: branches/HORNETQ-316/src/main/org/hornetq/integration/discovery/jgroups/BroadcastGroupConstants.java =================================================================== --- branches/HORNETQ-316/src/main/org/hornetq/integration/discovery/jgroups/BroadcastGroupConstants.java (rev 0) +++ branches/HORNETQ-316/src/main/org/hornetq/integration/discovery/jgroups/BroadcastGroupConstants.java 2011-02-10 10:25:47 UTC (rev 10196) @@ -0,0 +1,27 @@ +/* + * Copyright 2010 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.integration.discovery.jgroups; + +/** + * A BroadcastGroupConstants + * + * @author "<a href=\"tm.igarashi(a)gmail.com\">Tomohisa Igarashi</a>" + * + * + */ +public class BroadcastGroupConstants +{ + public static final String JGROUPS_CONFIGURATION_FILE_NAME = "jgroups-configuration-file"; + public static final Object BROADCAST_PERIOD_NAME = "broadcast-period"; +} Property changes on: branches/HORNETQ-316/src/main/org/hornetq/integration/discovery/jgroups/BroadcastGroupConstants.java ___________________________________________________________________ Name: svn:mime-type + text/plain Added: branches/HORNETQ-316/src/main/org/hornetq/integration/discovery/jgroups/DiscoveryGroupConstants.java =================================================================== --- branches/HORNETQ-316/src/main/org/hornetq/integration/discovery/jgroups/DiscoveryGroupConstants.java (rev 0) +++ branches/HORNETQ-316/src/main/org/hornetq/integration/discovery/jgroups/DiscoveryGroupConstants.java 2011-02-10 10:25:47 UTC (rev 10196) @@ -0,0 +1,28 @@ +/* + * Copyright 2010 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.integration.discovery.jgroups; + +/** + * A DiscoveryGroupConstants + * + * @author "<a href=\"tm.igarashi(a)gmail.com\">Tomohisa Igarashi</a>" + * + * + */ +public class DiscoveryGroupConstants +{ + public static final String JGROUPS_CONFIGURATION_FILE_NAME = "jgroups-configuration-filename"; + public static final String INITIAL_WAIT_TIMEOUT_NAME = "initial-wait-timeout"; + public static final String REFRESH_TIMEOUT_NAME = "refresh-timeout"; +} Added: branches/HORNETQ-316/src/main/org/hornetq/integration/discovery/jgroups/JGroupsBroadcastGroupImpl.java =================================================================== --- branches/HORNETQ-316/src/main/org/hornetq/integration/discovery/jgroups/JGroupsBroadcastGroupImpl.java (rev 0) +++ branches/HORNETQ-316/src/main/org/hornetq/integration/discovery/jgroups/JGroupsBroadcastGroupImpl.java 2011-02-10 10:25:47 UTC (rev 10196) @@ -0,0 +1,236 @@ +/* + * Copyright 2010 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.integration.discovery.jgroups; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; + +import org.hornetq.api.core.HornetQBuffer; +import org.hornetq.api.core.HornetQBuffers; +import org.hornetq.api.core.SimpleString; +import org.hornetq.api.core.TransportConfiguration; +import org.hornetq.api.core.management.NotificationType; +import org.hornetq.core.config.BroadcastGroupConfiguration; +import org.hornetq.core.logging.Logger; +import org.hornetq.core.server.cluster.BroadcastGroup; +import org.hornetq.core.server.management.Notification; +import org.hornetq.core.server.management.NotificationService; +import org.hornetq.utils.ConfigurationHelper; +import org.hornetq.utils.TypedProperties; +import org.hornetq.utils.UUIDGenerator; +import org.jgroups.JChannel; +import org.jgroups.Message; + +/** + * A JGroupsBroadcastGroupImpl + * + * @author "<a href=\"tm.igarashi(a)gmail.com\">Tomohisa Igarashi</a>" + * + * + */ +public class JGroupsBroadcastGroupImpl implements BroadcastGroup, Runnable +{ + private static final Logger log = Logger.getLogger(JGroupsBroadcastGroupImpl.class); + + private final String nodeID; + + private final String name; + + private final BroadcastGroupConfiguration broadcastGroupConfiguration; + + private final List<TransportConfiguration> connectors = new ArrayList<TransportConfiguration>(); + + private String jgroupsConfigurationFileName; + + private JChannel broadcastChannel; + + private boolean started; + + private ScheduledFuture<?> future; + + private boolean active; + + // Each broadcast group has a unique id - we use this to detect when more than one group broadcasts the same node id + // on the network which would be an error + private final String uniqueID; + + private NotificationService notificationService; + + public JGroupsBroadcastGroupImpl(final String nodeID, + final String name, + final boolean active, + final BroadcastGroupConfiguration config) + { + this.nodeID = nodeID; + + this.name = name; + + this.active = active; + + this.broadcastGroupConfiguration = config; + + uniqueID = UUIDGenerator.getInstance().generateStringUUID(); + } + + public void setNotificationService(NotificationService notificationService) + { + this.notificationService = notificationService; + } + + public void start() throws Exception + { + if (started) + { + return; + } + + Map<String,Object> params = this.broadcastGroupConfiguration.getParams(); + this.jgroupsConfigurationFileName = ConfigurationHelper.getStringProperty(DiscoveryGroupConstants.JGROUPS_CONFIGURATION_FILE_NAME, null, params); + + this.broadcastChannel = new JChannel(Thread.currentThread().getContextClassLoader().getResource(this.jgroupsConfigurationFileName)); + + this.broadcastChannel.connect(this.name); + + started = true; + + if (notificationService != null) + { + TypedProperties props = new TypedProperties(); + props.putSimpleStringProperty(new SimpleString("name"), new SimpleString(name)); + Notification notification = new Notification(nodeID, NotificationType.BROADCAST_GROUP_STARTED, props); + notificationService.sendNotification(notification); + } + } + + public void stop() throws Exception + { + if (!started) + { + return; + } + + if (future != null) + { + future.cancel(false); + } + + started = false; + + if (notificationService != null) + { + TypedProperties props = new TypedProperties(); + props.putSimpleStringProperty(new SimpleString("name"), new SimpleString(name)); + Notification notification = new Notification(nodeID, NotificationType.BROADCAST_GROUP_STOPPED, props); + try + { + notificationService.sendNotification(notification); + } + catch (Exception e) + { + JGroupsBroadcastGroupImpl.log.warn("unable to send notification when broadcast group is stopped", e); + } + } + + } + + public boolean isStarted() + { + return this.started; + } + + public String getName() + { + return this.name; + } + + public void addConnector(TransportConfiguration tcConfig) + { + this.connectors.add(tcConfig); + } + + public void removeConnector(TransportConfiguration tcConfig) + { + this.connectors.remove(tcConfig); + } + + public int size() + { + return this.connectors.size(); + } + + public void activate() + { + this.active = true; + } + + public void broadcastConnectors() throws Exception + { + if (!active) + { + return; + } + + HornetQBuffer buff = HornetQBuffers.dynamicBuffer(4096); + + buff.writeString(nodeID); + + buff.writeString(uniqueID); + + buff.writeInt(connectors.size()); + + for (TransportConfiguration tcConfig : connectors) + { + tcConfig.encode(buff); + } + + byte[] data = buff.toByteBuffer().array(); + + Message msg = new Message(); + + msg.setBuffer(data); + + this.broadcastChannel.send(msg); + } + + public void run() + { + if (!started) + { + return; + } + + try + { + broadcastConnectors(); + } + catch (Exception e) + { + JGroupsBroadcastGroupImpl.log.error("Failed to broadcast connector configs", e); + } + } + + public void schedule(ScheduledExecutorService scheduler) + { + Map<String,Object> params = broadcastGroupConfiguration.getParams(); + + this.future = scheduler.scheduleWithFixedDelay(this, + 0L, + Long.parseLong((String)params.get(BroadcastGroupConstants.BROADCAST_PERIOD_NAME)), + TimeUnit.MILLISECONDS); + } +} Property changes on: branches/HORNETQ-316/src/main/org/hornetq/integration/discovery/jgroups/JGroupsBroadcastGroupImpl.java ___________________________________________________________________ Name: svn:mime-type + text/plain Added: branches/HORNETQ-316/src/main/org/hornetq/integration/discovery/jgroups/JGroupsDiscoveryGroupImpl.java =================================================================== --- branches/HORNETQ-316/src/main/org/hornetq/integration/discovery/jgroups/JGroupsDiscoveryGroupImpl.java (rev 0) +++ branches/HORNETQ-316/src/main/org/hornetq/integration/discovery/jgroups/JGroupsDiscoveryGroupImpl.java 2011-02-10 10:25:47 UTC (rev 10196) @@ -0,0 +1,358 @@ +/* + * Copyright 2010 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.integration.discovery.jgroups; + +import java.net.URL; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.Iterator; +import java.util.List; +import java.util.Map; + +import org.hornetq.api.core.HornetQBuffer; +import org.hornetq.api.core.HornetQBuffers; +import org.hornetq.api.core.SimpleString; +import org.hornetq.api.core.TransportConfiguration; +import org.hornetq.api.core.management.NotificationType; +import org.hornetq.core.cluster.DiscoveryEntry; +import org.hornetq.core.cluster.DiscoveryGroup; +import org.hornetq.core.cluster.DiscoveryListener; +import org.hornetq.core.logging.Logger; +import org.hornetq.core.server.management.Notification; +import org.hornetq.core.server.management.NotificationService; +import org.hornetq.utils.TypedProperties; +import org.jgroups.JChannel; +import org.jgroups.Message; +import org.jgroups.ReceiverAdapter; + +/** + * A JGroupsDiscoveryGroupImpl + * + * @author "<a href=\"tm.igarashi(a)gmail.com\">Tomohisa Igarashi</a>" + * + * + */ +public class JGroupsDiscoveryGroupImpl extends ReceiverAdapter implements DiscoveryGroup +{ + private static final Logger log = Logger.getLogger(JGroupsDiscoveryGroupImpl.class); + + private final List<DiscoveryListener> listeners = new ArrayList<DiscoveryListener>(); + + private final String name; + + private final URL configURL; + + private final String nodeID; + + private volatile boolean started; + + private boolean received; + + private final Object waitLock = new Object(); + + private final Map<String, DiscoveryEntry> connectors = new HashMap<String, DiscoveryEntry>(); + + private final long timeout; + + private final Map<String, String> uniqueIDMap = new HashMap<String, String>(); + + private JChannel discoveryChannel; + + private NotificationService notificationService; + + public JGroupsDiscoveryGroupImpl(final String nodeID, + final String name, + final URL confURL, + final long timeout) + { + this.nodeID = nodeID; + this.name = name; + this.configURL = confURL; + this.timeout = timeout; + } + + public void setNotificationService(NotificationService notificationService) + { + this.notificationService = notificationService; + } + + public void start() throws Exception + { + if (started) + { + return; + } + + try + { + this.discoveryChannel = new JChannel(configURL); + + this.discoveryChannel.setReceiver(this); + + this.discoveryChannel.connect(this.name); + } + catch(Exception e) + { + log.error("Failed to join jgroups channel", e); + return; + } + + started = true; + + if (notificationService != null) + { + TypedProperties props = new TypedProperties(); + + props.putSimpleStringProperty(new SimpleString("name"), new SimpleString(name)); + + Notification notification = new Notification(nodeID, NotificationType.DISCOVERY_GROUP_STARTED, props); + + notificationService.sendNotification(notification); + } + } + + public void stop() throws Exception + { + synchronized (this) + { + if (!started) + { + return; + } + + started = false; + } + + synchronized (waitLock) + { + waitLock.notify(); + } + + this.discoveryChannel.shutdown(); + + this.discoveryChannel = null; + + if (notificationService != null) + { + TypedProperties props = new TypedProperties(); + props.putSimpleStringProperty(new SimpleString("name"), new SimpleString(name)); + Notification notification = new Notification(nodeID, NotificationType.DISCOVERY_GROUP_STOPPED, props); + try + { + notificationService.sendNotification(notification); + } + catch (Exception e) + { + JGroupsDiscoveryGroupImpl.log.warn("unable to send notification when discovery group is stopped", e); + } + } + } + + public String getName() + { + return this.name; + } + + public List<DiscoveryEntry> getDiscoveryEntries() + { + List<DiscoveryEntry> list = new ArrayList<DiscoveryEntry>(); + + list.addAll(connectors.values()); + + return list; + } + + public boolean isStarted() + { + return this.started; + } + + public boolean waitForBroadcast(long timeout) + { + synchronized (waitLock) + { + long start = System.currentTimeMillis(); + + long toWait = timeout; + + while (started && !received && (toWait > 0 || timeout == 0)) + { + try + { + waitLock.wait(toWait); + } + catch (InterruptedException e) + { + } + + if (timeout != 0) + { + long now = System.currentTimeMillis(); + + toWait -= now - start; + + start = now; + } + } + + boolean ret = received; + + received = false; + + return ret; + } + } + + @Override + public void receive(Message msg) + { + if(!started) + { + return; + } + + HornetQBuffer buffer = HornetQBuffers.wrappedBuffer(msg.getBuffer()); + + String originatingNodeID = buffer.readString(); + + String uniqueID = buffer.readString(); + + checkUniqueID(originatingNodeID, uniqueID); + + if (nodeID.equals(originatingNodeID)) + { + if (checkExpiration()) + { + callListeners(); + } + + // Ignore traffic from own node + return; + } + + int size = buffer.readInt(); + + boolean changed = false; + + synchronized (this) + { + for (int i = 0; i < size; i++) + { + TransportConfiguration connector = new TransportConfiguration(); + + connector.decode(buffer); + + DiscoveryEntry entry = new DiscoveryEntry(originatingNodeID, connector, System.currentTimeMillis()); + + DiscoveryEntry oldVal = connectors.put(originatingNodeID, entry); + + if (oldVal == null) + { + changed = true; + } + } + + changed = changed || checkExpiration(); + } + + if (changed) + { + callListeners(); + } + + synchronized (waitLock) + { + received = true; + + waitLock.notify(); + } + } + + public void registerListener(DiscoveryListener listener) + { + listeners.add(listener); + + if (!connectors.isEmpty()) + { + listener.connectorsChanged(); + } + } + + public void unregisterListener(DiscoveryListener listener) + { + listeners.remove(listener); + } + + private void callListeners() + { + for (DiscoveryListener listener : listeners) + { + try + { + listener.connectorsChanged(); + } + catch (Throwable t) + { + // Catch it so exception doesn't prevent other listeners from running + JGroupsDiscoveryGroupImpl.log.error("Failed to call discovery listener", t); + } + } + } + + private void checkUniqueID(final String originatingNodeID, final String uniqueID) + { + String currentUniqueID = uniqueIDMap.get(originatingNodeID); + + if (currentUniqueID == null) + { + uniqueIDMap.put(originatingNodeID, uniqueID); + } + else + { + if (!currentUniqueID.equals(uniqueID)) + { + log.warn("There are more than one servers on the network broadcasting the same node id. " + "You will see this message exactly once (per node) if a node is restarted, in which case it can be safely " + + "ignored. But if it is logged continuously it means you really do have more than one node on the same network " + + "active concurrently with the same node id. This could occur if you have a backup node active at the same time as " + + "its live node. nodeID=" + originatingNodeID); + uniqueIDMap.put(originatingNodeID, uniqueID); + } + } + } + + private boolean checkExpiration() + { + boolean changed = false; + long now = System.currentTimeMillis(); + + Iterator<Map.Entry<String, DiscoveryEntry>> iter = connectors.entrySet().iterator(); + + // Weed out any expired connectors + + while (iter.hasNext()) + { + Map.Entry<String, DiscoveryEntry> entry = iter.next(); + + if (entry.getValue().getLastUpdate() + timeout <= now) + { + iter.remove(); + + changed = true; + } + } + + return changed; + } + +} Property changes on: branches/HORNETQ-316/src/main/org/hornetq/integration/discovery/jgroups/JGroupsDiscoveryGroupImpl.java ___________________________________________________________________ Name: svn:mime-type + text/plain Added: branches/HORNETQ-316/src/main/org/hornetq/integration/discovery/jgroups/JGroupsServerLocatorImpl.java =================================================================== --- branches/HORNETQ-316/src/main/org/hornetq/integration/discovery/jgroups/JGroupsServerLocatorImpl.java (rev 0) +++ branches/HORNETQ-316/src/main/org/hornetq/integration/discovery/jgroups/JGroupsServerLocatorImpl.java 2011-02-10 10:25:47 UTC (rev 10196) @@ -0,0 +1,333 @@ +/* + * Copyright 2010 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.integration.discovery.jgroups; + +import java.lang.reflect.Array; +import java.util.List; +import java.util.Map; +import java.util.concurrent.Executor; + +import org.hornetq.api.core.DiscoveryGroupConfiguration; +import org.hornetq.api.core.HornetQException; +import org.hornetq.api.core.TransportConfiguration; +import org.hornetq.api.core.client.ClientSessionFactory; +import org.hornetq.api.core.client.HornetQClient; +import org.hornetq.core.client.impl.AbstractServerLocator; +import org.hornetq.core.client.impl.ClientSessionFactoryImpl; +import org.hornetq.core.client.impl.ClientSessionFactoryInternal; +import org.hornetq.core.cluster.DiscoveryEntry; +import org.hornetq.core.cluster.DiscoveryGroup; +import org.hornetq.core.cluster.DiscoveryListener; +import org.hornetq.core.config.impl.ConfigurationImpl; +import org.hornetq.core.logging.Logger; +import org.hornetq.utils.ConfigurationHelper; + +/** + * A JGroupsServerLocatorImpl + * + * @author "<a href=\"tm.igarashi(a)gmail.com\">Tomohisa Igarashi</a>" + * + * + */ +public class JGroupsServerLocatorImpl extends AbstractServerLocator implements DiscoveryListener +{ + private static final long serialVersionUID = 1720602999991968346L; + + private static final Logger log = Logger.getLogger(JGroupsServerLocatorImpl.class); + + private String discoveryGroupName; + + private String jgroupsConfigurationFileName; + + private long initialWaitTimeout; + + private long refreshTimeout; + + private DiscoveryGroup discoveryGroup; + + private volatile boolean closing; + + private synchronized void initialise() throws Exception { + if (!isReadOnly()) + { + setThreadPools(); + + instantiateLoadBalancingPolicy(); + + this.discoveryGroupName = getDiscoveryGroupConfiguration().getName(); + + Map<String,Object> params = getDiscoveryGroupConfiguration().getParams(); + + this.initialWaitTimeout = ConfigurationHelper.getLongProperty(DiscoveryGroupConstants.INITIAL_WAIT_TIMEOUT_NAME, HornetQClient.DEFAULT_DISCOVERY_INITIAL_WAIT_TIMEOUT, params); + + this.refreshTimeout = ConfigurationHelper.getLongProperty(DiscoveryGroupConstants.REFRESH_TIMEOUT_NAME, ConfigurationImpl.DEFAULT_BROADCAST_REFRESH_TIMEOUT, params); + + this.jgroupsConfigurationFileName = ConfigurationHelper.getStringProperty(DiscoveryGroupConstants.JGROUPS_CONFIGURATION_FILE_NAME, null, params); + + this.discoveryGroup = new JGroupsDiscoveryGroupImpl(getNodeID(), + this.discoveryGroupName, + Thread.currentThread().getContextClassLoader().getResource(this.jgroupsConfigurationFileName), + this.refreshTimeout); + + this.discoveryGroup.registerListener(this); + + this.discoveryGroup.start(); + + setReadOnly(true); + } + } + + public JGroupsServerLocatorImpl(boolean useHA, DiscoveryGroupConfiguration discoveryGroupConfiguration) + { + super(useHA, discoveryGroupConfiguration); + } + + public void start(Executor executor) throws Exception + { + initialise(); + + executor.execute(new Runnable() + { + public void run() + { + try + { + connect(); + } + catch (Exception e) + { + if (!closing) + { + log.warn("did not connect the cluster connection to other nodes", e); + } + } + } + }); + } + + public ClientSessionFactory connect() throws Exception + { + ClientSessionFactoryInternal sf; + + // wait for discovery group to get the list of initial connectors + sf = (ClientSessionFactoryInternal)createSessionFactory(); + + addFactory(sf); + return sf; + } + + public ClientSessionFactory createSessionFactory() throws Exception + { + if (isClosed()) + { + throw new IllegalStateException("Cannot create session factory, server locator is closed (maybe it has been garbage collected)"); + } + + try + { + initialise(); + } + catch (Exception e) + { + throw new HornetQException(HornetQException.INTERNAL_ERROR, "Failed to initialise session factory", e); + } + + if (getInitialConnectors() == null) + { + // Wait for an initial broadcast to give us at least one node in the cluster + long timeout = isClusterConnection() ? 0 : this.initialWaitTimeout; + boolean ok = discoveryGroup.waitForBroadcast(timeout); + + if (!ok) + { + throw new HornetQException(HornetQException.CONNECTION_TIMEDOUT, + "Timed out waiting to receive initial broadcast from cluster"); + } + } + + ClientSessionFactoryInternal factory = null; + + synchronized (this) + { + boolean retry; + int attempts = 0; + do + { + retry = false; + + TransportConfiguration tc = selectConnector(); + + // try each factory in the list until we find one which works + + try + { + factory = new ClientSessionFactoryImpl(this, + tc, + getCallTimeout(), + getClientFailureCheckPeriod(), + getConnectionTTL(), + getRetryInterval(), + getRetryIntervalMultiplier(), + getMaxRetryInterval(), + getReconnectAttempts(), + getThreadPool(), + getScheduledThreadPool(), + getInterceptors()); + factory.connect(getInitialConnectAttempts(), isFailoverOnInitialConnection()); + } + catch (HornetQException e) + { + factory.close(); + factory = null; + if (e.getCode() == HornetQException.NOT_CONNECTED) + { + attempts++; + + if (attempts == getConnectorLength()) + { + throw new HornetQException(HornetQException.NOT_CONNECTED, + "Cannot connect to server(s). Tried with all available servers."); + } + retry = true; + } + else + { + throw e; + } + } + } + while (retry); + + if (isHA()) + { + long toWait = 30000; + long start = System.currentTimeMillis(); + while (!isReceivedTopology() && toWait > 0) + { + // Now wait for the topology + + try + { + wait(toWait); + } + catch (InterruptedException ignore) + { + } + + long now = System.currentTimeMillis(); + + toWait -= now - start; + + start = now; + } + + if (toWait <= 0) + { + throw new HornetQException(HornetQException.CONNECTION_TIMEDOUT, + "Timed out waiting to receive cluster topology"); + } + } + + addFactory(factory); + + return factory; + } + } + + public ClientSessionFactory createSessionFactory(TransportConfiguration transportConfiguration) throws Exception + { + if (isClosed()) + { + throw new IllegalStateException("Cannot create session factory, server locator is closed (maybe it has been garbage collected)"); + } + + try + { + initialise(); + } + catch (Exception e) + { + throw new HornetQException(HornetQException.INTERNAL_ERROR, "Failed to initialise session factory", e); + } + + ClientSessionFactoryInternal factory = new ClientSessionFactoryImpl(this, + transportConfiguration, + getCallTimeout(), + getClientFailureCheckPeriod(), + getConnectionTTL(), + getRetryInterval(), + getRetryIntervalMultiplier(), + getMaxRetryInterval(), + getReconnectAttempts(), + getThreadPool(), + getScheduledThreadPool(), + getInterceptors()); + + factory.connect(getReconnectAttempts(), isFailoverOnInitialConnection()); + + addFactory(factory); + + return factory; + } + + public void close() + { + if (isClosed()) + { + return; + } + + closing = true; + + try + { + this.discoveryGroup.stop(); + } + catch (Exception e) + { + log.error("Failed to stop discovery group", e); + } + + super.close(); + } + + public synchronized void connectorsChanged() + { + List<DiscoveryEntry> newConnectors = discoveryGroup.getDiscoveryEntries(); + + TransportConfiguration[] initialConnectors = (TransportConfiguration[])Array.newInstance(TransportConfiguration.class, + newConnectors.size()); + int count = 0; + for (DiscoveryEntry entry : newConnectors) + { + initialConnectors[count++] = entry.getConnector(); + } + + if (isHA() && isClusterConnection() && !isReceivedTopology() && initialConnectors.length > 0) + { + // FIXME the node is alone in the cluster. We create a connection to the new node + // to trigger the node notification to form the cluster. + try + { + connect(); + } + catch (Exception e) + { + e.printStackTrace(); // To change body of catch statement use File | Settings | File Templates. + } + } + + setInitialConnectors(initialConnectors); + } +} Modified: branches/HORNETQ-316/src/main/org/hornetq/ra/ConnectionFactoryProperties.java =================================================================== --- branches/HORNETQ-316/src/main/org/hornetq/ra/ConnectionFactoryProperties.java 2011-02-10 02:23:25 UTC (rev 10195) +++ branches/HORNETQ-316/src/main/org/hornetq/ra/ConnectionFactoryProperties.java 2011-02-10 10:25:47 UTC (rev 10196) @@ -103,6 +103,8 @@ private Integer threadPoolMaxSize; + private Map<String, Object> discoveryPluginParameters; + /** * @return the transportType */ @@ -122,6 +124,17 @@ hasBeenUpdated = true; } + public Map<String, Object> getParsedDiscoveryPluginParameters() + { + return discoveryPluginParameters; + } + + public void setParsedDiscoveryPluginParameters(final Map<String, Object> discoveryPluginParameters) + { + this.discoveryPluginParameters = discoveryPluginParameters; + hasBeenUpdated = true; + } + public void setConnectorClassName(final String value) { connectorClassName = value; Modified: branches/HORNETQ-316/src/main/org/hornetq/ra/HornetQResourceAdapter.java =================================================================== --- branches/HORNETQ-316/src/main/org/hornetq/ra/HornetQResourceAdapter.java 2011-02-10 02:23:25 UTC (rev 10195) +++ branches/HORNETQ-316/src/main/org/hornetq/ra/HornetQResourceAdapter.java 2011-02-10 10:25:47 UTC (rev 10196) @@ -85,6 +85,11 @@ private String unparsedProperties; /** + * The discovery plugin properties for resource adapter before parsing + */ + private String unparsedDiscoveryPluginProperties; + + /** * Have the factory been configured */ private final AtomicBoolean configured; @@ -262,6 +267,20 @@ } } + public String getDiscoveryPluginParameters() + { + return unparsedDiscoveryPluginProperties; + } + + public void setDiscoveryPluginProperties(final String config) + { + if(config != null) + { + this.unparsedDiscoveryPluginProperties = config; + raProperties.setParsedDiscoveryPluginParameters(Util.parseConfig(config)); + } + } + /** * Get the discovery group name * @@ -1405,7 +1424,6 @@ } else if (discoveryAddress != null) { - // FIXME make discovery stategy pluggable with configuration Map<String,Object> params = new HashMap<String,Object>(); Integer discoveryPort = overrideProperties.getDiscoveryPort() != null ? overrideProperties.getDiscoveryPort() @@ -1433,6 +1451,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");

14 years, 1 month

1
0
0 / 0

JBoss hornetq SVN: r10195 - branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl.

by do-not-reply＠jboss.org

Author: clebert.suconic(a)jboss.com Date: 2011-02-09 21:23:25 -0500 (Wed, 09 Feb 2011) New Revision: 10195 Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java Log: fixing tests Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2011-02-10 01:02:38 UTC (rev 10194) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2011-02-10 02:23:25 UTC (rev 10195) @@ -547,7 +547,7 @@ if (!configuration.isBackup()) { - if (configuration.isSharedStore()) + if (configuration.isSharedStore() && configuration.isPersistenceEnabled()) { activation = new SharedStoreLiveActivation();

14 years, 1 month

1
0
0 / 0

JBoss hornetq SVN: r10194 - branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl.

by do-not-reply＠jboss.org

Author: clebert.suconic(a)jboss.com Date: 2011-02-09 20:02:38 -0500 (Wed, 09 Feb 2011) New Revision: 10194 Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/FileLockNodeManager.java Log: tweak to fix tests Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/FileLockNodeManager.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/FileLockNodeManager.java 2011-02-09 15:34:49 UTC (rev 10193) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/FileLockNodeManager.java 2011-02-10 01:02:38 UTC (rev 10194) @@ -215,7 +215,10 @@ public void pauseLiveServer() throws Exception { setPaused(); - liveLock.release(); + if (liveLock != null) + { + liveLock.release(); + } } public void crashLiveServer() throws Exception

14 years, 1 month

1
0
0 / 0

JBoss hornetq SVN: r10193 - branches/Branch_2_2_EAP/src/main/org/hornetq/core/config/impl.

by do-not-reply＠jboss.org

Author: clebert.suconic(a)jboss.com Date: 2011-02-09 10:34:49 -0500 (Wed, 09 Feb 2011) New Revision: 10193 Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java Log: Changing default on shared store Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java 2011-02-09 12:51:21 UTC (rev 10192) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java 2011-02-09 15:34:49 UTC (rev 10193) @@ -55,7 +55,7 @@ public static final boolean DEFAULT_ALLOW_AUTO_FAILBACK = true; - public static final boolean DEFAULT_SHARED_STORE = false; + public static final boolean DEFAULT_SHARED_STORE = true; public static final boolean DEFAULT_FILE_DEPLOYMENT_ENABLED = false;

14 years, 1 month

1
0
0 / 0

JBoss hornetq SVN: r10192 - projects/jopr-plugin/tags.

by do-not-reply＠jboss.org

Author: ataylor Date: 2011-02-09 07:51:21 -0500 (Wed, 09 Feb 2011) New Revision: 10192 Added: projects/jopr-plugin/tags/HornetQ-jopr-plugin_2_0_0_CR2/ Log: tag for hornetq jopr plugin 2.0.0.CR2 Copied: projects/jopr-plugin/tags/HornetQ-jopr-plugin_2_0_0_CR2 (from rev 10191, projects/jopr-plugin/trunk)

14 years, 1 month

1
0
0 / 0

JBoss hornetq SVN: r10191 - in projects/jopr-plugin/trunk: src/main/java/org/jboss/as/integration/hornetq/jopr and 1 other directories.

by do-not-reply＠jboss.org

Author: ataylor Date: 2011-02-09 07:46:44 -0500 (Wed, 09 Feb 2011) New Revision: 10191 Modified: projects/jopr-plugin/trunk/pom.xml projects/jopr-plugin/trunk/src/main/java/org/jboss/as/integration/hornetq/jopr/JMSManagerComponent.java projects/jopr-plugin/trunk/src/main/java/org/jboss/as/integration/hornetq/jopr/JMSQueueComponent.java projects/jopr-plugin/trunk/src/main/java/org/jboss/as/integration/hornetq/jopr/JMSTopicComponent.java projects/jopr-plugin/trunk/src/resources/META-INF/rhq-plugin.xml Log: added support for new address settings and added versioning info po manifest Modified: projects/jopr-plugin/trunk/pom.xml =================================================================== --- projects/jopr-plugin/trunk/pom.xml 2011-02-09 02:31:45 UTC (rev 10190) +++ projects/jopr-plugin/trunk/pom.xml 2011-02-09 12:46:44 UTC (rev 10191) @@ -5,7 +5,7 @@ <groupId>org.hornetq</groupId> <artifactId>hornetq-jopr-plugin</artifactId> <packaging>jar</packaging> - <version>2.0.0.CR1</version> + <version>2.0.0.CR2</version> <name>JBoss Application Server JOPR plugin</name> <url>http://hornetq.org</url> <description>HornetQ JOPR plugin</description> @@ -58,11 +58,16 @@ </configuration> </plugin> <plugin> - <groupId>org.apache.maven.plugins</groupId> - <artifactId>maven-jar-plugin</artifactId> - <configuration> - </configuration> - </plugin> + <artifactId>maven-jar-plugin</artifactId> + <configuration> + <archive> + <manifest> + <addDefaultSpecificationEntries>true</addDefaultSpecificationEntries> + <addDefaultImplementationEntries>true</addDefaultImplementationEntries> + </manifest> + </archive> + </configuration> + </plugin> </plugins> </build> <dependencies> Modified: projects/jopr-plugin/trunk/src/main/java/org/jboss/as/integration/hornetq/jopr/JMSManagerComponent.java =================================================================== --- projects/jopr-plugin/trunk/src/main/java/org/jboss/as/integration/hornetq/jopr/JMSManagerComponent.java 2011-02-09 02:31:45 UTC (rev 10190) +++ projects/jopr-plugin/trunk/src/main/java/org/jboss/as/integration/hornetq/jopr/JMSManagerComponent.java 2011-02-09 12:46:44 UTC (rev 10191) @@ -138,6 +138,7 @@ String expiryAddress = simpleProps.get("expiryAddress").getStringValue(); int maxSize = simpleProps.get("maxSize").getIntegerValue(); int pageSize = simpleProps.get("pageSize").getIntegerValue(); + int pageMaxCacheSize = simpleProps.get("pageMaxCacheSize").getIntegerValue(); int maxDeliveryAttempts = simpleProps.get("maxDeliveryAttempts").getIntegerValue(); long redeliveryDelay = simpleProps.get("redeliveryDelay").getLongValue(); boolean lastValueQueue = simpleProps.get("lastValueQueue").getBooleanValue(); @@ -146,11 +147,11 @@ String addressFullMessagePolicy = simpleProps.get("addressFullMessagePolicy").getStringValue(); if ("JMS Queue".equalsIgnoreCase(createResourceReport.getResourceType().getName())) { - createQueue(createResourceReport, managementView, name, jndiName, DLA, expiryAddress, maxSize, pageSize, maxDeliveryAttempts, redeliveryDelay, lastValueQueue, redistributionDelay, sendToDLAOnNoRoute, addressFullMessagePolicy); + createQueue(createResourceReport, managementView, name, jndiName, DLA, expiryAddress, maxSize, pageSize, pageMaxCacheSize, maxDeliveryAttempts, redeliveryDelay, lastValueQueue, redistributionDelay, sendToDLAOnNoRoute, addressFullMessagePolicy); } else if ("JMS Topic".equalsIgnoreCase(createResourceReport.getResourceType().getName())) { - createTopic(createResourceReport, managementView, name, jndiName, DLA, expiryAddress, maxSize, pageSize, maxDeliveryAttempts, redeliveryDelay, lastValueQueue, redistributionDelay, sendToDLAOnNoRoute, addressFullMessagePolicy); + createTopic(createResourceReport, managementView, name, jndiName, DLA, expiryAddress, maxSize, pageSize, pageMaxCacheSize, maxDeliveryAttempts, redeliveryDelay, lastValueQueue, redistributionDelay, sendToDLAOnNoRoute, addressFullMessagePolicy); } } @@ -255,7 +256,7 @@ createResourceReport.setResourceName(name); } - private void createQueue(CreateResourceReport createResourceReport, ManagementView managementView, String name, String jndiName, String DLA, String expiryAddress, int maxSize, int pageSize, int maxDeliveryAttempts, long redeliveryDelay, boolean lastValueQueue, long redistributionDelay, boolean sendToDLAOnNoRoute, String addressFullMessagePolicy) + private void createQueue(CreateResourceReport createResourceReport, ManagementView managementView, String name, String jndiName, String DLA, String expiryAddress, int maxSize, int pageSize, int pageMaxCacheSize, int maxDeliveryAttempts, long redeliveryDelay, boolean lastValueQueue, long redistributionDelay, boolean sendToDLAOnNoRoute, String addressFullMessagePolicy) throws Exception { ManagedOperation operation = ManagementSupport.getOperation(managementView, JMSConstants.Queue.COMPONENT_NAME, "createQueue", JMSConstants.Queue.COMPONENT_TYPE); @@ -270,6 +271,7 @@ new SimpleValueSupport(SimpleMetaType.STRING, expiryAddress), new SimpleValueSupport(SimpleMetaType.INTEGER_PRIMITIVE, maxSize), new SimpleValueSupport(SimpleMetaType.INTEGER_PRIMITIVE, pageSize), + new SimpleValueSupport(SimpleMetaType.INTEGER_PRIMITIVE, pageMaxCacheSize), new SimpleValueSupport(SimpleMetaType.INTEGER_PRIMITIVE, maxDeliveryAttempts), new SimpleValueSupport(SimpleMetaType.LONG_PRIMITIVE, redeliveryDelay), new SimpleValueSupport(SimpleMetaType.BOOLEAN_PRIMITIVE, lastValueQueue), @@ -283,7 +285,7 @@ createResourceReport.setResourceName("jms.queue." + name); } - private void createTopic(CreateResourceReport createResourceReport, ManagementView managementView, String name, String jndiName, String DLA, String expiryAddress, int maxSize, int pageSize, int maxDeliveryAttempts, long redeliveryDelay, boolean lastValueQueue, long redistributionDelay, boolean sendToDLAOnNoRoute, String addressFullMessagePolicy) + private void createTopic(CreateResourceReport createResourceReport, ManagementView managementView, String name, String jndiName, String DLA, String expiryAddress, int maxSize, int pageSize, int pageMaxCacheSize, int maxDeliveryAttempts, long redeliveryDelay, boolean lastValueQueue, long redistributionDelay, boolean sendToDLAOnNoRoute, String addressFullMessagePolicy) throws Exception { ManagedOperation operation = ManagementSupport.getOperation(managementView, JMSConstants.Topic.COMPONENT_NAME, @@ -303,6 +305,7 @@ new SimpleValueSupport(SimpleMetaType.STRING, expiryAddress), new SimpleValueSupport(SimpleMetaType.INTEGER_PRIMITIVE, maxSize), new SimpleValueSupport(SimpleMetaType.INTEGER_PRIMITIVE, pageSize), + new SimpleValueSupport(SimpleMetaType.INTEGER_PRIMITIVE, pageMaxCacheSize), new SimpleValueSupport(SimpleMetaType.INTEGER_PRIMITIVE, maxDeliveryAttempts), new SimpleValueSupport(SimpleMetaType.LONG_PRIMITIVE, redeliveryDelay), new SimpleValueSupport(SimpleMetaType.BOOLEAN_PRIMITIVE, lastValueQueue), Modified: projects/jopr-plugin/trunk/src/main/java/org/jboss/as/integration/hornetq/jopr/JMSQueueComponent.java =================================================================== --- projects/jopr-plugin/trunk/src/main/java/org/jboss/as/integration/hornetq/jopr/JMSQueueComponent.java 2011-02-09 02:31:45 UTC (rev 10190) +++ projects/jopr-plugin/trunk/src/main/java/org/jboss/as/integration/hornetq/jopr/JMSQueueComponent.java 2011-02-09 12:46:44 UTC (rev 10191) @@ -140,6 +140,7 @@ String expiryAddress = simpleProps.get("expiryAddress").getStringValue(); int maxSize = simpleProps.get("maxSize").getIntegerValue(); int pageSize = simpleProps.get("pageSize").getIntegerValue(); + int pageMaxCacheSize = simpleProps.get("pageMaxCacheSize").getIntegerValue(); int maxDeliveryAttempts = simpleProps.get("maxDeliveryAttempts").getIntegerValue(); long redeliveryDelay = simpleProps.get("redeliveryDelay").getLongValue(); boolean lastValueQueue = simpleProps.get("lastValueQueue").getBooleanValue(); @@ -160,6 +161,7 @@ new SimpleValueSupport(SimpleMetaType.STRING, expiryAddress), new SimpleValueSupport(SimpleMetaType.INTEGER_PRIMITIVE, maxSize), new SimpleValueSupport(SimpleMetaType.INTEGER_PRIMITIVE, pageSize), + new SimpleValueSupport(SimpleMetaType.INTEGER_PRIMITIVE, pageMaxCacheSize), new SimpleValueSupport(SimpleMetaType.INTEGER_PRIMITIVE, maxDeliveryAttempts), new SimpleValueSupport(SimpleMetaType.LONG_PRIMITIVE, redeliveryDelay), new SimpleValueSupport(SimpleMetaType.BOOLEAN_PRIMITIVE, lastValueQueue), Modified: projects/jopr-plugin/trunk/src/main/java/org/jboss/as/integration/hornetq/jopr/JMSTopicComponent.java =================================================================== --- projects/jopr-plugin/trunk/src/main/java/org/jboss/as/integration/hornetq/jopr/JMSTopicComponent.java 2011-02-09 02:31:45 UTC (rev 10190) +++ projects/jopr-plugin/trunk/src/main/java/org/jboss/as/integration/hornetq/jopr/JMSTopicComponent.java 2011-02-09 12:46:44 UTC (rev 10191) @@ -117,6 +117,7 @@ String expiryAddress = simpleProps.get("expiryAddress").getStringValue(); int maxSize = simpleProps.get("maxSize").getIntegerValue(); int pageSize = simpleProps.get("pageSize").getIntegerValue(); + int pageMaxCacheSize = simpleProps.get("pageMaxCacheSize").getIntegerValue(); int maxDeliveryAttempts = simpleProps.get("maxDeliveryAttempts").getIntegerValue(); long redeliveryDelay = simpleProps.get("redeliveryDelay").getLongValue(); boolean lastValueQueue = simpleProps.get("lastValueQueue").getBooleanValue(); @@ -142,6 +143,7 @@ new SimpleValueSupport(SimpleMetaType.STRING, expiryAddress), new SimpleValueSupport(SimpleMetaType.INTEGER_PRIMITIVE, maxSize), new SimpleValueSupport(SimpleMetaType.INTEGER_PRIMITIVE, pageSize), + new SimpleValueSupport(SimpleMetaType.INTEGER_PRIMITIVE, pageMaxCacheSize), new SimpleValueSupport(SimpleMetaType.INTEGER_PRIMITIVE, maxDeliveryAttempts), new SimpleValueSupport(SimpleMetaType.LONG_PRIMITIVE, redeliveryDelay), new SimpleValueSupport(SimpleMetaType.BOOLEAN_PRIMITIVE, lastValueQueue), Modified: projects/jopr-plugin/trunk/src/resources/META-INF/rhq-plugin.xml =================================================================== --- projects/jopr-plugin/trunk/src/resources/META-INF/rhq-plugin.xml 2011-02-09 02:31:45 UTC (rev 10190) +++ projects/jopr-plugin/trunk/src/resources/META-INF/rhq-plugin.xml 2011-02-09 12:46:44 UTC (rev 10191) @@ -26,7 +26,6 @@ displayName="HornetQPlugin" description="Supports management and monitoring of HornetQ Servers" package="org.jboss.as.integration.hornetq.jopr" - version="2.1.10" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="urn:xmlns:rhq-plugin" xmlns:c="urn:xmlns:rhq-configuration"> @@ -309,6 +308,10 @@ required="true"> <c:description>The size in bytes that an address can reach before paging starts. -1 means no limit.</c:description> </c:simple-property> + <c:simple-property name="pageMaxCacheSize" default="5" type="integer" displayName="Page Cache Max Size" + required="true"> + <c:description>The max number of pages in the soft memory cache.</c:description> + </c:simple-property> <c:simple-property name="maxDeliveryAttempts" type="integer" default="10" displayName="Max Delivery Attempts" required="true"> <c:description>The maximum time to attempt delivery of a message to a consumer, once hit the message is routed to which ever address is specified by 'Dead Letter Address'</c:description> @@ -506,6 +509,10 @@ required="true"> <c:description>The size in bytes an address can reach before paging occurs</c:description> </c:simple-property> + <c:simple-property name="pageMaxCacheSize" default="5" type="integer" displayName="Page Cache Max Size" + required="true"> + <c:description>The max number of pages in the soft memory cache.</c:description> + </c:simple-property> <c:simple-property name="maxDeliveryAttempts" type="integer" default="10" displayName="Max Delivery Attempts" required="true"> <c:description>The maximum time to attempt delivery of a message to a consumer, once hit the message is routed to which ever address is specified by 'Dead Letter Address'</c:description>

14 years, 1 month

1
0
0 / 0

JBoss hornetq SVN: r10190 - branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/management.

by do-not-reply＠jboss.org

Author: clebert.suconic(a)jboss.com Date: 2011-02-08 21:31:45 -0500 (Tue, 08 Feb 2011) New Revision: 10190 Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/management/HornetQServerControlUsingCoreTest.java Log: fix test Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/management/HornetQServerControlUsingCoreTest.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/management/HornetQServerControlUsingCoreTest.java 2011-02-08 21:20:33 UTC (rev 10189) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/management/HornetQServerControlUsingCoreTest.java 2011-02-09 02:31:45 UTC (rev 10190) @@ -526,6 +526,7 @@ deliveryAttempts, maxSizeBytes, pageSizeBytes, + pageMaxCacheSize, redeliveryDelay, redistributionDelay, sendToDLAOnNoRoute,

14 years, 1 month

1
0
0 / 0

2025

2024

2023

2022

2021

2020

2019

2018

2017

2016

2015

2014

2013

2012

2011

2010

2009

hornetq-commits February 2011