JBoss hornetq SVN: r11673 - in trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration: replication and 1 other directory.
by do-not-reply＠jboss.org
Author: borges
Date: 2011-11-08 11:33:21 -0500 (Tue, 08 Nov 2011)
New Revision: 11673
Modified:
trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncJournalTest.java
trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/FailoverTestBase.java
trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/replication/ReplicationTest.java
Log:
Fix ReplicationTest.testNoActions()
Modified: trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncJournalTest.java
===================================================================
--- trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncJournalTest.java 2011-11-08 16:33:01 UTC (rev 11672)
+++ trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncJournalTest.java 2011-11-08 16:33:21 UTC (rev 11673)
@@ -65,12 +65,12 @@
backupServer.start();
// Deliver messages with Backup in-sync
- waitForBackup(sessionFactory, BACKUP_WAIT_TIME, false);
+ waitForBackup(sessionFactory, BACKUP_WAIT_TIME, false, backupServer.getServer());
sendMessages(session, producer, n_msgs);
// Deliver messages with Backup up-to-date
syncDelay.deliverUpToDateMsg();
- waitForBackup(sessionFactory, BACKUP_WAIT_TIME, true);
+ waitForBackup(sessionFactory, BACKUP_WAIT_TIME, true, backupServer.getServer());
// SEND more messages, now with the backup replicating
sendMessages(session, producer, n_msgs);
@@ -114,14 +114,14 @@
{
syncDelay.deliverUpToDateMsg();
backupServer.start();
- waitForBackup(sessionFactory, BACKUP_WAIT_TIME, true);
+ waitForBackup(sessionFactory, BACKUP_WAIT_TIME, true, backupServer.getServer());
}
public void testReplicationDuringSync() throws Exception
{
createProducerSendSomeMessages();
backupServer.start();
- waitForBackup(sessionFactory, BACKUP_WAIT_TIME, false);
+ waitForBackup(sessionFactory, BACKUP_WAIT_TIME, false, backupServer.getServer());
sendMessages(session, producer, n_msgs);
session.commit();
@@ -136,7 +136,7 @@
private void finishSyncAndFailover() throws Exception
{
syncDelay.deliverUpToDateMsg();
- waitForBackup(sessionFactory, BACKUP_WAIT_TIME, true);
+ waitForBackup(sessionFactory, BACKUP_WAIT_TIME, true, backupServer.getServer());
assertFalse("should not be initialized", backupServer.getServer().isInitialised());
crash(session);
waitForServerInitialization(backupServer, 5);
Modified: trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/FailoverTestBase.java
===================================================================
--- trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/FailoverTestBase.java 2011-11-08 16:33:01 UTC (rev 11672)
+++ trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/FailoverTestBase.java 2011-11-08 16:33:21 UTC (rev 11673)
@@ -38,6 +38,7 @@
import org.hornetq.core.config.Configuration;
import org.hornetq.core.remoting.impl.invm.InVMConnector;
import org.hornetq.core.remoting.impl.invm.InVMRegistry;
+import org.hornetq.core.server.HornetQServer;
import org.hornetq.core.server.NodeManager;
import org.hornetq.core.server.impl.HornetQServerImpl;
import org.hornetq.core.server.impl.InVMNodeManager;
@@ -287,30 +288,36 @@
* @param seconds
* @throws Exception
*/
- protected void waitForBackup(ClientSessionFactoryInternal sessionFactory, long seconds) throws Exception
+ protected void waitForBackup(ClientSessionFactoryInternal sessionFactory, int seconds) throws Exception
{
- waitForBackup(sessionFactory, seconds, true);
- }
+ final HornetQServerImpl actualServer = (HornetQServerImpl)backupServer.getServer();
+ waitForBackup(sessionFactory, seconds, true, actualServer);
+ }
/**
* @param sessionFactory
* @param seconds
- * @param waitForSync whether to wait for sync'ing data with the live to finish or not
+ * @param waitForSync
+ * @param actualServer
*/
- protected void waitForBackup(ClientSessionFactoryInternal sessionFactory, long seconds, boolean waitForSync)
+ public static void waitForBackup(ClientSessionFactoryInternal sessionFactory,
+ int seconds,
+ boolean waitForSync,
+ final HornetQServer backup)
{
+ final HornetQServerImpl actualServer = (HornetQServerImpl)backup;
final long toWait = seconds * 1000;
final long time = System.currentTimeMillis();
- final HornetQServerImpl actualServer = (HornetQServerImpl)backupServer.getServer();
while (true)
{
- if (sessionFactory.getBackupConnector() != null && (actualServer.isRemoteBackupUpToDate() || !waitForSync))
+ if ((sessionFactory == null || sessionFactory.getBackupConnector() != null) &&
+ (actualServer.isRemoteBackupUpToDate() || !waitForSync))
{
break;
}
if (System.currentTimeMillis() > (time + toWait))
{
- fail("backup server never started (" + backupServer.isStarted() + "), or never finished synchronizing (" +
+ fail("backup server never started (" + actualServer.isStarted() + "), or never finished synchronizing (" +
actualServer.isRemoteBackupUpToDate() + ")");
}
try
Modified: trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/replication/ReplicationTest.java
===================================================================
--- trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/replication/ReplicationTest.java 2011-11-08 16:33:01 UTC (rev 11672)
+++ trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/replication/ReplicationTest.java 2011-11-08 16:33:21 UTC (rev 11673)
@@ -75,6 +75,7 @@
import org.hornetq.core.settings.HierarchicalRepository;
import org.hornetq.core.settings.impl.AddressSettings;
import org.hornetq.spi.core.protocol.RemotingConnection;
+import org.hornetq.tests.integration.cluster.failover.FailoverTestBase;
import org.hornetq.tests.util.ReplicatedBackupUtils;
import org.hornetq.tests.util.ServiceTestBase;
import org.hornetq.tests.util.TransportConfigurationUtils;
@@ -114,6 +115,7 @@
Configuration backupConfig = createDefaultConfig(netty);
Configuration liveConfig = createDefaultConfig(netty);
+
backupConfig.setBackup(backup);
if (interceptors.length > 0)
{
@@ -431,6 +433,7 @@
StorageManager storage = getStorage();
manager = liveServer.getReplicationManager();
waitForComponent(manager);
+ FailoverTestBase.waitForBackup(null, 5, true, backupServer);
Journal replicatedJournal = new ReplicatedJournal((byte)1, new FakeJournal(), manager);
@@ -452,7 +455,7 @@
Assert.assertTrue(latch.await(1, TimeUnit.SECONDS));
- Assert.assertEquals(0, manager.getActiveTokens().size());
+ Assert.assertEquals("should be empty " + manager.getActiveTokens(), 0, manager.getActiveTokens().size());
}
public void testOrderOnNonPersistency() throws Exception
12 years, 7 months
- 1
- 0
JBoss hornetq SVN: r11672 - in trunk: hornetq-jms/src/main/java/org/hornetq/jms/client and 1 other directory.
by do-not-reply＠jboss.org
Author: borges
Date: 2011-11-08 11:33:01 -0500 (Tue, 08 Nov 2011)
New Revision: 11672
Modified:
trunk/hornetq-commons/src/main/java/org/hornetq/utils/Base64.java
trunk/hornetq-jms/src/main/java/org/hornetq/jms/client/HornetQConnectionFactory.java
Log:
Fix NPEs
Modified: trunk/hornetq-commons/src/main/java/org/hornetq/utils/Base64.java
===================================================================
--- trunk/hornetq-commons/src/main/java/org/hornetq/utils/Base64.java 2011-11-08 16:00:23 UTC (rev 11671)
+++ trunk/hornetq-commons/src/main/java/org/hornetq/utils/Base64.java 2011-11-08 16:33:01 UTC (rev 11672)
@@ -1692,7 +1692,10 @@
{
try
{
- bis.close();
+ if (bis != null)
+ {
+ bis.close();
+ }
}
catch (Exception e)
{
Modified: trunk/hornetq-jms/src/main/java/org/hornetq/jms/client/HornetQConnectionFactory.java
===================================================================
--- trunk/hornetq-jms/src/main/java/org/hornetq/jms/client/HornetQConnectionFactory.java 2011-11-08 16:00:23 UTC (rev 11671)
+++ trunk/hornetq-jms/src/main/java/org/hornetq/jms/client/HornetQConnectionFactory.java 2011-11-08 16:33:01 UTC (rev 11672)
@@ -38,7 +38,7 @@
/**
* HornetQ implementation of a JMS ConnectionFactory.
- *
+ *
* @author <a href="mailto:ovidiu@feodorov.com">Ovidiu Feodorov</a>
* @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a>
* @version <tt>$Revision$</tt> $Id$
@@ -71,11 +71,11 @@
{
serverLocator = null;
}
-
+
public HornetQConnectionFactory(final ServerLocator serverLocator)
{
this.serverLocator = serverLocator;
-
+
serverLocator.disableFinalizeCheck();
}
@@ -89,7 +89,7 @@
{
serverLocator = HornetQClient.createServerLocatorWithoutHA(groupConfiguration);
}
-
+
serverLocator.disableFinalizeCheck();
}
@@ -103,7 +103,7 @@
{
serverLocator = HornetQClient.createServerLocatorWithoutHA(initialConnectors);
}
-
+
serverLocator.disableFinalizeCheck();
}
@@ -128,7 +128,7 @@
public QueueConnection createQueueConnection(final String username, final String password) throws JMSException
{
- return (QueueConnection)createConnectionInternal(username, password, false, HornetQConnection.TYPE_QUEUE_CONNECTION);
+ return createConnectionInternal(username, password, false, HornetQConnection.TYPE_QUEUE_CONNECTION);
}
// TopicConnectionFactory implementation --------------------------------------------------------
@@ -140,7 +140,7 @@
public TopicConnection createTopicConnection(final String username, final String password) throws JMSException
{
- return (TopicConnection)createConnectionInternal(username, password, false, HornetQConnection.TYPE_TOPIC_CONNECTION);
+ return createConnectionInternal(username, password, false, HornetQConnection.TYPE_TOPIC_CONNECTION);
}
// XAConnectionFactory implementation -----------------------------------------------------------
@@ -473,7 +473,7 @@
checkWrite();
return serverLocator.getInitialConnectAttempts();
}
-
+
public synchronized boolean isFailoverOnInitialConnection()
{
return serverLocator.isFailoverOnInitialConnection();
@@ -538,22 +538,22 @@
{
return serverLocator.getGroupID();
}
-
+
public boolean isCompressLargeMessage()
{
return serverLocator.isCompressLargeMessage();
}
-
+
public void setCompressLargeMessage(boolean compress)
{
serverLocator.setCompressLargeMessage(compress);
}
-
+
public void close()
{
serverLocator.close();
}
-
+
public ServerLocator getServerLocator()
{
return serverLocator;
@@ -564,7 +564,7 @@
return JMSFactoryType.CF.intValue();
}
/**
- *
+ *
* @deprecated use {@link ServerLocator#createSessionFactory()}
* @return
*/
@@ -582,7 +582,7 @@
throw ex;
}
}
-
+
// Package protected ----------------------------------------------------------------------------
// Protected ------------------------------------------------------------------------------------
@@ -595,7 +595,7 @@
readOnly = true;
ClientSessionFactory factory;
-
+
try
{
factory = serverLocator.createSessionFactory();
@@ -603,15 +603,15 @@
catch (Exception e)
{
JMSException jmse = new JMSException("Failed to create session factory");
-
+
jmse.initCause(e);
jmse.setLinkedException(e);
-
+
throw jmse;
}
HornetQConnection connection = null;
-
+
if (isXA)
{
if (type == HornetQConnection.TYPE_GENERIC_CONNECTION)
@@ -676,9 +676,15 @@
dupsOKBatchSize,
transactionBatchSize,
factory);
- }
+ }
}
+
+ if (connection == null)
+ {
+ new JMSException("Failed to create connection: invalid type " + type);
+ }
connection.setReference(this);
+
try
{
connection.authorize();
@@ -708,6 +714,7 @@
}
}
+ @Override
public void finalize() throws Throwable
{
try
12 years, 7 months
- 1
- 0
JBoss hornetq SVN: r11671 - branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/impl.
by do-not-reply＠jboss.org
Author: ataylor
Date: 2011-11-08 11:00:23 -0500 (Tue, 08 Nov 2011)
New Revision: 11671
Modified:
branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java
Log:
added constructor back for bindings manager
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java 2011-11-08 15:18:23 UTC (rev 11670)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java 2011-11-08 16:00:23 UTC (rev 11671)
@@ -149,6 +149,17 @@
configFileName = null;
}
+ public JMSServerManagerImpl(final HornetQServer server, final BindingRegistry registry) throws Exception
+ {
+ this.server = server;
+
+ this.coreConfig = server.getConfiguration();
+
+ configFileName = null;
+
+ this.registry = registry;
+ }
+
public JMSServerManagerImpl(final HornetQServer server, final String configFileName) throws Exception
{
this.server = server;
12 years, 7 months
- 1
- 0
JBoss hornetq SVN: r11670 - trunk/hornetq-core/src/main/java/org/hornetq/core/server/cluster/impl.
by do-not-reply＠jboss.org
Author: borges
Date: 2011-11-08 10:18:23 -0500 (Tue, 08 Nov 2011)
New Revision: 11670
Modified:
trunk/hornetq-core/src/main/java/org/hornetq/core/server/cluster/impl/ClusterManagerImpl.java
Log:
Fix NPE on non-existent bridge
Modified: trunk/hornetq-core/src/main/java/org/hornetq/core/server/cluster/impl/ClusterManagerImpl.java
===================================================================
--- trunk/hornetq-core/src/main/java/org/hornetq/core/server/cluster/impl/ClusterManagerImpl.java 2011-11-08 15:18:07 UTC (rev 11669)
+++ trunk/hornetq-core/src/main/java/org/hornetq/core/server/cluster/impl/ClusterManagerImpl.java 2011-11-08 15:18:23 UTC (rev 11670)
@@ -548,8 +548,8 @@
managementService.unregisterBridge(name);
}
}
-
- bridge.flushExecutor();
+ if (bridge != null)
+ bridge.flushExecutor();
}
// for testing
12 years, 7 months
- 1
- 0
JBoss hornetq SVN: r11669 - trunk.
by do-not-reply＠jboss.org
Author: borges
Date: 2011-11-08 10:18:07 -0500 (Tue, 08 Nov 2011)
New Revision: 11669
Modified:
trunk/pom.xml
Log:
Mvn findbugs plugin to output using Findbugs native XML format.
Modified: trunk/pom.xml
===================================================================
--- trunk/pom.xml 2011-11-08 15:17:56 UTC (rev 11668)
+++ trunk/pom.xml 2011-11-08 15:18:07 UTC (rev 11669)
@@ -560,6 +560,7 @@
<artifactId>findbugs-maven-plugin</artifactId>
<version>2.3.2</version>
<configuration>
+ <findbugsXmlOutput>true</findbugsXmlOutput>
<xmlOutput>true</xmlOutput>
<effort>Max</effort>
<failOnError>false</failOnError>
12 years, 7 months
- 1
- 0
JBoss hornetq SVN: r11668 - trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution.
by do-not-reply＠jboss.org
Author: borges
Date: 2011-11-08 10:17:56 -0500 (Tue, 08 Nov 2011)
New Revision: 11668
Modified:
trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/SymmetricClusterTest.java
trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/SymmetricClusterWithBackupTest.java
Log:
make sure test will clean up properly
Modified: trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/SymmetricClusterTest.java
===================================================================
--- trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/SymmetricClusterTest.java 2011-11-08 15:17:39 UTC (rev 11667)
+++ trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/SymmetricClusterTest.java 2011-11-08 15:17:56 UTC (rev 11668)
@@ -18,11 +18,11 @@
/**
* A SymmetricClusterTest
- *
+ *
* Most of the cases are covered in OneWayTwoNodeClusterTest - we don't duplicate them all here
*
* @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a>
- *
+ *
* Created 3 Feb 2009 09:10:43
*
*
@@ -43,9 +43,14 @@
protected void tearDown() throws Exception
{
log.info("#test tearDown");
- stopServers();
-
- super.tearDown();
+ try
+ {
+ stopServers();
+ }
+ finally
+ {
+ super.tearDown();
+ }
}
protected boolean isNetty()
@@ -1085,7 +1090,7 @@
public void testRouteWhenNoConsumersFalseNoLocalConsumerLoadBalancedQueues() throws Exception
{
setupCluster(false);
-
+
startServers();
for (int i = 0 ; i <= 4; i++)
@@ -1382,7 +1387,7 @@
{
doTestStartStopServers(1, 3000);
}
-
+
public void doTestStartStopServers(long pauseBeforeServerRestarts, long pauseAfterServerRestarts) throws Exception
{
setupCluster();
@@ -1478,7 +1483,7 @@
System.out.println(clusterDescription(servers[3]));
System.out.println(clusterDescription(servers[4]));
System.out.println("#####################################");
-
+
waitForBindings(0, "queues.testaddress", 23, 23, false);
waitForBindings(1, "queues.testaddress", 23, 23, false);
waitForBindings(2, "queues.testaddress", 23, 23, false);
@@ -1521,9 +1526,9 @@
System.out.println(clusterDescription(servers[3]));
System.out.println(clusterDescription(servers[4]));
System.out.println("#####################################");
-
+
Thread.sleep(pauseBeforeServerRestarts);
-
+
startServers(3, 0);
Thread.sleep(pauseAfterServerRestarts);
@@ -1572,7 +1577,7 @@
waitForBindings(2, "queues.testaddress", 5, 5, true);
waitForBindings(3, "queues.testaddress", 6, 6, true);
waitForBindings(4, "queues.testaddress", 7, 7, true);
-
+
System.out.println("### after servers 0 & 3 restarted ###");
System.out.println(clusterDescription(servers[0]));
System.out.println(clusterDescription(servers[1]));
@@ -1716,12 +1721,12 @@
closeSessionFactory(0);
stopServers(0);
-
+
waitForBindings(1, "queues.testaddress", 5, 5, true);
waitForBindings(2, "queues.testaddress", 5, 5, true);
waitForBindings(3, "queues.testaddress", 6, 6, true);
waitForBindings(4, "queues.testaddress", 7, 7, true);
-
+
waitForBindings(1, "queues.testaddress", 18, 18, false);
waitForBindings(2, "queues.testaddress", 18, 18, false);
waitForBindings(3, "queues.testaddress", 17, 17, false);
@@ -1737,11 +1742,11 @@
closeSessionFactory(1);
stopServers(1);
-
+
waitForBindings(2, "queues.testaddress", 5, 5, true);
waitForBindings(3, "queues.testaddress", 6, 6, true);
waitForBindings(4, "queues.testaddress", 7, 7, true);
-
+
waitForBindings(2, "queues.testaddress", 13, 13, false);
waitForBindings(3, "queues.testaddress", 12, 12, false);
waitForBindings(4, "queues.testaddress", 11, 11, false);
@@ -1756,10 +1761,10 @@
closeSessionFactory(2);
stopServers(2);
-
+
waitForBindings(3, "queues.testaddress", 6, 6, true);
waitForBindings(4, "queues.testaddress", 7, 7, true);
-
+
waitForBindings(3, "queues.testaddress", 7, 7, false);
waitForBindings(4, "queues.testaddress", 6, 6, false);
@@ -1774,9 +1779,9 @@
closeSessionFactory(3);
stopServers(3);
-
+
waitForBindings(4, "queues.testaddress", 7, 7, true);
-
+
waitForBindings(4, "queues.testaddress", 0, 0, false);
}
@@ -1822,12 +1827,11 @@
stopServers(0, 1, 2, 3, 4);
}
-
+
+ @Override
protected boolean isFileStorage()
{
return false;
}
-
-
}
Modified: trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/SymmetricClusterWithBackupTest.java
===================================================================
--- trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/SymmetricClusterWithBackupTest.java 2011-11-08 15:17:39 UTC (rev 11667)
+++ trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/distribution/SymmetricClusterWithBackupTest.java 2011-11-08 15:17:56 UTC (rev 11668)
@@ -27,17 +27,13 @@
/**
* A SymmetricClusterWithBackupTest
- *
- * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a>
- *
- * Created 13 Mar 2009 11:00:31
- *
- *
+ * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> Created 13 Mar 2009 11:00:31
*/
public class SymmetricClusterWithBackupTest extends SymmetricClusterTest
{
private static final Logger log = Logger.getLogger(SymmetricClusterWithBackupTest.class);
+ @Override
public void testStopAllStartAll() throws Throwable
{
try
@@ -129,12 +125,6 @@
verifyReceiveRoundRobinInSomeOrder(10, 0, 1, 2, 3, 4);
verifyNotReceive(0, 1, 2, 3, 4);
-
- closeAllConsumers();
-
- closeAllSessionFactories();
-
- closeAllServerLocatorsFactories();
}
catch (Throwable e)
{
@@ -256,12 +246,6 @@
verifyReceiveRoundRobinInSomeOrder(10, 23, 24, 25);
verifyReceiveRoundRobinInSomeOrder(10, 26, 27);
-
- closeAllConsumers();
-
- closeAllSessionFactories();
-
- closeAllServerLocatorsFactories();
}
//@Override
@@ -460,12 +444,6 @@
verifyReceiveRoundRobinInSomeOrder(10, 23, 24, 25);
verifyReceiveRoundRobinInSomeOrder(10, 26, 27);
-
- closeAllConsumers();
-
- closeAllSessionFactories();
-
- closeAllServerLocatorsFactories();
}
@Override
12 years, 7 months
- 1
- 0
JBoss hornetq SVN: r11667 - trunk/hornetq-journal/src/main/java/org/hornetq/core/journal/impl.
by do-not-reply＠jboss.org
Author: borges
Date: 2011-11-08 10:17:39 -0500 (Tue, 08 Nov 2011)
New Revision: 11667
Modified:
trunk/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/JournalImpl.java
Log:
HORNETQ-720 Also during replication, ensure minimum files during loading
Modified: trunk/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/JournalImpl.java
===================================================================
--- trunk/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/JournalImpl.java 2011-11-08 00:45:05 UTC (rev 11666)
+++ trunk/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/JournalImpl.java 2011-11-08 15:17:39 UTC (rev 11667)
@@ -2008,6 +2008,7 @@
if (replicationSync)
{
assert filesRepository.getDataFiles().isEmpty();
+ filesRepository.ensureMinFiles();
setJournalState(JournalState.SYNCING);
return new JournalLoadInformation(0, -1);
}
@@ -3081,13 +3082,11 @@
{
if (!currentFile.getFile().isOpen())
currentFile.getFile().open();
-
currentFile.getFile().position(currentFile.getFile().calculateBlockStart(lastDataPos));
}
else
{
currentFile = filesRepository.getFreeFile();
-
filesRepository.openFile(currentFile, true);
}
12 years, 7 months
- 1
- 0
JBoss hornetq SVN: r11666 - in branches/Branch_2_2_AS7: src/main/org/hornetq/api/core/client and 31 other directories.
by do-not-reply＠jboss.org
Author: clebert.suconic(a)jboss.com
Date: 2011-11-07 19:45:05 -0500 (Mon, 07 Nov 2011)
New Revision: 11666
Modified:
branches/Branch_2_2_AS7/src/config/common/schema/hornetq-configuration.xsd
branches/Branch_2_2_AS7/src/main/org/hornetq/api/core/client/ServerLocator.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientProducerCreditManagerImpl.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientProducerCredits.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientProducerCreditsImpl.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ServerLocatorImpl.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/Topology.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/config/BridgeConfiguration.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/config/ClusterConnectionConfiguration.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/deployers/impl/FileConfigurationParser.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/SequentialFile.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/impl/AIOSequentialFile.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/impl/AbstractSequentialFile.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/impl/NIOSequentialFile.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/management/impl/HornetQServerControlImpl.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/paging/impl/PageImpl.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/persistence/impl/journal/LargeServerMessageImpl.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/persistence/impl/journal/PrintData.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/postoffice/impl/BindingsImpl.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/postoffice/impl/SimpleAddressManager.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/protocol/core/ServerSessionPacketHandler.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/protocol/core/impl/CoreProtocolManager.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/remoting/impl/netty/NettyConnection.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/Bridge.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/ClusterConnection.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionBridge.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/ClusterManagerImpl.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/RemoteQueueBindingImpl.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/DivertImpl.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java
branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/QueueImpl.java
branches/Branch_2_2_AS7/src/main/org/hornetq/jms/client/HornetQConnectionFactory.java
branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/HornetQXAResourceWrapper.java
branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/XARecoveryConfig.java
branches/Branch_2_2_AS7/src/main/org/hornetq/ra/HornetQRAManagedConnectionFactory.java
branches/Branch_2_2_AS7/src/main/org/hornetq/ra/HornetQResourceAdapter.java
branches/Branch_2_2_AS7/src/main/org/hornetq/ra/recovery/RecoveryManager.java
branches/Branch_2_2_AS7/tests/config/ConfigurationTest-full-config.xml
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/IncompatibleVersionTest.java
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/InterruptedLargeMessageTest.java
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/JournalCrashTest.java
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/LargeMessageCompressTest.java
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/PagingOrderTest.java
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/PagingTest.java
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeReconnectTest.java
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeStartTest.java
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeTest.java
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeWithDiscoveryGroupStartTest.java
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeWithPagingTest.java
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusterTestBase.java
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusteredRequestResponseTest.java
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/MessageRedistributionTest.java
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/TwoWayTwoNodeClusterTest.java
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/topology/HAClientTopologyWithDiscoveryTest.java
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/topology/IsolatedTopologyTest.java
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/topology/TopologyClusterTestBase.java
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageTestBase.java
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/management/BridgeControlTest.java
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/management/BridgeControlUsingCoreTest.java
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerSecurityTest.java
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerTest.java
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerXATest.java
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/ResourceAdapterTest.java
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/core/config/impl/FileConfigurationTest.java
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/core/journal/impl/fakes/FakeSequentialFileFactory.java
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/core/persistence/impl/BatchIDGeneratorUnitTest.java
branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/util/UnitTestCase.java
Log:
merging changes from EAP Branch
Modified: branches/Branch_2_2_AS7/src/config/common/schema/hornetq-configuration.xsd
===================================================================
--- branches/Branch_2_2_AS7/src/config/common/schema/hornetq-configuration.xsd 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/config/common/schema/hornetq-configuration.xsd 2011-11-08 00:45:05 UTC (rev 11666)
@@ -313,6 +313,8 @@
</xsd:element>
<xsd:element maxOccurs="1" minOccurs="0" name="transformer-class-name" type="xsd:string">
</xsd:element>
+ <xsd:element maxOccurs="1" minOccurs="0" name="min-large-message-size" type="xsd:int">
+ </xsd:element>
<xsd:element maxOccurs="1" minOccurs="0" name="check-period" type="xsd:long">
</xsd:element>
<xsd:element maxOccurs="1" minOccurs="0" name="connection-ttl" type="xsd:long">
@@ -364,6 +366,8 @@
</xsd:element>
<xsd:element maxOccurs="1" minOccurs="0" name="connection-ttl" type="xsd:long">
</xsd:element>
+ <xsd:element maxOccurs="1" minOccurs="0" name="min-large-message-size" type="xsd:int">
+ </xsd:element>
<xsd:element maxOccurs="1" minOccurs="0" name="call-timeout" type="xsd:long">
</xsd:element>
<xsd:element maxOccurs="1" minOccurs="0" name="retry-interval" type="xsd:long">
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/api/core/client/ServerLocator.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/api/core/client/ServerLocator.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/api/core/client/ServerLocator.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -54,6 +54,15 @@
* Create a ClientSessionFactory to a specific server. The server must already be known about by this ServerLocator.
* This method allows the user to make a connection to a specific server bypassing any load balancing policy in force
* @param transportConfiguration
+ * @return The ClientSesionFactory or null if the node is not present on the topology
+ * @throws Exception if a failure happened in creating the ClientSessionFactory or the ServerLocator does not know about the passed in transportConfiguration
+ */
+ ClientSessionFactory createSessionFactory(final String nodeID) throws Exception;
+
+ /**
+ * Create a ClientSessionFactory to a specific server. The server must already be known about by this ServerLocator.
+ * This method allows the user to make a connection to a specific server bypassing any load balancing policy in force
+ * @param transportConfiguration
* @return The ClientSesionFactory
* @throws Exception if a failure happened in creating the ClientSessionFactory or the ServerLocator does not know about the passed in transportConfiguration
*/
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -14,6 +14,8 @@
package org.hornetq.core.client.impl;
import java.io.File;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
import java.util.Iterator;
import java.util.concurrent.Executor;
@@ -116,6 +118,8 @@
private final SessionQueueQueryResponseMessage queueInfo;
private volatile boolean ackIndividually;
+
+ private final ClassLoader contextClassLoader;
// Constructors
// ---------------------------------------------------------------------------------
@@ -130,7 +134,8 @@
final TokenBucketLimiter rateLimiter,
final Executor executor,
final Channel channel,
- final SessionQueueQueryResponseMessage queueInfo)
+ final SessionQueueQueryResponseMessage queueInfo,
+ final ClassLoader contextClassLoader)
{
this.id = id;
@@ -153,6 +158,8 @@
this.ackBatchSize = ackBatchSize;
this.queueInfo = queueInfo;
+
+ this.contextClassLoader = contextClassLoader;
}
// ClientConsumer implementation
@@ -861,7 +868,7 @@
{
return;
}
-
+
session.workDone();
// We pull the message from the buffer from inside the Runnable so we can ensure priority
@@ -894,6 +901,8 @@
return;
}
+
+
boolean expired = message.isExpired();
flowControlBeforeConsumption(message);
@@ -906,8 +915,33 @@
{
ClientConsumerImpl.log.trace("Calling handler.onMessage");
}
- theHandler.onMessage(message);
+ final ClassLoader originalLoader = AccessController.doPrivileged(new PrivilegedAction<ClassLoader>()
+ {
+ public ClassLoader run()
+ {
+ ClassLoader originalLoader = Thread.currentThread().getContextClassLoader();
+ Thread.currentThread().setContextClassLoader(contextClassLoader);
+
+ return originalLoader;
+ }
+ });
+ try
+ {
+ theHandler.onMessage(message);
+ }
+ finally
+ {
+ AccessController.doPrivileged(new PrivilegedAction<Object>()
+ {
+ public Object run()
+ {
+ Thread.currentThread().setContextClassLoader(originalLoader);
+ return null;
+ }
+ });
+ }
+
if (ClientConsumerImpl.trace)
{
ClientConsumerImpl.log.trace("Handler.onMessage done");
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientProducerCreditManagerImpl.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientProducerCreditManagerImpl.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientProducerCreditManagerImpl.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -48,29 +48,44 @@
this.windowSize = windowSize;
}
- public synchronized ClientProducerCredits getCredits(final SimpleString address, final boolean anon)
+ public ClientProducerCredits getCredits(final SimpleString address, final boolean anon)
{
- ClientProducerCredits credits = producerCredits.get(address);
-
- if (credits == null)
+ boolean needInit = false;
+ ClientProducerCredits credits;
+
+ synchronized(this)
{
- // Doesn't need to be fair since session is single threaded
- credits = new ClientProducerCreditsImpl(session, address, windowSize);
-
- producerCredits.put(address, credits);
+ credits = producerCredits.get(address);
+
+ if (credits == null)
+ {
+ // Doesn't need to be fair since session is single threaded
+ credits = new ClientProducerCreditsImpl(session, address, windowSize);
+ needInit = true;
+
+ producerCredits.put(address, credits);
+ }
+
+ if (!anon)
+ {
+ credits.incrementRefCount();
+
+ // Remove from anon credits (if there)
+ unReferencedCredits.remove(address);
+ }
+ else
+ {
+ addToUnReferencedCache(address, credits);
+ }
}
-
- if (!anon)
+
+ // The init is done outside of the lock
+ // otherwise packages may arrive with flow control
+ // while this is still sending requests causing a dead lock
+ if (needInit)
{
- credits.incrementRefCount();
-
- // Remove from anon credits (if there)
- unReferencedCredits.remove(address);
+ credits.init();
}
- else
- {
- addToUnReferencedCache(address, credits);
- }
return credits;
}
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientProducerCredits.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientProducerCredits.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientProducerCredits.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -27,6 +27,8 @@
void receiveCredits(int credits);
boolean isBlocked();
+
+ void init();
void reset();
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientProducerCreditsImpl.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientProducerCreditsImpl.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientProducerCreditsImpl.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -58,7 +58,10 @@
// Doesn't need to be fair since session is single threaded
semaphore = new Semaphore(0, false);
-
+ }
+
+ public void init()
+ {
// We initial request twice as many credits as we request in subsequent requests
// This allows the producer to keep sending as more arrive, minimising pauses
checkCredits(windowSize);
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -885,9 +885,8 @@
}
// Should never get here
- throw new IllegalStateException("Internal Error! ClientSessionFactoryImpl::createSessionInternal " +
- "just reached a condition that was not supposed to happen. " +
- "Please inform this condition to the HornetQ team");
+ throw new IllegalStateException("Internal Error! ClientSessionFactoryImpl::createSessionInternal " + "just reached a condition that was not supposed to happen. "
+ + "Please inform this condition to the HornetQ team");
}
private void callFailureListeners(final HornetQException me, final boolean afterReconnect, final boolean failedOver)
@@ -1293,7 +1292,9 @@
ClientSessionFactoryImpl.log.trace(this + "::Subscribing Topology");
}
- channel0.send(new SubscribeClusterTopologyUpdatesMessageV2(serverLocator.isClusterConnection(), VersionLoader.getVersion().getIncrementingVersion()));
+ channel0.send(new SubscribeClusterTopologyUpdatesMessageV2(serverLocator.isClusterConnection(),
+ VersionLoader.getVersion()
+ .getIncrementingVersion()));
}
}
@@ -1314,13 +1315,16 @@
/**
* @param channel0
*/
- public void sendNodeAnnounce(final long currentEventID, String nodeID, boolean isBackup, TransportConfiguration config, TransportConfiguration backupConfig)
+ public void sendNodeAnnounce(final long currentEventID,
+ String nodeID,
+ boolean isBackup,
+ TransportConfiguration config,
+ TransportConfiguration backupConfig)
{
Channel channel0 = connection.getChannel(0, -1);
if (ClientSessionFactoryImpl.isDebug)
{
- ClientSessionFactoryImpl.log.debug("Announcing node " + serverLocator.getNodeID() +
- ", isBackup=" + isBackup);
+ ClientSessionFactoryImpl.log.debug("Announcing node " + serverLocator.getNodeID() + ", isBackup=" + isBackup);
}
channel0.send(new NodeAnnounceMessage(currentEventID, nodeID, isBackup, config, backupConfig));
}
@@ -1489,7 +1493,10 @@
" csf created at\nserverLocator=" +
serverLocator, e);
}
- serverLocator.notifyNodeUp(System.currentTimeMillis(), topMessage.getNodeID(), topMessage.getPair(), topMessage.isLast());
+ serverLocator.notifyNodeUp(System.currentTimeMillis(),
+ topMessage.getNodeID(),
+ topMessage.getPair(),
+ topMessage.isLast());
}
}
else if (type == PacketImpl.CLUSTER_TOPOLOGY_V2)
@@ -1516,7 +1523,10 @@
" csf created at\nserverLocator=" +
serverLocator, e);
}
- serverLocator.notifyNodeUp(topMessage.getUniqueEventID(), topMessage.getNodeID(), topMessage.getPair(), topMessage.isLast());
+ serverLocator.notifyNodeUp(topMessage.getUniqueEventID(),
+ topMessage.getNodeID(),
+ topMessage.getPair(),
+ topMessage.isLast());
}
}
}
@@ -1642,6 +1652,20 @@
}
/* (non-Javadoc)
+ * @see java.lang.Object#toString()
+ */
+ @Override
+ public String toString()
+ {
+ return "ClientSessionFactoryImpl [serverLocator=" + serverLocator +
+ ", connectorConfig=" +
+ connectorConfig +
+ ", backupConfig=" +
+ backupConfig +
+ "]";
+ }
+
+ /* (non-Javadoc)
* @see org.hornetq.core.client.impl.ClientSessionFactoryInternal#setReconnectAttempts(int)
*/
public void setReconnectAttempts(final int attempts)
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -12,6 +12,8 @@
*/
package org.hornetq.core.client.impl;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
@@ -82,6 +84,7 @@
import org.hornetq.core.remoting.FailureListener;
import org.hornetq.spi.core.protocol.RemotingConnection;
import org.hornetq.spi.core.remoting.Connection;
+import org.hornetq.spi.core.remoting.ConnectorFactory;
import org.hornetq.utils.ConcurrentHashSet;
import org.hornetq.utils.IDGenerator;
import org.hornetq.utils.SimpleIDGenerator;
@@ -925,7 +928,7 @@
{
synchronized (this)
{
- if (closed)
+ if (closed/* || closing*/)
{
return;
}
@@ -1687,7 +1690,8 @@
username +
", closed=" +
closed +
- " metaData=(" +
+ ", factory = " + this.sessionFactory +
+ ", metaData=(" +
buffer +
")]@" +
Integer.toHexString(hashCode());
@@ -1778,7 +1782,8 @@
: null,
executor,
channel,
- queueInfo);
+ queueInfo,
+ lookupTCCL());
addConsumer(consumer);
@@ -1848,7 +1853,19 @@
throw new HornetQException(HornetQException.OBJECT_CLOSED, "Session is closed");
}
}
+
+ private ClassLoader lookupTCCL()
+ {
+ return AccessController.doPrivileged(new PrivilegedAction<ClassLoader>()
+ {
+ public ClassLoader run()
+ {
+ return Thread.currentThread().getContextClassLoader();
+ }
+ });
+ }
+
private void doCleanup(boolean failingOver)
{
if (remotingConnection == null)
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ServerLocatorImpl.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ServerLocatorImpl.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ServerLocatorImpl.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -57,6 +57,15 @@
*/
public class ServerLocatorImpl implements ServerLocatorInternal, DiscoveryListener, Serializable
{
+ /*needed for backward compatibility*/
+ private final Set<ClusterTopologyListener> topologyListeners = new HashSet<ClusterTopologyListener>();
+
+ /*end of compatibility fixes*/
+ private enum STATE
+ {
+ INITIALIZED, CLOSED, CLOSING
+ };
+
private static final long serialVersionUID = -1615857864410205260L;
private static final Logger log = Logger.getLogger(ServerLocatorImpl.class);
@@ -69,8 +78,10 @@
private transient String identity;
- private Set<ClientSessionFactoryInternal> factories = new HashSet<ClientSessionFactoryInternal>();
+ private final Set<ClientSessionFactoryInternal> factories = new HashSet<ClientSessionFactoryInternal>();
+ private final Set<ClientSessionFactoryInternal> connectingFactories = new HashSet<ClientSessionFactoryInternal>();
+
private TransportConfiguration[] initialConnectors;
private DiscoveryGroupConfiguration discoveryGroupConfiguration;
@@ -79,9 +90,9 @@
private final Topology topology;
- private Pair<TransportConfiguration, TransportConfiguration>[] topologyArray;
+ private volatile Pair<TransportConfiguration, TransportConfiguration>[] topologyArray;
- private boolean receivedTopology;
+ private volatile boolean receivedTopology;
private boolean compressLargeMessage;
@@ -154,10 +165,8 @@
private int initialMessagePacketSize;
- private volatile boolean closed;
+ private volatile STATE state;
- private volatile boolean closing;
-
private final List<Interceptor> interceptors = new CopyOnWriteArrayList<Interceptor>();
private static ExecutorService globalThreadPool;
@@ -252,7 +261,7 @@
return globalScheduledThreadPool;
}
- private void setThreadPools()
+ private synchronized void setThreadPools()
{
if (threadPool != null)
{
@@ -329,10 +338,16 @@
});
}
- private synchronized void initialise() throws Exception
+ private synchronized void initialise() throws HornetQException
{
- if (!readOnly)
+ if (readOnly)
{
+ return;
+ }
+
+ try
+ {
+ state = STATE.INITIALIZED;
setThreadPools();
instantiateLoadBalancingPolicy();
@@ -366,6 +381,11 @@
readOnly = true;
}
+ catch (Exception e)
+ {
+ state = null;
+ throw new HornetQException(HornetQException.INTERNAL_ERROR, "Failed to initialise session factory", e);
+ }
}
private ServerLocatorImpl(final Topology topology,
@@ -375,7 +395,7 @@
{
e.fillInStackTrace();
- this.topology = topology;
+ this.topology = topology == null ? new Topology(this) : topology;
this.ha = useHA;
@@ -454,7 +474,7 @@
*/
public ServerLocatorImpl(final boolean useHA, final DiscoveryGroupConfiguration groupConfiguration)
{
- this(useHA ? new Topology(null) : null, useHA, groupConfiguration, null);
+ this(new Topology(null), useHA, groupConfiguration, null);
if (useHA)
{
// We only set the owner at where the Topology was created.
@@ -470,7 +490,7 @@
*/
public ServerLocatorImpl(final boolean useHA, final TransportConfiguration... transportConfigs)
{
- this(useHA ? new Topology(null) : null, useHA, null, transportConfigs);
+ this(new Topology(null), useHA, null, transportConfigs);
if (useHA)
{
// We only set the owner at where the Topology was created.
@@ -505,7 +525,7 @@
this(topology, useHA, null, transportConfigs);
}
- private TransportConfiguration selectConnector()
+ private synchronized TransportConfiguration selectConnector()
{
if (receivedTopology)
{
@@ -515,14 +535,12 @@
return pair.getA();
}
- else
- {
- // Get from initialconnectors
- int pos = loadBalancingPolicy.select(initialConnectors.length);
+ // Get from initialconnectors
- return initialConnectors[pos];
- }
+ int pos = loadBalancingPolicy.select(initialConnectors.length);
+
+ return initialConnectors[pos];
}
public void start(Executor executor) throws Exception
@@ -541,7 +559,7 @@
}
catch (Exception e)
{
- if (!closing)
+ if (!isClosed())
{
log.warn("did not connect the cluster connection to other nodes", e);
}
@@ -565,19 +583,15 @@
public ClientSessionFactoryInternal connect() throws Exception
{
- ClientSessionFactoryInternal sf;
// static list of initial connectors
if (initialConnectors != null && discoveryGroup == null)
{
- sf = (ClientSessionFactoryInternal)staticConnector.connect();
+ ClientSessionFactoryInternal sf = (ClientSessionFactoryInternal)staticConnector.connect();
+ addFactory(sf);
+ return sf;
}
// wait for discovery group to get the list of initial connectors
- else
- {
- sf = (ClientSessionFactoryInternal)createSessionFactory();
- }
- addFactory(sf);
- return sf;
+ return (ClientSessionFactoryInternal)createSessionFactory();
}
/* (non-Javadoc)
@@ -593,27 +607,45 @@
return afterConnectListener;
}
- public boolean isClosed()
+ public ClientSessionFactory createSessionFactory(String nodeID) throws Exception
{
- return closed || closing;
- }
+ log.info(topology.describe("full topology"));
+ TopologyMember topologyMember = topology.getMember(nodeID);
- public ClientSessionFactory createSessionFactory(final TransportConfiguration transportConfiguration) throws Exception
- {
- if (closed)
+ log.info("Creating connection factory towards " + nodeID + " = " + topologyMember);
+
+ if (topologyMember == null)
{
- throw new IllegalStateException("Cannot create session factory, server locator is closed (maybe it has been garbage collected)");
+ return null;
}
-
- try
+ else if (topologyMember.getA() != null)
{
- initialise();
+ ClientSessionFactoryInternal factory = (ClientSessionFactoryInternal)createSessionFactory(topologyMember.getA());
+ if (topologyMember.getB() != null)
+ {
+ factory.setBackupConnector(topologyMember.getA(), topologyMember.getB());
+ }
+ return factory;
}
- catch (Exception e)
+ else if (topologyMember.getA() == null && topologyMember.getB() != null)
{
- throw new HornetQException(HornetQException.INTERNAL_ERROR, "Failed to initialise session factory", e);
+ // This shouldn't happen, however I wanted this to consider all possible cases
+ ClientSessionFactoryInternal factory = (ClientSessionFactoryInternal)createSessionFactory(topologyMember.getB());
+ return factory;
}
+ else
+ {
+ // it shouldn't happen
+ return null;
+ }
+ }
+ public ClientSessionFactory createSessionFactory(final TransportConfiguration transportConfiguration) throws Exception
+ {
+ assertOpen();
+
+ initialise();
+
ClientSessionFactoryInternal factory = new ClientSessionFactoryImpl(this,
transportConfiguration,
callTimeout,
@@ -627,29 +659,39 @@
scheduledThreadPool,
interceptors);
- factory.connect(reconnectAttempts, failoverOnInitialConnection);
+ addToConnecting(factory);
+ try
+ {
+ factory.connect(reconnectAttempts, failoverOnInitialConnection);
+ addFactory(factory);
+ return factory;
+ }
+ finally
+ {
+ removeFromConnecting(factory);
+ }
+ }
- addFactory(factory);
-
- return factory;
+ private void removeFromConnecting(ClientSessionFactoryInternal factory)
+ {
+ connectingFactories.remove(factory);
}
- public ClientSessionFactory createSessionFactory() throws Exception
+ private void addToConnecting(ClientSessionFactoryInternal factory)
{
- if (closed || closing)
+ synchronized (connectingFactories)
{
- throw new IllegalStateException("Cannot create session factory, server locator is closed (maybe it has been garbage collected)");
+ assertOpen();
+ connectingFactories.add(factory);
}
+ }
- try
- {
- initialise();
- }
- catch (Exception e)
- {
- throw new HornetQException(HornetQException.INTERNAL_ERROR, "Failed to initialise session factory", e);
- }
+ public ClientSessionFactory createSessionFactory() throws Exception
+ {
+ assertOpen();
+ initialise();
+
if (initialConnectors == null && discoveryGroup != null)
{
// Wait for an initial broadcast to give us at least one node in the cluster
@@ -691,7 +733,15 @@
threadPool,
scheduledThreadPool,
interceptors);
- factory.connect(initialConnectAttempts, failoverOnInitialConnection);
+ try
+ {
+ addToConnecting(factory);
+ factory.connect(initialConnectAttempts, failoverOnInitialConnection);
+ }
+ finally
+ {
+ removeFromConnecting(factory);
+ }
}
catch (HornetQException e)
{
@@ -723,10 +773,8 @@
if (ha || clusterConnection)
{
- long timeout = System.currentTimeMillis() + 30000;
- while (!ServerLocatorImpl.this.closed && !ServerLocatorImpl.this.closing &&
- !receivedTopology &&
- timeout > System.currentTimeMillis())
+ final long timeout = System.currentTimeMillis() + 30000;
+ while (!isClosed() && !receivedTopology && timeout > System.currentTimeMillis())
{
// Now wait for the topology
@@ -740,7 +788,7 @@
}
- if (System.currentTimeMillis() > timeout && !receivedTopology && !closed && !closing)
+ if (System.currentTimeMillis() > timeout && !receivedTopology)
{
throw new HornetQException(HornetQException.CONNECTION_TIMEDOUT,
"Timed out waiting to receive cluster topology. Group:" + discoveryGroup);
@@ -1184,7 +1232,7 @@
protected void doClose(final boolean sendClose)
{
- if (closed)
+ if (state == STATE.CLOSED)
{
if (log.isDebugEnabled())
{
@@ -1193,45 +1241,55 @@
return;
}
- if (log.isDebugEnabled())
- {
- log.debug(this + " is calling close", new Exception("trace"));
- }
+ state = STATE.CLOSING;
- closing = true;
-
if (discoveryGroup != null)
{
- try
+ synchronized (this)
{
- discoveryGroup.stop();
+ try
+ {
+ discoveryGroup.stop();
+ }
+ catch (Exception e)
+ {
+ log.error("Failed to stop discovery group", e);
+ }
}
- catch (Exception e)
- {
- log.error("Failed to stop discovery group", e);
- }
}
else
{
staticConnector.disconnect();
}
- Set<ClientSessionFactoryInternal> clonedFactory = new HashSet<ClientSessionFactoryInternal>(factories);
-
- for (ClientSessionFactory factory : clonedFactory)
+ synchronized (connectingFactories)
{
- if (sendClose)
+ for (ClientSessionFactoryInternal csf : connectingFactories)
{
- factory.close();
+ csf.close();
}
- else
+ connectingFactories.clear();
+ }
+
+ synchronized (factories)
+ {
+ Set<ClientSessionFactoryInternal> clonedFactory = new HashSet<ClientSessionFactoryInternal>(factories);
+
+ for (ClientSessionFactory factory : clonedFactory)
{
- factory.cleanup();
+ if (sendClose)
+ {
+ factory.close();
+ }
+ else
+ {
+ factory.cleanup();
+ }
}
+
+ factories.clear();
}
- factories.clear();
-
if (shutdownPool)
{
if (threadPool != null)
@@ -1268,7 +1326,8 @@
}
readOnly = false;
- closed = true;
+ state = STATE.CLOSED;
+
}
/** This is directly called when the connection to the node is gone,
@@ -1277,7 +1336,7 @@
public void notifyNodeDown(final long eventTime, final String nodeID)
{
- if (topology == null)
+ if (!ha)
{
// there's no topology here
return;
@@ -1288,25 +1347,31 @@
log.debug("nodeDown " + this + " nodeID=" + nodeID + " as being down", new Exception("trace"));
}
- if (topology.removeMember(eventTime, nodeID))
+ topology.removeMember(eventTime, nodeID);
+
+ if (clusterConnection)
{
- if (topology.isEmpty())
+ updateArraysAndPairs();
+ }
+ else
+ {
+ synchronized (this)
{
- // Resetting the topology to its original condition as it was brand new
- synchronized (this)
+ if (topology.isEmpty())
{
+ // Resetting the topology to its original condition as it was brand new
+ receivedTopology = false;
topologyArray = null;
- receivedTopology = false;
}
- }
- else
- {
- updateArraysAndPairs();
+ else
+ {
+ updateArraysAndPairs();
- if (topology.nodes() == 1 && topology.getMember(this.nodeID) != null)
- {
- // Resetting the topology to its original condition as it was brand new
- receivedTopology = false;
+ if (topology.nodes() == 1 && topology.getMember(this.nodeID) != null)
+ {
+ // Resetting the topology to its original condition as it was brand new
+ receivedTopology = false;
+ }
}
}
}
@@ -1318,7 +1383,7 @@
final Pair<TransportConfiguration, TransportConfiguration> connectorPair,
final boolean last)
{
- if (topology == null)
+ if (!ha)
{
// there's no topology
return;
@@ -1331,23 +1396,21 @@
TopologyMember member = new TopologyMember(connectorPair.getA(), connectorPair.getB());
- if (topology.updateMember(uniqueEventID, nodeID, member))
- {
+ topology.updateMember(uniqueEventID, nodeID, member);
- TopologyMember actMember = topology.getMember(nodeID);
+ TopologyMember actMember = topology.getMember(nodeID);
- if (actMember != null && actMember.getConnector().getA() != null && actMember.getConnector().getB() != null)
+ if (actMember != null && actMember.getConnector().getA() != null && actMember.getConnector().getB() != null)
+ {
+ for (ClientSessionFactory factory : factories)
{
- for (ClientSessionFactory factory : factories)
- {
- ((ClientSessionFactoryInternal)factory).setBackupConnector(actMember.getConnector().getA(),
- actMember.getConnector().getB());
- }
+ ((ClientSessionFactoryInternal)factory).setBackupConnector(actMember.getConnector().getA(),
+ actMember.getConnector().getB());
}
-
- updateArraysAndPairs();
}
+ updateArraysAndPairs();
+
if (last)
{
synchronized (this)
@@ -1374,13 +1437,10 @@
discoveryGroupConfiguration +
"]";
}
- else
- {
- return "ServerLocatorImpl [initialConnectors=" + Arrays.toString(initialConnectors) +
- ", discoveryGroupConfiguration=" +
- discoveryGroupConfiguration +
- "]";
- }
+ return "ServerLocatorImpl [initialConnectors=" + Arrays.toString(initialConnectors) +
+ ", discoveryGroupConfiguration=" +
+ discoveryGroupConfiguration +
+ "]";
}
private synchronized void updateArraysAndPairs()
@@ -1409,7 +1469,7 @@
{
this.initialConnectors[count++] = entry.getConnector();
- if (topology != null && topology.getMember(entry.getNodeID()) == null)
+ if (ha && topology.getMember(entry.getNodeID()) == null)
{
TopologyMember member = new TopologyMember(entry.getConnector(), null);
// on this case we set it as zero as any update coming from server should be accepted
@@ -1461,13 +1521,24 @@
topology.removeClusterTopologyListener(listener);
}
- public synchronized void addFactory(ClientSessionFactoryInternal factory)
+ private synchronized void addFactory(ClientSessionFactoryInternal factory)
{
- if (factory != null)
+ if (factory == null)
{
+ return;
+ }
+
+ synchronized (factories)
+ {
+ if (isClosed())
+ {
+ factory.close();
+ return;
+ }
+
TransportConfiguration backup = null;
- if (topology != null)
+ if (ha)
{
backup = topology.getBackupForConnector(factory.getConnectorConfiguration());
}
@@ -1485,19 +1556,9 @@
public ClientSessionFactory connect() throws HornetQException
{
- if (closed)
- {
- throw new IllegalStateException("Cannot create session factory, server locator is closed (maybe it has been garbage collected)");
- }
+ assertOpen();
- try
- {
- initialise();
- }
- catch (Exception e)
- {
- throw new HornetQException(HornetQException.INTERNAL_ERROR, "Failed to initialise session factory", e);
- }
+ initialise();
ClientSessionFactory csf = null;
@@ -1507,7 +1568,7 @@
{
int retryNumber = 0;
- while (csf == null && !ServerLocatorImpl.this.closed && !ServerLocatorImpl.this.closing)
+ while (csf == null && !isClosed())
{
retryNumber++;
for (Connector conn : connectors)
@@ -1560,7 +1621,7 @@
break;
}
- if (!closed && !closing)
+ if (!isClosed())
{
Thread.sleep(retryInterval);
}
@@ -1573,7 +1634,7 @@
throw new HornetQException(HornetQException.NOT_CONNECTED, "Failed to connect to any static connectors", e);
}
- if (csf == null && !closed)
+ if (csf == null && !isClosed())
{
log.warn("Failed to connecto to any static connector, throwing exception now");
throw new HornetQException(HornetQException.NOT_CONNECTED, "Failed to connect to any static connectors");
@@ -1632,7 +1693,7 @@
public void finalize() throws Throwable
{
- if (!closed && finalizeCheck)
+ if (!isClosed() && finalizeCheck)
{
log.warn("I'm closing a core ServerLocator you left open. Please make sure you close all ServerLocators explicitly " + "before letting them go out of scope! " +
System.identityHashCode(this));
@@ -1677,7 +1738,15 @@
ClientSessionFactoryInternal factoryToUse = factory;
if (factoryToUse != null)
{
- factory.connect(1, false);
+ try
+ {
+ addToConnecting(factoryToUse);
+ factoryToUse.connect(1, false);
+ }
+ finally
+ {
+ removeFromConnecting(factoryToUse);
+ }
}
return factoryToUse;
}
@@ -1700,9 +1769,6 @@
}
}
- /* (non-Javadoc)
- * @see java.lang.Object#toString()
- */
@Override
public String toString()
{
@@ -1711,4 +1777,17 @@
}
}
+
+ private void assertOpen()
+ {
+ if (state != null && state != STATE.INITIALIZED)
+ {
+ throw new IllegalStateException("Cannot create session factory, server locator is closed (maybe it has been garbage collected)");
+ }
+ }
+
+ public boolean isClosed()
+ {
+ return state != STATE.INITIALIZED;
+ }
}
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/Topology.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/Topology.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/Topology.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -58,9 +58,9 @@
* keys are node IDs
* values are a pair of live/backup transport configurations
*/
- private final Map<String, TopologyMember> mapTopology = new ConcurrentHashMap<String, TopologyMember>();
+ private final Map<String, TopologyMember> topology = new ConcurrentHashMap<String, TopologyMember>();
- private final Map<String, Long> mapDelete = new ConcurrentHashMap<String, Long>();
+ private transient final Map<String, Long> mapDelete = new ConcurrentHashMap<String, Long>();
public Topology(final Object owner)
{
@@ -111,8 +111,8 @@
log.debug(this + "::node " + nodeId + "=" + memberInput);
}
memberInput.setUniqueEventID(System.currentTimeMillis());
- mapTopology.remove(nodeId);
- mapTopology.put(nodeId, memberInput);
+ topology.remove(nodeId);
+ topology.put(nodeId, memberInput);
sendMemberUp(memberInput.getUniqueEventID(), nodeId, memberInput);
}
}
@@ -130,17 +130,17 @@
TopologyMember currentMember = getMember(nodeId);
if (currentMember == null)
{
- log.warn("There's no live to be updated on backup update, node=" + nodeId + " memberInput=" + memberInput,
+ log.debug("There's no live to be updated on backup update, node=" + nodeId + " memberInput=" + memberInput,
new Exception("trace"));
currentMember = memberInput;
- mapTopology.put(nodeId, currentMember);
+ topology.put(nodeId, currentMember);
}
TopologyMember newMember = new TopologyMember(currentMember.getA(), memberInput.getB());
newMember.setUniqueEventID(System.currentTimeMillis());
- mapTopology.remove(nodeId);
- mapTopology.put(nodeId, newMember);
+ topology.remove(nodeId);
+ topology.put(nodeId, newMember);
sendMemberUp(newMember.getUniqueEventID(), nodeId, newMember);
return newMember;
@@ -172,7 +172,7 @@
synchronized (this)
{
- TopologyMember currentMember = mapTopology.get(nodeId);
+ TopologyMember currentMember = topology.get(nodeId);
if (currentMember == null)
{
@@ -184,7 +184,7 @@
memberInput, new Exception("trace"));
}
memberInput.setUniqueEventID(uniqueEventID);
- mapTopology.put(nodeId, memberInput);
+ topology.put(nodeId, memberInput);
sendMemberUp(uniqueEventID, nodeId, memberInput);
return true;
}
@@ -217,8 +217,8 @@
newMember.setUniqueEventID(uniqueEventID);
- mapTopology.remove(nodeId);
- mapTopology.put(nodeId, newMember);
+ topology.remove(nodeId);
+ topology.put(nodeId, newMember);
sendMemberUp(uniqueEventID, nodeId, newMember);
return true;
@@ -296,7 +296,7 @@
synchronized (this)
{
- member = mapTopology.get(nodeId);
+ member = topology.get(nodeId);
if (member != null)
{
if (member.getUniqueEventID() > uniqueEventID)
@@ -307,7 +307,7 @@
else
{
mapDelete.put(nodeId, uniqueEventID);
- member = mapTopology.remove(nodeId);
+ member = topology.remove(nodeId);
}
}
}
@@ -320,7 +320,7 @@
", result=" +
member +
", size = " +
- mapTopology.size(), new Exception("trace"));
+ topology.size(), new Exception("trace"));
}
if (member != null)
@@ -414,7 +414,7 @@
synchronized (Topology.this)
{
- copy = new HashMap<String, TopologyMember>(mapTopology);
+ copy = new HashMap<String, TopologyMember>(topology);
}
for (Map.Entry<String, TopologyMember> entry : copy.entrySet())
@@ -439,12 +439,12 @@
public synchronized TopologyMember getMember(final String nodeID)
{
- return mapTopology.get(nodeID);
+ return topology.get(nodeID);
}
public synchronized boolean isEmpty()
{
- return mapTopology.isEmpty();
+ return topology.isEmpty();
}
public Collection<TopologyMember> getMembers()
@@ -452,7 +452,7 @@
ArrayList<TopologyMember> members;
synchronized (this)
{
- members = new ArrayList<TopologyMember>(mapTopology.values());
+ members = new ArrayList<TopologyMember>(topology.values());
}
return members;
}
@@ -460,7 +460,7 @@
public synchronized int nodes()
{
int count = 0;
- for (TopologyMember member : mapTopology.values())
+ for (TopologyMember member : topology.values())
{
if (member.getA() != null)
{
@@ -483,12 +483,12 @@
{
String desc = text + "topology on " + this + ":\n";
- for (Entry<String, TopologyMember> entry : new HashMap<String, TopologyMember>(mapTopology).entrySet())
+ for (Entry<String, TopologyMember> entry : new HashMap<String, TopologyMember>(topology).entrySet())
{
desc += "\t" + entry.getKey() + " => " + entry.getValue() + "\n";
}
desc += "\t" + "nodes=" + nodes() + "\t" + "members=" + members();
- if (mapTopology.isEmpty())
+ if (topology.isEmpty())
{
desc += "\tEmpty";
}
@@ -497,7 +497,7 @@
public int members()
{
- return mapTopology.size();
+ return topology.size();
}
/** The owner exists mainly for debug purposes.
@@ -510,7 +510,7 @@
public TransportConfiguration getBackupForConnector(final TransportConfiguration connectorConfiguration)
{
- for (TopologyMember member : mapTopology.values())
+ for (TopologyMember member : topology.values())
{
if (member.getA() != null && member.getA().equals(connectorConfiguration))
{
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/config/BridgeConfiguration.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/config/BridgeConfiguration.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/config/BridgeConfiguration.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -64,6 +64,8 @@
private final long connectionTTL;
private final long maxRetryInterval;
+
+ private final int minLargeMessageSize;
public BridgeConfiguration(final String name,
@@ -71,6 +73,7 @@
final String forwardingAddress,
final String filterString,
final String transformerClassName,
+ final int minLargeMessageSize,
final long clientFailureCheckPeriod,
final long connectionTTL,
final long retryInterval,
@@ -87,6 +90,7 @@
this.name = name;
this.queueName = queueName;
this.forwardingAddress = forwardingAddress;
+ this.minLargeMessageSize = minLargeMessageSize;
this.filterString = filterString;
this.transformerClassName = transformerClassName;
this.retryInterval = retryInterval;
@@ -108,6 +112,7 @@
final String forwardingAddress,
final String filterString,
final String transformerClassName,
+ final int minLargeMessageSize,
final long clientFailureCheckPeriod,
final long connectionTTL,
final long retryInterval,
@@ -126,6 +131,7 @@
this.forwardingAddress = forwardingAddress;
this.filterString = filterString;
this.transformerClassName = transformerClassName;
+ this.minLargeMessageSize = minLargeMessageSize;
this.retryInterval = retryInterval;
this.retryIntervalMultiplier = retryIntervalMultiplier;
this.reconnectAttempts = reconnectAttempts;
@@ -244,6 +250,14 @@
}
/**
+ * @return the minLargeMessageSize
+ */
+ public int getMinLargeMessageSize()
+ {
+ return minLargeMessageSize;
+ }
+
+ /**
* @param forwardingAddress the forwardingAddress to set
*/
public void setForwardingAddress(final String forwardingAddress)
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/config/ClusterConnectionConfiguration.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/config/ClusterConnectionConfiguration.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/config/ClusterConnectionConfiguration.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -66,6 +66,8 @@
private final boolean allowDirectConnectionsOnly;
+ private final int minLargeMessageSize;
+
public ClusterConnectionConfiguration(final String name,
final String address,
final String connectorName,
@@ -80,6 +82,7 @@
this(name,
address,
connectorName,
+ HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE,
ConfigurationImpl.DEFAULT_CLUSTER_FAILURE_CHECK_PERIOD,
ConfigurationImpl.DEFAULT_CLUSTER_CONNECTION_TTL,
retryInterval,
@@ -99,6 +102,7 @@
public ClusterConnectionConfiguration(final String name,
final String address,
final String connectorName,
+ final int minLargeMessageSize,
final long clientFailureCheckPeriod,
final long connectionTTL,
final long retryInterval,
@@ -130,6 +134,7 @@
this.maxHops = maxHops;
this.confirmationWindowSize = confirmationWindowSize;
this.allowDirectConnectionsOnly = allowDirectConnectionsOnly;
+ this.minLargeMessageSize = minLargeMessageSize;
}
@@ -146,6 +151,7 @@
this(name,
address,
connectorName,
+ HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE,
ConfigurationImpl.DEFAULT_CLUSTER_FAILURE_CHECK_PERIOD,
ConfigurationImpl.DEFAULT_CLUSTER_CONNECTION_TTL,
retryInterval,
@@ -164,6 +170,7 @@
public ClusterConnectionConfiguration(final String name,
final String address,
final String connectorName,
+ final int minLargeMessageSize,
final long clientFailureCheckPeriod,
final long connectionTTL,
final long retryInterval,
@@ -193,6 +200,7 @@
this.staticConnectors = null;
this.maxHops = maxHops;
this.confirmationWindowSize = confirmationWindowSize;
+ this.minLargeMessageSize = minLargeMessageSize;
allowDirectConnectionsOnly = false;
}
@@ -295,4 +303,15 @@
{
return allowDirectConnectionsOnly;
}
+
+
+ /**
+ * @return the minLargeMessageSize
+ */
+ public int getMinLargeMessageSize()
+ {
+ return minLargeMessageSize;
+ }
+
+
}
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/deployers/impl/FileConfigurationParser.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/deployers/impl/FileConfigurationParser.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/deployers/impl/FileConfigurationParser.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -1021,6 +1021,8 @@
double retryIntervalMultiplier = XMLConfigurationUtil.getDouble(e, "retry-interval-multiplier",
ConfigurationImpl.DEFAULT_CLUSTER_RETRY_INTERVAL_MULTIPLIER, Validators.GT_ZERO);
+ int minLargeMessageSize = XMLConfigurationUtil.getInteger(e, "min-large-message-size", HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, Validators.GT_ZERO);
+
long maxRetryInterval = XMLConfigurationUtil.getLong(e, "max-retry-interval", ConfigurationImpl.DEFAULT_CLUSTER_MAX_RETRY_INTERVAL, Validators.GT_ZERO);
int reconnectAttempts = XMLConfigurationUtil.getInteger(e, "reconnect-attempts", ConfigurationImpl.DEFAULT_CLUSTER_RECONNECT_ATTEMPTS, Validators.MINUS_ONE_OR_GE_ZERO);
@@ -1065,6 +1067,7 @@
config = new ClusterConnectionConfiguration(name,
address,
connectorName,
+ minLargeMessageSize,
clientFailureCheckPeriod,
connectionTTL,
retryInterval,
@@ -1084,6 +1087,7 @@
config = new ClusterConnectionConfiguration(name,
address,
connectorName,
+ minLargeMessageSize,
clientFailureCheckPeriod,
connectionTTL,
retryInterval,
@@ -1147,6 +1151,11 @@
long connectionTTL = XMLConfigurationUtil.getLong(brNode, "connection-ttl",
HornetQClient.DEFAULT_CONNECTION_TTL, Validators.GT_ZERO) ;
+ int minLargeMessageSize = XMLConfigurationUtil.getInteger(brNode,
+ "min-large-message-size",
+ HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE,
+ Validators.GT_ZERO);
+
long maxRetryInterval = XMLConfigurationUtil.getLong(brNode, "max-retry-interval", HornetQClient.DEFAULT_MAX_RETRY_INTERVAL, Validators.GT_ZERO);
@@ -1211,6 +1220,7 @@
forwardingAddress,
filterString,
transformerClassName,
+ minLargeMessageSize,
clientFailureCheckPeriod,
connectionTTL,
retryInterval,
@@ -1231,6 +1241,7 @@
forwardingAddress,
filterString,
transformerClassName,
+ minLargeMessageSize,
clientFailureCheckPeriod,
connectionTTL,
retryInterval,
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/SequentialFile.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/SequentialFile.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/SequentialFile.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -93,7 +93,7 @@
void renameTo(String newFileName) throws Exception;
- SequentialFile copy();
+ SequentialFile cloneFile();
void copyTo(SequentialFile newFileName) throws Exception;
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/impl/AIOSequentialFile.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/impl/AIOSequentialFile.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/impl/AIOSequentialFile.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -86,7 +86,7 @@
return pos;
}
- public SequentialFile copy()
+ public SequentialFile cloneFile()
{
return new AIOSequentialFile(factory,
-1,
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/impl/AbstractSequentialFile.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/impl/AbstractSequentialFile.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/impl/AbstractSequentialFile.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -109,7 +109,10 @@
{
log.debug("Copying " + this + " as " + newFileName);
newFileName.open();
- this.open();
+ if (!isOpen())
+ {
+ this.open();
+ }
ByteBuffer buffer = ByteBuffer.allocate(10 * 1024);
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/impl/NIOSequentialFile.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/impl/NIOSequentialFile.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/journal/impl/NIOSequentialFile.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -252,7 +252,7 @@
return "NIOSequentialFile " + getFile();
}
- public SequentialFile copy()
+ public SequentialFile cloneFile()
{
return new NIOSequentialFile(factory, getFile(), maxIO, writerExecutor);
}
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/management/impl/HornetQServerControlImpl.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/management/impl/HornetQServerControlImpl.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/management/impl/HornetQServerControlImpl.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -1709,6 +1709,7 @@
forwardingAddress,
filterString,
transformerClassName,
+ HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE,
clientFailureCheckPeriod,
HornetQClient.DEFAULT_CONNECTION_TTL,
retryInterval,
@@ -1730,6 +1731,7 @@
forwardingAddress,
filterString,
transformerClassName,
+ HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE,
clientFailureCheckPeriod,
HornetQClient.DEFAULT_CONNECTION_TTL,
retryInterval,
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/paging/impl/PageImpl.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/paging/impl/PageImpl.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/paging/impl/PageImpl.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -212,7 +212,10 @@
public void open() throws Exception
{
- file.open();
+ if (!file.isOpen())
+ {
+ file.open();
+ }
size.set((int)file.size());
file.position(0);
}
@@ -307,6 +310,21 @@
{
return otherPage.getPageId() - this.pageId;
}
+
+ public void finalize()
+ {
+ try
+ {
+ if (file != null && file.isOpen())
+ {
+ file.close();
+ }
+ }
+ catch (Exception e)
+ {
+ log.warn(e.getMessage(), e);
+ }
+ }
/* (non-Javadoc)
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -114,8 +114,17 @@
for (PagingStore store : reloadedStores)
{
- store.start();
- stores.put(store.getStoreName(), store);
+ PagingStore oldStore = stores.get(store.getStoreName());
+ if (oldStore != null)
+ {
+ oldStore.stop();
+ oldStore.start();
+ }
+ else
+ {
+ store.start();
+ stores.put(store.getStoreName(), store);
+ }
}
}
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -426,6 +426,10 @@
{
currentPageId = 0;
+ if (currentPage != null)
+ {
+ currentPage.close();
+ }
currentPage = null;
List<String> files = fileFactory.listFiles("page");
@@ -815,6 +819,11 @@
PagingStoreImpl.log.warn("Messages are being dropped on address " + getStoreName());
}
+
+ if (log.isDebugEnabled())
+ {
+ log.debug("Message " + message + " beig dropped for fullAddressPolicy==DROP");
+ }
// Address is full, we just pretend we are paging, and drop the data
return true;
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -1275,11 +1275,11 @@
log.info("Deleting unreferenced message id=" + msg.getMessageID() + " from the journal");
try
{
- deleteMessage(msg.getMessageID());
+ deleteMessage(msg.getMessageID());
}
catch (Exception ignored)
{
- log.warn("It wasn't possible to delete message " + msg.getMessageID());
+ log.warn("It wasn't possible to delete message " + msg.getMessageID(), ignored);
}
}
}
@@ -1985,7 +1985,7 @@
}
}
- private OperationContext getContext(final boolean sync)
+ protected OperationContext getContext(final boolean sync)
{
if (sync)
{
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/persistence/impl/journal/LargeServerMessageImpl.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/persistence/impl/journal/LargeServerMessageImpl.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/persistence/impl/journal/LargeServerMessageImpl.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -170,11 +170,21 @@
public synchronized void incrementDelayDeletionCount()
{
delayDeletionCount.incrementAndGet();
+ try
+ {
+ incrementRefCount();
+ }
+ catch (Exception e)
+ {
+ log.warn(e.getMessage(), e);
+ }
}
public synchronized void decrementDelayDeletionCount() throws Exception
{
int count = delayDeletionCount.decrementAndGet();
+
+ decrementRefCount();
if (count == 0)
{
@@ -373,6 +383,10 @@
log.warn("Error on copying large message this for DLA or Expiry", e);
return null;
}
+ finally
+ {
+ releaseResources();
+ }
}
}
@@ -416,7 +430,7 @@
file = storageManager.createFileForLargeMessage(getMessageID(), durable);
- file.open();
+ openFile();
bodySize = file.size();
}
@@ -427,6 +441,27 @@
throw new HornetQException(HornetQException.INTERNAL_ERROR, e.getMessage(), e);
}
}
+
+ protected void openFile() throws Exception
+ {
+ if (file == null)
+ {
+ validateFile();
+ }
+ else
+ if (!file.isOpen())
+ {
+ file.open();
+ }
+ }
+
+ protected void closeFile() throws Exception
+ {
+ if (file != null && file.isOpen())
+ {
+ file.close();
+ }
+ }
/* (non-Javadoc)
* @see org.hornetq.core.server.LargeServerMessage#setLinkedMessage(org.hornetq.core.server.LargeServerMessage)
@@ -444,9 +479,9 @@
file = storageManager.createFileForLargeMessage(message.getMessageID(), durable);
try
{
- file.open();
+ openFile();
bodySize = file.size();
- file.close();
+ closeFile();
}
catch (Exception e)
{
@@ -464,7 +499,11 @@
{
try
{
- cFile = file.copy();
+ if (cFile != null && cFile.isOpen())
+ {
+ cFile.close();
+ }
+ cFile = file.cloneFile();
cFile.open();
}
catch (Exception e)
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/persistence/impl/journal/PrintData.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/persistence/impl/journal/PrintData.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/persistence/impl/journal/PrintData.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -13,6 +13,9 @@
package org.hornetq.core.persistence.impl.journal;
+import java.io.File;
+
+import org.hornetq.core.server.impl.FileLockNodeManager;
/**
* A PrintData
*
@@ -42,6 +45,25 @@
System.exit(-1);
}
+ File serverLockFile = new File(arg[1], "server.lock");
+
+ if (serverLockFile.isFile())
+ {
+ try
+ {
+ FileLockNodeManager fileLock = new FileLockNodeManager(arg[1]);
+ fileLock.start();
+ System.out.println("********************************************");
+ System.out.println("Server's ID=" + fileLock.getNodeId().toString());
+ System.out.println("********************************************");
+ fileLock.stop();
+ }
+ catch (Exception e)
+ {
+ e.printStackTrace();
+ }
+ }
+
System.out.println("********************************************");
System.out.println("B I N D I N G S J O U R N A L");
System.out.println("********************************************");
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/postoffice/impl/BindingsImpl.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/postoffice/impl/BindingsImpl.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/postoffice/impl/BindingsImpl.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -595,6 +595,10 @@
{
binding.route(message, context);
}
+ else
+ {
+ log.warn("Couldn't find binding with id=" + bindingID + " on routeFromCluster for message=" + message + " binding = " + this);
+ }
}
}
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -473,9 +473,9 @@
String uid = UUIDGenerator.getInstance().generateStringUUID();
- if (isTrace)
+ if (log.isDebugEnabled())
{
- log.trace("Sending notification for addBinding " + binding + " from server " + server);
+ log.debug("ClusterCommunication::Sending notification for addBinding " + binding + " from server " + server);
}
managementService.sendNotification(new Notification(uid, NotificationType.BINDING_ADDED, props));
@@ -611,6 +611,19 @@
{
bindings.route(message, context);
}
+ else
+ {
+ // this is a debug and not warn because this could be a regular scenario on publish-subscribe queues (or topic subscriptions on JMS)
+ if (log.isDebugEnabled())
+ {
+ log.debug("Couldn't find any bindings for address=" + address + " on message=" + message);
+ }
+ }
+
+ if (log.isTraceEnabled())
+ {
+ log.trace("Message after routed=" + message);
+ }
if (context.getQueueCount() == 0)
{
@@ -625,6 +638,11 @@
// Send to the DLA for the address
SimpleString dlaAddress = addressSettings.getDeadLetterAddress();
+
+ if (log.isDebugEnabled())
+ {
+ log.debug("sending message to dla address = " + dlaAddress + ", message=" + message);
+ }
if (dlaAddress == null)
{
@@ -641,6 +659,13 @@
route(message, context.getTransaction(), false);
}
}
+ else
+ {
+ if (log.isDebugEnabled())
+ {
+ log.debug("Message " + message + " is not going anywhere as it didn't have a binding on address:" + address);
+ }
+ }
}
else
{
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/postoffice/impl/SimpleAddressManager.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/postoffice/impl/SimpleAddressManager.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/postoffice/impl/SimpleAddressManager.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -52,6 +52,11 @@
{
throw new IllegalStateException("Binding already exists " + binding);
}
+
+ if (log.isDebugEnabled())
+ {
+ log.debug("Adding binding " + binding + " with address = " + binding.getUniqueName(), new Exception ("trace"));
+ }
return addMappingInternal(binding.getAddress(), binding);
}
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/protocol/core/ServerSessionPacketHandler.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/protocol/core/ServerSessionPacketHandler.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/protocol/core/ServerSessionPacketHandler.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -514,6 +514,7 @@
{
if (requiresResponse)
{
+ log.debug("Sending exception to client", e);
response = new SessionXAResponseMessage(true, e.errorCode, e.getMessage());
}
else
@@ -525,6 +526,7 @@
{
if (requiresResponse)
{
+ log.debug("Sending exception to client", e);
response = new HornetQExceptionMessage((HornetQException)e);
}
else
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/protocol/core/impl/CoreProtocolManager.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/protocol/core/impl/CoreProtocolManager.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/protocol/core/impl/CoreProtocolManager.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -39,6 +39,7 @@
import org.hornetq.core.protocol.core.impl.wireformat.SubscribeClusterTopologyUpdatesMessageV2;
import org.hornetq.core.remoting.CloseListener;
import org.hornetq.core.server.HornetQServer;
+import org.hornetq.core.server.cluster.ClusterConnection;
import org.hornetq.spi.core.protocol.ConnectionEntry;
import org.hornetq.spi.core.protocol.ProtocolManager;
import org.hornetq.spi.core.protocol.RemotingConnection;
@@ -208,8 +209,23 @@
{
log.trace("Server " + server + " receiving nodeUp from NodeID=" + msg.getNodeID() + ", pair=" + pair);
}
-
- acceptorUsed.getClusterConnection().nodeAnnounced(msg.getCurrentEventID(), msg.getNodeID(), pair, msg.isBackup());
+
+ if (acceptorUsed != null)
+ {
+ ClusterConnection clusterConn = acceptorUsed.getClusterConnection();
+ if (clusterConn != null)
+ {
+ clusterConn.nodeAnnounced(msg.getCurrentEventID(), msg.getNodeID(), pair, msg.isBackup());
+ }
+ else
+ {
+ log.debug("Cluster connection is null on acceptor = " + acceptorUsed);
+ }
+ }
+ else
+ {
+ log.debug("there is no acceptor used configured at the CoreProtocolManager " + this);
+ }
}
}
});
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/remoting/impl/netty/NettyConnection.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/remoting/impl/netty/NettyConnection.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/remoting/impl/netty/NettyConnection.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -14,7 +14,7 @@
package org.hornetq.core.remoting.impl.netty;
import java.util.Set;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.Semaphore;
import org.hornetq.api.core.HornetQBuffer;
import org.hornetq.api.core.HornetQBuffers;
@@ -60,27 +60,27 @@
private volatile HornetQBuffer batchBuffer;
- private final AtomicBoolean writeLock = new AtomicBoolean(false);
-
- private Set<ReadyListener> readyListeners = new ConcurrentHashSet<ReadyListener>();
+ private final Semaphore writeLock = new Semaphore(1);
+ private final Set<ReadyListener> readyListeners = new ConcurrentHashSet<ReadyListener>();
+
// Static --------------------------------------------------------
// Constructors --------------------------------------------------
public NettyConnection(final Channel channel,
final ConnectionLifeCycleListener listener,
- boolean batchingEnabled,
- boolean directDeliver)
+ final boolean batchingEnabled,
+ final boolean directDeliver)
{
this(null, channel, listener, batchingEnabled, directDeliver);
}
-
+
public NettyConnection(final Acceptor acceptor,
final Channel channel,
final ConnectionLifeCycleListener listener,
- boolean batchingEnabled,
- boolean directDeliver)
+ final boolean batchingEnabled,
+ final boolean directDeliver)
{
this.channel = channel;
@@ -152,7 +152,7 @@
return;
}
- if (writeLock.compareAndSet(false, true))
+ if (writeLock.tryAcquire())
{
try
{
@@ -160,12 +160,12 @@
{
channel.write(batchBuffer.channelBuffer());
- batchBuffer = HornetQBuffers.dynamicBuffer(BATCHING_BUFFER_SIZE);
+ batchBuffer = HornetQBuffers.dynamicBuffer(NettyConnection.BATCHING_BUFFER_SIZE);
}
}
finally
{
- writeLock.set(false);
+ writeLock.release();
}
}
}
@@ -177,73 +177,78 @@
public void write(HornetQBuffer buffer, final boolean flush, final boolean batched)
{
- while (!writeLock.compareAndSet(false, true))
- {
- Thread.yield();
- }
try
{
- if (batchBuffer == null && batchingEnabled && batched && !flush)
- {
- // Lazily create batch buffer
+ writeLock.acquire();
- batchBuffer = HornetQBuffers.dynamicBuffer(BATCHING_BUFFER_SIZE);
- }
-
- if (batchBuffer != null)
+ try
{
- batchBuffer.writeBytes(buffer, 0, buffer.writerIndex());
-
- if (batchBuffer.writerIndex() >= BATCHING_BUFFER_SIZE || !batched || flush)
+ if (batchBuffer == null && batchingEnabled && batched && !flush)
{
- // If the batch buffer is full or it's flush param or not batched then flush the buffer
+ // Lazily create batch buffer
- buffer = batchBuffer;
+ batchBuffer = HornetQBuffers.dynamicBuffer(NettyConnection.BATCHING_BUFFER_SIZE);
}
- else
- {
- return;
- }
- if (!batched || flush)
+ if (batchBuffer != null)
{
- batchBuffer = null;
- }
- else
- {
- // Create a new buffer
+ batchBuffer.writeBytes(buffer, 0, buffer.writerIndex());
- batchBuffer = HornetQBuffers.dynamicBuffer(BATCHING_BUFFER_SIZE);
+ if (batchBuffer.writerIndex() >= NettyConnection.BATCHING_BUFFER_SIZE || !batched || flush)
+ {
+ // If the batch buffer is full or it's flush param or not batched then flush the buffer
+
+ buffer = batchBuffer;
+ }
+ else
+ {
+ return;
+ }
+
+ if (!batched || flush)
+ {
+ batchBuffer = null;
+ }
+ else
+ {
+ // Create a new buffer
+
+ batchBuffer = HornetQBuffers.dynamicBuffer(NettyConnection.BATCHING_BUFFER_SIZE);
+ }
}
- }
- ChannelFuture future = channel.write(buffer.channelBuffer());
+ ChannelFuture future = channel.write(buffer.channelBuffer());
- if (flush)
- {
- while (true)
+ if (flush)
{
- try
+ while (true)
{
- boolean ok = future.await(10000);
+ try
+ {
+ boolean ok = future.await(10000);
- if (!ok)
+ if (!ok)
+ {
+ NettyConnection.log.warn("Timed out waiting for packet to be flushed");
+ }
+
+ break;
+ }
+ catch (InterruptedException ignore)
{
- NettyConnection.log.warn("Timed out waiting for packet to be flushed");
}
-
- break;
}
- catch (InterruptedException ignore)
- {
- }
}
}
+ finally
+ {
+ writeLock.release();
+ }
}
- finally
+ catch (InterruptedException e)
{
- writeLock.set(false);
+ Thread.currentThread().interrupt();
}
}
@@ -256,20 +261,20 @@
{
return directDeliver;
}
-
+
public void addReadyListener(final ReadyListener listener)
{
readyListeners.add(listener);
}
-
+
public void removeReadyListener(final ReadyListener listener)
{
readyListeners.remove(listener);
}
-
+
public void fireReady(final boolean ready)
{
- for (ReadyListener listener: readyListeners)
+ for (ReadyListener listener : readyListeners)
{
listener.readyForWriting(ready);
}
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/Bridge.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/Bridge.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/Bridge.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -59,4 +59,6 @@
* Basically this is for cluster bridges being disconnected
*/
void disconnect();
+
+ boolean isConnected();
}
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/ClusterConnection.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/ClusterConnection.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/ClusterConnection.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -65,4 +65,6 @@
void informTopology();
void announceBackup();
+
+ boolean isNodeActive(String id);
}
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -25,9 +25,9 @@
import org.hornetq.api.core.Message;
import org.hornetq.api.core.SimpleString;
import org.hornetq.api.core.client.ClientProducer;
+import org.hornetq.api.core.client.ClientSession.BindingQuery;
import org.hornetq.api.core.client.SendAcknowledgementHandler;
import org.hornetq.api.core.client.SessionFailureListener;
-import org.hornetq.api.core.client.ClientSession.BindingQuery;
import org.hornetq.api.core.management.NotificationType;
import org.hornetq.core.client.impl.ClientSessionFactoryInternal;
import org.hornetq.core.client.impl.ClientSessionInternal;
@@ -46,7 +46,6 @@
import org.hornetq.core.server.management.Notification;
import org.hornetq.core.server.management.NotificationService;
import org.hornetq.spi.core.protocol.RemotingConnection;
-import org.hornetq.utils.ConcurrentHashSet;
import org.hornetq.utils.Future;
import org.hornetq.utils.TypedProperties;
import org.hornetq.utils.UUID;
@@ -132,6 +131,8 @@
private NotificationService notificationService;
+ private boolean stopping = false;
+
// Static --------------------------------------------------------
// Constructors --------------------------------------------------
@@ -199,7 +200,7 @@
{
this.notificationService = notificationService;
}
-
+
public synchronized void start() throws Exception
{
if (started)
@@ -209,6 +210,8 @@
started = true;
+ stopping = false;
+
if (activated)
{
activate();
@@ -222,7 +225,7 @@
notificationService.sendNotification(notification);
}
}
-
+
public String debug()
{
return toString();
@@ -305,20 +308,32 @@
});
}
- /** The cluster manager needs to use the same executor to close the serverLocator, otherwise the stop will break.
- * This method is intended to expose this executor to the ClusterManager */
+ public boolean isConnected()
+ {
+ return session != null;
+ }
+
+ /** The cluster manager needs to use the same executor to close the serverLocator, otherwise the stop will break.
+ * This method is intended to expose this executor to the ClusterManager */
public Executor getExecutor()
{
return executor;
}
-
+
public void stop() throws Exception
{
+ if (stopping)
+ {
+ return;
+ }
+
+ stopping = true;
+
if (log.isDebugEnabled())
{
log.debug("Bridge " + this.name + " being stopped");
}
-
+
if (futureScheduledReconnection != null)
{
futureScheduledReconnection.cancel(true);
@@ -471,7 +486,10 @@
{
if (log.isDebugEnabled())
{
- log.debug("The transformer " + transformer + " made a copy of the message " + message + " as transformedMessage");
+ log.debug("The transformer " + transformer +
+ " made a copy of the message " +
+ message +
+ " as transformedMessage");
}
}
return transformedMessage;
@@ -544,12 +562,12 @@
// that this will throw a disconnect, we need to remove the message
// from the acks so it will get resent, duplicate detection will cope
// with any messages resent
-
+
if (log.isTraceEnabled())
{
log.trace("going to send message " + message);
}
-
+
try
{
producer.send(dest, message);
@@ -580,7 +598,7 @@
{
producer.close();
}
-
+
csf.cleanup();
}
catch (Throwable dontCare)
@@ -681,7 +699,6 @@
return csf;
}
-
/* Hook for creating session factory */
protected ClientSessionFactoryInternal createSessionFactory() throws Exception
{
@@ -702,6 +719,12 @@
if (csf == null || csf.isClosed())
{
csf = createSessionFactory();
+ if (csf == null)
+ {
+ // Retrying. This probably means the node is not available (for the cluster connection case)
+ scheduleRetryConnect();
+ return;
+ }
// Session is pre-acknowledge
session = (ClientSessionInternal)csf.createSession(user, password, false, true, true, true, 1);
}
@@ -801,6 +824,12 @@
return;
}
+ if (stopping)
+ {
+ log.info("Bridge is stopping, will not retry");
+ return;
+ }
+
if (reconnectAttemptsInUse >= 0 && retryCount > reconnectAttempts)
{
log.warn("Bridge " + this.name +
@@ -963,7 +992,10 @@
{
public synchronized void run()
{
- connect();
+ if (!stopping)
+ {
+ connect();
+ }
}
}
}
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionBridge.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionBridge.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionBridge.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -71,11 +71,9 @@
private final SimpleString idsHeaderName;
private final String targetNodeID;
-
+
private final long targetNodeEventUID;
- private final TransportConfiguration connector;
-
private final ServerLocatorInternal discoveryLocator;
public ClusterConnectionBridge(final ClusterConnection clusterConnection,
@@ -138,7 +136,6 @@
this.managementAddress = managementAddress;
this.managementNotificationAddress = managementNotificationAddress;
this.flowRecord = flowRecord;
- this.connector = connector;
// we need to disable DLQ check on the clustered bridges
queue.setInternalQueue(true);
@@ -152,7 +149,15 @@
protected ClientSessionFactoryInternal createSessionFactory() throws Exception
{
- ClientSessionFactoryInternal factory = super.createSessionFactory();
+ ClientSessionFactoryInternal factory = (ClientSessionFactoryInternal)serverLocator.createSessionFactory(targetNodeID);
+
+ if (factory == null)
+ {
+ log.warn("NodeID=" + targetNodeID +
+ " is not available on the topology. Retrying the connection to that node now");
+ return null;
+ }
+ factory.setReconnectAttempts(0);
factory.getConnection().addFailureListener(new FailureListener()
{
@@ -177,7 +182,7 @@
// nodes could have same queue ids
// Note we must copy since same message may get routed to other nodes which require different headers
ServerMessage messageCopy = message.copy();
-
+
if (log.isTraceEnabled())
{
log.trace("Clustered bridge copied message " + message + " as " + messageCopy + " before delivery");
@@ -188,7 +193,7 @@
Set<SimpleString> propNames = new HashSet<SimpleString>(messageCopy.getPropertyNames());
byte[] queueIds = message.getBytesProperty(idsHeaderName);
-
+
if (queueIds == null)
{
// Sanity check only
@@ -211,7 +216,7 @@
messageCopy.putBytesProperty(MessageImpl.HDR_ROUTE_TO_IDS, queueIds);
messageCopy = super.beforeForward(messageCopy);
-
+
return messageCopy;
}
@@ -339,7 +344,7 @@
if (permanently)
{
log.debug("cluster node for bridge " + this.getName() + " is permanently down");
- discoveryLocator.notifyNodeDown(targetNodeEventUID+1, targetNodeID);
+ discoveryLocator.notifyNodeDown(System.currentTimeMillis(), targetNodeID);
}
}
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionImpl.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -143,20 +143,23 @@
private final ClusterManagerInternal manager;
-
+ private final int minLargeMessageSize;
+
+
// Stuff that used to be on the ClusterManager
-
private final Topology topology = new Topology(this);
private volatile ServerLocatorInternal backupServerLocator;
+ private boolean stopping = false;
public ClusterConnectionImpl(final ClusterManagerInternal manager,
final TransportConfiguration[] tcConfigs,
final TransportConfiguration connector,
final SimpleString name,
final SimpleString address,
+ final int minLargeMessageSize,
final long clientFailureCheckPeriod,
final long connectionTTL,
final long retryInterval,
@@ -214,7 +217,7 @@
this.executorFactory = executorFactory;
this.executor = executorFactory.getExecutor();
-
+
this.topology.setExecutor(executor);
this.server = server;
@@ -238,9 +241,19 @@
this.manager = manager;
this.callTimeout = callTimeout;
+
+ this.minLargeMessageSize = minLargeMessageSize;
clusterConnector = new StaticClusterConnector(tcConfigs);
+ backupServerLocator = clusterConnector.createServerLocator(false);
+
+ if (backupServerLocator != null)
+ {
+ backupServerLocator.setReconnectAttempts(-1);
+ backupServerLocator.setInitialConnectAttempts(-1);
+ }
+
if (tcConfigs != null && tcConfigs.length > 0)
{
// a cluster connection will connect to other nodes only if they are directly connected
@@ -258,6 +271,7 @@
final TransportConfiguration connector,
final SimpleString name,
final SimpleString address,
+ final int minLargeMessageSize,
final long clientFailureCheckPeriod,
final long connectionTTL,
final long retryInterval,
@@ -303,6 +317,8 @@
this.retryIntervalMultiplier = retryIntervalMultiplier;
this.maxRetryInterval = maxRetryInterval;
+
+ this.minLargeMessageSize = minLargeMessageSize;
this.reconnectAttempts = reconnectAttempts;
@@ -317,7 +333,7 @@
this.executorFactory = executorFactory;
this.executor = executorFactory.getExecutor();
-
+
this.topology.setExecutor(executor);
this.server = server;
@@ -340,10 +356,18 @@
clusterConnector = new DiscoveryClusterConnector(dg);
+ backupServerLocator = clusterConnector.createServerLocator(true);
+
+ if (backupServerLocator != null)
+ {
+ backupServerLocator.setReconnectAttempts(-1);
+ backupServerLocator.setInitialConnectAttempts(-1);
+ }
+
this.manager = manager;
}
- public void start() throws Exception
+ public void start() throws Exception
{
synchronized (this)
{
@@ -351,10 +375,10 @@
{
return;
}
-
-
+
+ stopping = false;
started = true;
-
+
if (!backup)
{
activate();
@@ -362,7 +386,7 @@
}
}
-
+
public void flushExecutor()
{
Future future = new Future();
@@ -379,7 +403,7 @@
{
return;
}
-
+ stopping = true;
if (log.isDebugEnabled())
{
log.debug(this + "::stopping ClusterConnection");
@@ -419,9 +443,7 @@
props);
managementService.sendNotification(notification);
}
-
-
executor.execute(new Runnable()
{
public void run()
@@ -447,15 +469,8 @@
started = false;
}
-
public void announceBackup()
{
- this.backupServerLocator = clusterConnector.createServerLocator(false);
-
- backupServerLocator.setReconnectAttempts(-1);
- backupServerLocator.setInitialConnectAttempts(-1);
-
-
executor.execute(new Runnable()
{
public void run()
@@ -491,7 +506,7 @@
{
return topology.getMember(manager.getNodeId());
}
-
+
public void addClusterTopologyListener(final ClusterTopologyListener listener, final boolean clusterConnection)
{
topology.addClusterTopologyListener(listener);
@@ -509,7 +524,7 @@
{
return topology;
}
-
+
public void nodeAnnounced(final long uniqueEventID,
final String nodeID,
final Pair<TransportConfiguration, TransportConfiguration> connectorPair,
@@ -571,6 +586,16 @@
return server;
}
+ public boolean isNodeActive(String nodeId)
+ {
+ MessageFlowRecord rec = records.get(nodeId);
+ if (rec == null)
+ {
+ return false;
+ }
+ return rec.getBridge().isConnected();
+ }
+
public Map<String, String> getNodes()
{
synchronized (records)
@@ -600,7 +625,7 @@
}
backup = false;
-
+
topology.updateAsLive(manager.getNodeId(), new TopologyMember(connector, null));
if (backupServerLocator != null)
@@ -617,8 +642,6 @@
backupServerLocator = null;
}
-
-
serverLocator = clusterConnector.createServerLocator(true);
if (serverLocator != null)
@@ -686,6 +709,10 @@
public void nodeDown(final long eventUID, final String nodeID)
{
+ if (stopping)
+ {
+ return;
+ }
if (log.isDebugEnabled())
{
log.debug(this + " receiving nodeDown for nodeID=" + nodeID, new Exception("trace"));
@@ -721,6 +748,10 @@
final Pair<TransportConfiguration, TransportConfiguration> connectorPair,
final boolean last)
{
+ if (stopping)
+ {
+ return;
+ }
if (log.isDebugEnabled())
{
String ClusterTestBase = "receiving nodeUP for nodeID=";
@@ -821,13 +852,13 @@
}
}
}
-
+
public synchronized void informTopology()
{
String nodeID = server.getNodeID().toString();
-
+
TopologyMember localMember;
-
+
if (backup)
{
localMember = new TopologyMember(null, connector);
@@ -840,7 +871,6 @@
topology.updateAsLive(nodeID, localMember);
}
-
private void createNewRecord(final long eventUID,
final String targetNodeID,
final TransportConfiguration connector,
@@ -848,22 +878,22 @@
final Queue queue,
final boolean start) throws Exception
{
- final ServerLocatorInternal targetLocator = new ServerLocatorImpl(topology, false, connector);
-
+ final ServerLocatorInternal targetLocator = new ServerLocatorImpl(topology, true, connector);
+
String nodeId;
-
+
synchronized (this)
{
if (!started)
{
return;
}
-
+
if (serverLocator == null)
{
return;
}
-
+
nodeId = serverLocator.getNodeID();
}
@@ -881,6 +911,7 @@
targetLocator.setRetryInterval(retryInterval);
targetLocator.setMaxRetryInterval(maxRetryInterval);
targetLocator.setRetryIntervalMultiplier(retryIntervalMultiplier);
+ targetLocator.setMinLargeMessageSize(minLargeMessageSize);
targetLocator.setAfterConnectionInternalListener(this);
@@ -1125,9 +1156,9 @@
public synchronized void onMessage(final ClientMessage message)
{
- if (isTrace)
+ if (log.isDebugEnabled())
{
- log.trace("Flow record on " + clusterConnector + " Receiving message " + message);
+ log.debug("ClusterCommunication::Flow record on " + clusterConnector + " Receiving message " + message);
}
try
{
@@ -1504,8 +1535,9 @@
@Override
public String toString()
{
- return "ClusterConnectionImpl@" + System.identityHashCode(this) +
- "[nodeUUID=" + nodeUUID +
+ return "ClusterConnectionImpl@" + System.identityHashCode(this) +
+ "[nodeUUID=" +
+ nodeUUID +
", connector=" +
connector +
", address=" +
@@ -1555,7 +1587,7 @@
{
log.debug(ClusterConnectionImpl.this + "Creating a serverLocator for " + Arrays.toString(tcConfigs));
}
- ServerLocatorImpl locator = new ServerLocatorImpl(includeTopology?topology:null, true, tcConfigs);
+ ServerLocatorImpl locator = new ServerLocatorImpl(includeTopology ? topology : null, true, tcConfigs);
locator.setClusterConnection(true);
return locator;
}
@@ -1587,7 +1619,7 @@
public ServerLocatorInternal createServerLocator(boolean includeTopology)
{
- ServerLocatorImpl locator = new ServerLocatorImpl(includeTopology?topology:null, true, dg);
+ ServerLocatorImpl locator = new ServerLocatorImpl(includeTopology ? topology : null, true, dg);
return locator;
}
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/ClusterManagerImpl.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/ClusterManagerImpl.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/ClusterManagerImpl.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -470,6 +470,7 @@
serverLocator.setClientFailureCheckPeriod(config.getClientFailureCheckPeriod());
serverLocator.setBlockOnDurableSend(!config.isUseDuplicateDetection());
serverLocator.setBlockOnNonDurableSend(!config.isUseDuplicateDetection());
+ serverLocator.setMinLargeMessageSize(config.getMinLargeMessageSize());
if (!config.isUseDuplicateDetection())
{
log.debug("Bridge " + config.getName() +
@@ -622,6 +623,7 @@
connector,
new SimpleString(config.getName()),
new SimpleString(config.getAddress()),
+ config.getMinLargeMessageSize(),
config.getClientFailureCheckPeriod(),
config.getConnectionTTL(),
config.getRetryInterval(),
@@ -659,6 +661,7 @@
connector,
new SimpleString(config.getName()),
new SimpleString(config.getAddress()),
+ config.getMinLargeMessageSize(),
config.getClientFailureCheckPeriod(),
config.getConnectionTTL(),
config.getRetryInterval(),
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/RemoteQueueBindingImpl.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/RemoteQueueBindingImpl.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/cluster/impl/RemoteQueueBindingImpl.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -311,6 +311,11 @@
buff.putLong(remoteQueueID);
message.putBytesProperty(idsHeaderName, ids);
+
+ if (log.isTraceEnabled())
+ {
+ log.trace("Adding remoteQueue ID = " + remoteQueueID + " into message=" + message + " store-forward-queue=" + storeAndForwardQueue);
+ }
}
}
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/DivertImpl.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/DivertImpl.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/DivertImpl.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -84,6 +84,11 @@
// properly on ack, since the original address will be overwritten
// TODO we can optimise this so it doesn't copy if it's not routed anywhere else
+
+ if (log.isTraceEnabled())
+ {
+ log.trace("Diverting message " + message + " into " + this);
+ }
long id = storageManager.generateUniqueID();
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -317,6 +317,7 @@
public synchronized void start() throws Exception
{
+ log.debug("Starting server " + this);
OperationContextImpl.clearContext();
try
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/QueueImpl.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/QueueImpl.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/QueueImpl.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -461,14 +461,14 @@
{
if (isTrace)
{
- log.trace("Force delivery scheduling depage");
+ log.trace("Force delivery scheduling depage");
}
scheduleDepage();
}
if (isTrace)
{
- log.trace("Force delivery deliverying async");
+ log.trace("Force delivery deliverying async");
}
deliverAsync();
@@ -2255,7 +2255,7 @@
}
catch (Exception e)
{
- QueueImpl.log.warn("Unable to decrement reference counting", e);
+ QueueImpl.log.warn("Unable to decrement reference counting", e);
}
}
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/jms/client/HornetQConnectionFactory.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/jms/client/HornetQConnectionFactory.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/jms/client/HornetQConnectionFactory.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -700,6 +700,24 @@
// Private --------------------------------------------------------------------------------------
+ /* (non-Javadoc)
+ * @see java.lang.Object#toString()
+ */
+ @Override
+ public String toString()
+ {
+ return "HornetQConnectionFactory [serverLocator=" + serverLocator +
+ ", clientID=" +
+ clientID +
+ ", dupsOKBatchSize=" +
+ dupsOKBatchSize +
+ ", transactionBatchSize=" +
+ transactionBatchSize +
+ ", readOnly=" +
+ readOnly +
+ "]";
+ }
+
private void checkWrite()
{
if (readOnly)
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/HornetQXAResourceWrapper.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/HornetQXAResourceWrapper.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/HornetQXAResourceWrapper.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -13,17 +13,15 @@
package org.hornetq.jms.server.recovery;
-import java.util.Map;
+import java.util.Arrays;
import javax.transaction.xa.XAException;
import javax.transaction.xa.XAResource;
import javax.transaction.xa.Xid;
import org.hornetq.api.core.HornetQException;
-import org.hornetq.api.core.TransportConfiguration;
import org.hornetq.api.core.client.ClientSession;
import org.hornetq.api.core.client.ClientSessionFactory;
-import org.hornetq.api.core.client.HornetQClient;
import org.hornetq.api.core.client.ServerLocator;
import org.hornetq.api.core.client.SessionFailureListener;
import org.hornetq.core.logging.Logger;
@@ -53,19 +51,25 @@
private static final Object lock = new Object();
private ServerLocator serverLocator;
-
+
private ClientSessionFactory csf;
private XAResource delegate;
private XARecoveryConfig[] xaRecoveryConfigs;
- //private TransportConfiguration currentConnection;
+ // private TransportConfiguration currentConnection;
public HornetQXAResourceWrapper(XARecoveryConfig... xaRecoveryConfigs)
{
+ this.xaRecoveryConfigs = xaRecoveryConfigs;
- this.xaRecoveryConfigs = xaRecoveryConfigs;
+ if (log.isDebugEnabled())
+ {
+ log.debug("Recovery configured with " + Arrays.toString(xaRecoveryConfigs) +
+ ", instance=" +
+ System.identityHashCode(this));
+ }
}
public Xid[] recover(final int flag) throws XAException
@@ -74,10 +78,18 @@
HornetQXAResourceWrapper.log.debug("Recover " + xaResource);
try
{
- return xaResource.recover(flag);
+ Xid[] xids = xaResource.recover(flag);
+
+ if (log.isDebugEnabled() && xids != null && xids.length > 0)
+ {
+ log.debug("Recovering these following IDs " + Arrays.toString(xids) + " at " + this);
+ }
+
+ return xids;
}
catch (XAException e)
{
+ log.warn(e.getMessage(), e);
throw check(e);
}
}
@@ -214,7 +226,8 @@
public void connectionFailed(final HornetQException me, boolean failedOver)
{
- HornetQXAResourceWrapper.log.warn("Notified of connection failure in xa recovery connectionFactory for provider " + csf + " will attempt reconnect on next pass",
+ HornetQXAResourceWrapper.log.warn("Notified of connection failure in xa recovery connectionFactory for provider " + csf +
+ " will attempt reconnect on next pass",
me);
close();
}
@@ -244,9 +257,9 @@
if (result == null)
{
- //we should always throw a retry for certain methods comit etc, if not the tx is marked as a heuristic and
- //all chaos is let loose
- if(retry)
+ // we should always throw a retry for certain methods comit etc, if not the tx is marked as a heuristic and
+ // all chaos is let loose
+ if (retry)
{
XAException xae = new XAException("Connection unavailable for xa recovery");
xae.errorCode = XAException.XA_RETRY;
@@ -294,6 +307,10 @@
for (XARecoveryConfig xaRecoveryConfig : xaRecoveryConfigs)
{
+ if (log.isDebugEnabled())
+ {
+ log.debug("Trying to connect recovery on " + xaRecoveryConfig + " of " + Arrays.toString(xaRecoveryConfigs));
+ }
ClientSession cs = null;
@@ -308,7 +325,13 @@
}
else
{
- cs = csf.createSession(xaRecoveryConfig.getUsername(), xaRecoveryConfig.getPassword(), true, false, false, false, 1);
+ cs = csf.createSession(xaRecoveryConfig.getUsername(),
+ xaRecoveryConfig.getPassword(),
+ true,
+ false,
+ false,
+ false,
+ 1);
}
}
catch (HornetQException e)
@@ -323,10 +346,29 @@
}
return delegate;
- }
+ }
+ log.warn("Can't connect to any hornetq server on recovery " + Arrays.toString(xaRecoveryConfigs));
throw new HornetQException(HornetQException.NOT_CONNECTED);
}
+ /* (non-Javadoc)
+ * @see java.lang.Object#toString()
+ */
+ @Override
+ public String toString()
+ {
+ return "HornetQXAResourceWrapper [serverLocator=" + serverLocator +
+ ", csf=" +
+ csf +
+ ", delegate=" +
+ delegate +
+ ", xaRecoveryConfigs=" +
+ Arrays.toString(xaRecoveryConfigs) +
+ ", instance=" +
+ System.identityHashCode(this) +
+ "]";
+ }
+
/**
* Close the connection
*/
@@ -366,6 +408,8 @@
*/
protected XAException check(final XAException e) throws XAException
{
+ log.warn(e.getMessage(), e);
+
if (e.errorCode == XAException.XA_RETRY)
{
close();
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/XARecoveryConfig.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/XARecoveryConfig.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/jms/server/recovery/XARecoveryConfig.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -65,8 +65,24 @@
return true;
}
+
+
+ /* (non-Javadoc)
+ * @see java.lang.Object#toString()
+ */
@Override
+ public String toString()
+ {
+ return "XARecoveryConfig [hornetQConnectionFactory=" + hornetQConnectionFactory +
+ ", username=" +
+ username +
+ ", password=" +
+ password +
+ "]";
+ }
+
+ @Override
public int hashCode()
{
int result = hornetQConnectionFactory != null ? hornetQConnectionFactory.hashCode() : 0;
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/ra/HornetQRAManagedConnectionFactory.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/ra/HornetQRAManagedConnectionFactory.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/ra/HornetQRAManagedConnectionFactory.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -138,6 +138,11 @@
cm);
}
+ if (connectionFactory == null)
+ {
+ connectionFactory = ra.createHornetQConnectionFactory(mcfProperties);
+ resourceRecovery = ra.getRecoveryManager().register(connectionFactory, null, null);
+ }
return cf;
}
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/ra/HornetQResourceAdapter.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/ra/HornetQResourceAdapter.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/ra/HornetQResourceAdapter.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -218,6 +218,18 @@
this.ctx = ctx;
+ if (!configured.getAndSet(true))
+ {
+ try
+ {
+ setup();
+ }
+ catch (HornetQException e)
+ {
+ throw new ResourceAdapterInternalException("Unable to create activation", e);
+ }
+ }
+
HornetQResourceAdapter.log.info("HornetQ resource adaptor started");
}
Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/ra/recovery/RecoveryManager.java
===================================================================
--- branches/Branch_2_2_AS7/src/main/org/hornetq/ra/recovery/RecoveryManager.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/src/main/org/hornetq/ra/recovery/RecoveryManager.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -54,6 +54,8 @@
public HornetQResourceRecovery register(HornetQConnectionFactory factory, String userName, String password)
{
+ log.debug("registering recovery for factory : " + factory);
+
if(!isRegistered(factory) && registry != null)
{
XARecoveryConfig xaRecoveryConfig = new XARecoveryConfig(factory, userName, password);
Modified: branches/Branch_2_2_AS7/tests/config/ConfigurationTest-full-config.xml
===================================================================
--- branches/Branch_2_2_AS7/tests/config/ConfigurationTest-full-config.xml 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/config/ConfigurationTest-full-config.xml 2011-11-08 00:45:05 UTC (rev 11666)
@@ -147,6 +147,7 @@
<forwarding-address>bridge-forwarding-address1</forwarding-address>
<filter string="sku > 1"/>
<transformer-class-name>org.foo.BridgeTransformer</transformer-class-name>
+ <min-large-message-size>4</min-large-message-size>
<retry-interval>3</retry-interval>
<retry-interval-multiplier>0.2</retry-interval-multiplier>
<reconnect-attempts>2</reconnect-attempts>
@@ -166,6 +167,7 @@
<cluster-connection name="cluster-connection1">
<address>queues1</address>
<connector-ref>connector1</connector-ref>
+ <min-large-message-size>321</min-large-message-size>
<call-timeout>123</call-timeout>
<retry-interval>3</retry-interval>
<use-duplicate-detection>true</use-duplicate-detection>
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/IncompatibleVersionTest.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/IncompatibleVersionTest.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/IncompatibleVersionTest.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -14,14 +14,10 @@
package org.hornetq.tests.integration.client;
import static org.hornetq.tests.util.RandomUtil.randomString;
-import org.hornetq.tests.util.SpawnedVMSupport;
import java.io.FileOutputStream;
import java.io.InputStream;
import java.util.Properties;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
import org.hornetq.api.core.HornetQException;
import org.hornetq.api.core.TransportConfiguration;
@@ -37,12 +33,12 @@
import org.hornetq.core.protocol.core.Packet;
import org.hornetq.core.protocol.core.impl.wireformat.CreateSessionMessage;
import org.hornetq.core.protocol.core.impl.wireformat.CreateSessionResponseMessage;
-import org.hornetq.core.remoting.impl.invm.InVMConnectorFactory;
import org.hornetq.core.remoting.server.impl.RemotingServiceImpl;
import org.hornetq.core.server.HornetQServer;
import org.hornetq.core.server.HornetQServers;
import org.hornetq.core.version.impl.VersionImpl;
import org.hornetq.tests.util.ServiceTestBase;
+import org.hornetq.tests.util.SpawnedVMSupport;
import org.hornetq.utils.VersionLoader;
/**
@@ -55,15 +51,19 @@
public class IncompatibleVersionTest extends ServiceTestBase
{
private static final Logger log = Logger.getLogger(IncompatibleVersionTest.class);
+
// Constants -----------------------------------------------------
// Attributes ----------------------------------------------------
private HornetQServer server;
- private CoreRemotingConnection connection;
private ServerLocator locator;
+ private ClientSessionFactory csf;
+
+ private CoreRemotingConnection connection;
+
// Static --------------------------------------------------------
// Constructors --------------------------------------------------
@@ -73,12 +73,13 @@
@Override
protected void setUp() throws Exception
{
+ super.setUp();
server = createServer(false, false);
server.getConfiguration().setConnectionTTLOverride(500);
server.start();
locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(ServiceTestBase.INVM_CONNECTOR_FACTORY));
- ClientSessionFactory csf = locator.createSessionFactory();
+ csf = locator.createSessionFactory();
connection = csf.getConnection();
}
@@ -86,11 +87,12 @@
@Override
protected void tearDown() throws Exception
{
- connection.destroy();
+ csf.close();
locator.close();
server.stop();
+ super.tearDown();
}
public void testCompatibleClientVersion() throws Exception
@@ -103,20 +105,36 @@
doTestClientVersionCompatibility(false);
}
- public void testCompatibleClientVersionWithRealConnection() throws Exception
+ public void testCompatibleClientVersionWithRealConnection1() throws Exception
{
- assertTrue(doTestClientVersionCompatibilityWithRealConnection("1-3,5,7-10",1));
- assertTrue(doTestClientVersionCompatibilityWithRealConnection("1-3,5,7-10",5));
- assertTrue(doTestClientVersionCompatibilityWithRealConnection("1-3,5,7-10",10));
+ assertTrue(doTestClientVersionCompatibilityWithRealConnection("1-3,5,7-10", 1));
}
- public void testIncompatibleClientVersionWithRealConnection() throws Exception
+ public void testCompatibleClientVersionWithRealConnection2() throws Exception
{
- assertFalse(doTestClientVersionCompatibilityWithRealConnection("1-3,5,7-10",0));
- assertFalse(doTestClientVersionCompatibilityWithRealConnection("1-3,5,7-10",4));
- assertFalse(doTestClientVersionCompatibilityWithRealConnection("1-3,5,7-10",100));
+ assertTrue(doTestClientVersionCompatibilityWithRealConnection("1-3,5,7-10", 5));
}
-
+
+ public void testCompatibleClientVersionWithRealConnection3() throws Exception
+ {
+ assertTrue(doTestClientVersionCompatibilityWithRealConnection("1-3,5,7-10", 10));
+ }
+
+ public void testIncompatibleClientVersionWithRealConnection1() throws Exception
+ {
+ assertFalse(doTestClientVersionCompatibilityWithRealConnection("1-3,5,7-10", 0));
+ }
+
+ public void testIncompatibleClientVersionWithRealConnection2() throws Exception
+ {
+ assertFalse(doTestClientVersionCompatibilityWithRealConnection("1-3,5,7-10", 4));
+ }
+
+ public void testIncompatibleClientVersionWithRealConnection3() throws Exception
+ {
+ assertFalse(doTestClientVersionCompatibilityWithRealConnection("1-3,5,7-10", 100));
+ }
+
private void doTestClientVersionCompatibility(boolean compatible) throws Exception
{
Channel channel1 = connection.getChannel(1, -1);
@@ -175,61 +193,69 @@
{
String propFileName = "compatibility-test-hornetq-version.properties";
String serverStartedString = "IncompatibleVersionTest---server---started";
-
+
Properties prop = new Properties();
InputStream in = VersionImpl.class.getClassLoader().getResourceAsStream("hornetq-version.properties");
prop.load(in);
prop.setProperty("hornetq.version.compatibleVersionList", verList);
prop.setProperty("hornetq.version.incrementingVersion", Integer.toString(ver));
prop.store(new FileOutputStream("tests/tmpfiles/" + propFileName), null);
-
+
Process server = null;
boolean result = false;
try
{
server = SpawnedVMSupport.spawnVM("org.hornetq.tests.integration.client.IncompatibleVersionTest",
- new String[]{"-D" + VersionLoader.VERSION_PROP_FILE_KEY + "=" + propFileName},
+ new String[] { "-D" + VersionLoader.VERSION_PROP_FILE_KEY +
+ "=" +
+ propFileName },
"server",
serverStartedString);
Thread.sleep(2000);
-
+
Process client = SpawnedVMSupport.spawnVM("org.hornetq.tests.integration.client.IncompatibleVersionTest",
- new String[]{"-D" + VersionLoader.VERSION_PROP_FILE_KEY + "=" + propFileName},
+ new String[] { "-D" + VersionLoader.VERSION_PROP_FILE_KEY +
+ "=" +
+ propFileName },
"client");
-
- if(client.waitFor() == 0)
+
+ if (client.waitFor() == 0)
{
result = true;
}
}
finally
{
- if(server != null)
+ if (server != null)
{
try
{
server.destroy();
}
- catch(Throwable t) {/* ignore */}
+ catch (Throwable t)
+ {/* ignore */
+ }
}
}
-
+
return result;
}
-
+
private static class ServerStarter
{
public void perform(String startedString) throws Exception
{
Configuration conf = new ConfigurationImpl();
conf.setSecurityEnabled(false);
- conf.getAcceptorConfigurations().add(new TransportConfiguration("org.hornetq.core.remoting.impl.netty.NettyAcceptorFactory"));
+ conf.getAcceptorConfigurations()
+ .add(new TransportConfiguration("org.hornetq.core.remoting.impl.netty.NettyAcceptorFactory"));
HornetQServer server = HornetQServers.newHornetQServer(conf, false);
server.start();
-
+
log.info("### server: " + startedString);
}
}
+
private static class ClientStarter
{
public void perform() throws Exception
@@ -243,15 +269,15 @@
locator.close();
}
}
-
+
public static void main(String[] args) throws Exception
{
- if(args[0].equals("server"))
+ if (args[0].equals("server"))
{
ServerStarter ss = new ServerStarter();
ss.perform(args[1]);
}
- else if(args[0].equals("client"))
+ else if (args[0].equals("client"))
{
ClientStarter cs = new ClientStarter();
cs.perform();
@@ -261,7 +287,7 @@
throw new Exception("args[0] must be \"server\" or \"client\"");
}
}
-
+
// Package protected ---------------------------------------------
// Protected -----------------------------------------------------
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/InterruptedLargeMessageTest.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/InterruptedLargeMessageTest.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/InterruptedLargeMessageTest.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -13,36 +13,30 @@
package org.hornetq.tests.integration.client;
-import java.io.ByteArrayOutputStream;
import java.util.HashMap;
import javax.transaction.xa.XAResource;
import javax.transaction.xa.Xid;
-import junit.framework.Assert;
-
import org.hornetq.api.core.HornetQException;
import org.hornetq.api.core.Interceptor;
import org.hornetq.api.core.Message;
import org.hornetq.api.core.SimpleString;
-import org.hornetq.api.core.client.*;
-import org.hornetq.core.client.impl.ClientConsumerInternal;
-import org.hornetq.core.config.Configuration;
+import org.hornetq.api.core.client.ClientConsumer;
+import org.hornetq.api.core.client.ClientMessage;
+import org.hornetq.api.core.client.ClientProducer;
+import org.hornetq.api.core.client.ClientSession;
+import org.hornetq.api.core.client.ClientSessionFactory;
+import org.hornetq.api.core.client.HornetQClient;
+import org.hornetq.api.core.client.ServerLocator;
import org.hornetq.core.logging.Logger;
-import org.hornetq.core.persistence.impl.journal.JournalStorageManager;
-import org.hornetq.core.persistence.impl.journal.LargeServerMessageImpl;
-import org.hornetq.core.persistence.impl.journal.OperationContextImpl;
import org.hornetq.core.protocol.core.Packet;
import org.hornetq.core.protocol.core.impl.wireformat.SessionSendContinuationMessage;
-import org.hornetq.core.server.HornetQServer;
-import org.hornetq.core.server.Queue;
import org.hornetq.core.server.ServerSession;
import org.hornetq.core.server.impl.ServerSessionImpl;
import org.hornetq.core.settings.impl.AddressSettings;
import org.hornetq.spi.core.protocol.RemotingConnection;
import org.hornetq.tests.integration.largemessage.LargeMessageTestBase;
-import org.hornetq.tests.util.RandomUtil;
-import org.hornetq.tests.util.UnitTestCase;
/**
* A LargeMessageTest
@@ -332,9 +326,9 @@
server.stop(false);
server.start();
- server.stop();
-
validateNoFilesOnLargeDir();
+
+ server.stop();
}
finally
{
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/JournalCrashTest.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/JournalCrashTest.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/JournalCrashTest.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -127,20 +127,15 @@
{
try
{
- int start = 4;
- int end = 8;
+ int start = Integer.parseInt(arg[1]);
- if (arg.length > 0)
- {
- start = Integer.parseInt(arg[0]);
- }
+ int end = Integer.parseInt(arg[2]);
+
+ long timeStart = Long.parseLong(arg[0]);
- if (arg.length > 1)
- {
- end = Integer.parseInt(arg[1]);
- }
-
JournalCrashTest restart = new JournalCrashTest();
+
+ restart.setTimeStart(timeStart);
restart.startServer();
@@ -196,10 +191,10 @@
public void testRestartJournal() throws Throwable
{
- runExternalProcess(0, JournalCrashTest.FIRST_RUN);
- runExternalProcess(JournalCrashTest.FIRST_RUN, JournalCrashTest.SECOND_RUN);
- runExternalProcess(JournalCrashTest.SECOND_RUN, JournalCrashTest.THIRD_RUN);
- runExternalProcess(JournalCrashTest.THIRD_RUN, JournalCrashTest.FOURTH_RUN);
+ runExternalProcess(getTimeStart(), 0, JournalCrashTest.FIRST_RUN);
+ runExternalProcess(getTimeStart(), JournalCrashTest.FIRST_RUN, JournalCrashTest.SECOND_RUN);
+ runExternalProcess(getTimeStart(), JournalCrashTest.SECOND_RUN, JournalCrashTest.THIRD_RUN);
+ runExternalProcess(getTimeStart(), JournalCrashTest.THIRD_RUN, JournalCrashTest.FOURTH_RUN);
printJournal();
@@ -241,7 +236,7 @@
* @throws Exception
* @throws InterruptedException
*/
- private void runExternalProcess(final int start, final int end) throws Exception, InterruptedException
+ private void runExternalProcess(final long timeStart, final int start, final int end) throws Exception, InterruptedException
{
System.err.println("running external process...");
Process process = SpawnedVMSupport.spawnVM(this.getClass().getCanonicalName(),
@@ -249,6 +244,7 @@
new String[] {},
true,
true,
+ Long.toString(timeStart),
Integer.toString(start),
Integer.toString(end));
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/LargeMessageCompressTest.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/LargeMessageCompressTest.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/LargeMessageCompressTest.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -159,8 +159,10 @@
msg1.setOutputStream(output);
- msg1.waitOutputStreamCompletion(0);
+ msg1.waitOutputStreamCompletion(0);
+ output.close();
+
msg1.acknowledge();
session.commit();
@@ -176,7 +178,7 @@
byte b = (byte)input.read();
assertEquals("position = " + i, getSamplebyte(i), b);
}
-
+ input.close();
testFile.delete();
validateNoFilesOnLargeDir();
}
@@ -239,6 +241,8 @@
msg1.saveToOutputStream(output);
msg1.acknowledge();
+
+ output.close();
session.commit();
@@ -253,6 +257,7 @@
byte b = (byte)input.read();
assertEquals("position = " + i, getSamplebyte(i), b);
}
+ input.close();
testFile.delete();
validateNoFilesOnLargeDir();
@@ -358,6 +363,7 @@
byte b = (byte)input.read();
assertEquals("position = " + i, msgs[i], b);
}
+ input.close();
testFile.delete();
validateNoFilesOnLargeDir();
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -74,6 +74,22 @@
return false;
}
+ /**
+ *
+ */
+ public LargeMessageTest()
+ {
+ super();
+ }
+
+ /**
+ * @param test
+ */
+ public LargeMessageTest(String test)
+ {
+ super(test);
+ }
+
public void testRollbackPartiallyConsumedBuffer() throws Exception
{
for (int i = 0 ; i < 1; i++)
@@ -1713,6 +1729,7 @@
100);
}
+
public void testPageOnLargeMessage() throws Exception
{
testPageOnLargeMessage(true, false);
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/PagingOrderTest.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/PagingOrderTest.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/PagingOrderTest.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -917,6 +917,8 @@
conn.close();
server.stop();
+
+ jmsServer.stop();
}
@@ -996,6 +998,8 @@
server.stop();
+
+ jmsServer.stop();
}
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/PagingTest.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/PagingTest.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/PagingTest.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -121,6 +121,8 @@
protected void tearDown() throws Exception
{
locator.close();
+
+ locator = null;
super.tearDown();
}
@@ -263,7 +265,7 @@
PagingTest.PAGE_MAX,
new HashMap<String, AddressSettings>());
server.start();
-
+
waitForServer(server);
queue = server.locateQueue(ADDRESS);
@@ -647,8 +649,7 @@
PagingTest.PAGE_MAX,
new HashMap<String, AddressSettings>());
server.start();
-
-
+
locator = createInVMNonHALocator();
locator.setBlockOnNonDurableSend(true);
locator.setBlockOnDurableSend(true);
@@ -662,7 +663,7 @@
session = sf.createSession(false, false, false);
producer = session.createProducer(PagingTest.ADDRESS);
-
+
for (int i = 0; i < numberOfMessages; i++)
{
message = session.createMessage(true);
@@ -679,9 +680,9 @@
session.commit();
}
}
-
+
session.commit();
-
+
server.stop();
server = createServer(true,
@@ -696,7 +697,7 @@
queue = server.locateQueue(ADDRESS);
- // assertEquals(numberOfMessages, queue.getMessageCount());
+ // assertEquals(numberOfMessages, queue.getMessageCount());
xids = new LinkedList<Xid>();
@@ -728,7 +729,6 @@
sessionConsumer.close();
-
}
finally
{
@@ -1041,50 +1041,56 @@
server.start();
- ServerLocator locator = createInVMNonHALocator();
+ try
+ {
- locator.setBlockOnNonDurableSend(true);
- locator.setBlockOnDurableSend(true);
- locator.setBlockOnAcknowledge(true);
+ ServerLocator locator = createInVMNonHALocator();
- ClientSessionFactory csf = locator.createSessionFactory();
+ locator.setBlockOnNonDurableSend(true);
+ locator.setBlockOnDurableSend(true);
+ locator.setBlockOnAcknowledge(true);
- ClientSession session = csf.createSession();
+ ClientSessionFactory csf = locator.createSessionFactory();
- session.start();
+ ClientSession session = csf.createSession();
- for (int i = 1; i <= 2; i++)
- {
- ClientConsumer cons = session.createConsumer("q" + i);
+ session.start();
- for (int j = 3; j < 6; j++)
+ for (int i = 1; i <= 2; i++)
{
- ClientMessage msg = cons.receive(5000);
+ ClientConsumer cons = session.createConsumer("q" + i);
- assertNotNull(msg);
+ for (int j = 3; j < 6; j++)
+ {
+ ClientMessage msg = cons.receive(5000);
- assertEquals("str-" + j, msg.getStringProperty("id"));
+ assertNotNull(msg);
- msg.acknowledge();
+ assertEquals("str-" + j, msg.getStringProperty("id"));
+
+ msg.acknowledge();
+ }
+
+ session.commit();
+ assertNull(cons.receive(500));
+
}
- session.commit();
- assertNull(cons.receive(500));
+ session.close();
- }
+ long timeout = System.currentTimeMillis() + 5000;
- session.close();
+ while (System.currentTimeMillis() < timeout && server.getPagingManager().getPageStore(ADDRESS).isPaging())
+ {
+ Thread.sleep(100);
+ }
- long timeout = System.currentTimeMillis() + 5000;
-
- while (System.currentTimeMillis() < timeout && server.getPagingManager().getPageStore(ADDRESS).isPaging())
+ locator.close();
+ }
+ finally
{
- Thread.sleep(100);
+ server.stop();
}
-
- locator.close();
-
- server.stop();
}
public void testTwoQueuesOneNoRouting() throws Exception
@@ -1277,27 +1283,28 @@
{
bb.put(getSamplebyte(j));
}
-
+
final AtomicBoolean running = new AtomicBoolean(true);
-
+
class TCount extends Thread
{
Queue queue;
-
+
TCount(Queue queue)
{
this.queue = queue;
}
+
public void run()
{
try
{
while (running.get())
{
- // log.info("Message count = " + queue.getMessageCount() + " on queue " + queue.getName());
+ // log.info("Message count = " + queue.getMessageCount() + " on queue " + queue.getName());
queue.getMessagesAdded();
queue.getMessageCount();
- //log.info("Message added = " + queue.getMessagesAdded() + " on queue " + queue.getName());
+ // log.info("Message added = " + queue.getMessagesAdded() + " on queue " + queue.getName());
Thread.sleep(10);
}
}
@@ -1307,10 +1314,9 @@
}
}
};
-
+
TCount tcount1 = null;
TCount tcount2 = null;
-
try
{
@@ -1337,8 +1343,7 @@
session.createQueue(PagingTest.ADDRESS.toString(), PagingTest.ADDRESS + "-2", null, true);
}
-
-
+
ClientProducer producer = session.createProducer(PagingTest.ADDRESS);
ClientMessage message = null;
@@ -1377,21 +1382,21 @@
PagingTest.PAGE_MAX,
new HashMap<String, AddressSettings>());
server.start();
-
+
Queue queue1 = server.locateQueue(PagingTest.ADDRESS.concat("-1"));
-
+
Queue queue2 = server.locateQueue(PagingTest.ADDRESS.concat("-2"));
-
+
assertNotNull(queue1);
-
+
assertNotNull(queue2);
-
+
assertNotSame(queue1, queue2);
tcount1 = new TCount(queue1);
-
+
tcount2 = new TCount(queue2);
-
+
tcount1.start();
tcount2.start();
@@ -1500,19 +1505,19 @@
finally
{
running.set(false);
-
+
if (tcount1 != null)
{
tcount1.interrupt();
tcount1.join();
}
-
+
if (tcount2 != null)
{
tcount2.interrupt();
tcount2.join();
}
-
+
try
{
server.stop();
@@ -2510,7 +2515,7 @@
producerThread.start();
- assertTrue(ready.await(10, TimeUnit.SECONDS));
+ assertTrue(ready.await(100, TimeUnit.SECONDS));
ClientConsumer consumer = session.createConsumer(PagingTest.ADDRESS);
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeReconnectTest.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeReconnectTest.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeReconnectTest.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -124,6 +124,7 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE,
HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
HornetQClient.DEFAULT_CONNECTION_TTL,
retryInterval,
@@ -268,6 +269,7 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE,
HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
HornetQClient.DEFAULT_CONNECTION_TTL,
retryInterval,
@@ -396,6 +398,7 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE,
HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
HornetQClient.DEFAULT_CONNECTION_TTL,
retryInterval,
@@ -540,6 +543,7 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE,
clientFailureCheckPeriod,
HornetQClient.DEFAULT_CONNECTION_TTL,
retryInterval,
@@ -679,6 +683,7 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE,
HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
HornetQClient.DEFAULT_CONNECTION_TTL,
retryInterval,
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeStartTest.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeStartTest.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeStartTest.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -109,6 +109,7 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE,
HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
HornetQClient.DEFAULT_CONNECTION_TTL,
1000,
@@ -279,6 +280,7 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE,
HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
HornetQClient.DEFAULT_CONNECTION_TTL,
500,
@@ -496,6 +498,7 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE,
HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
HornetQClient.DEFAULT_CONNECTION_TTL,
1000,
@@ -652,6 +655,7 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE,
HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
HornetQClient.DEFAULT_CONNECTION_TTL,
1000,
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeTest.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeTest.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeTest.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -141,6 +141,7 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE,
HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
HornetQClient.DEFAULT_CONNECTION_TTL,
1000,
@@ -349,6 +350,7 @@
forwardAddress,
filterString,
null,
+ HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE,
HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
HornetQClient.DEFAULT_CONNECTION_TTL,
1000,
@@ -538,6 +540,7 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE,
HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
HornetQClient.DEFAULT_CONNECTION_TTL,
100,
@@ -698,6 +701,7 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE,
HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
HornetQClient.DEFAULT_CONNECTION_TTL,
100,
@@ -891,6 +895,7 @@
forwardAddress,
null,
SimpleTransformer.class.getName(),
+ HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE,
HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
HornetQClient.DEFAULT_CONNECTION_TTL,
1000,
@@ -1044,6 +1049,7 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE,
HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
HornetQClient.DEFAULT_CONNECTION_TTL,
1000,
@@ -1290,6 +1296,7 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE,
HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
HornetQClient.DEFAULT_CONNECTION_TTL,
1000,
@@ -1441,6 +1448,7 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE,
HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
HornetQClient.DEFAULT_CONNECTION_TTL,
1000,
@@ -1601,6 +1609,7 @@
// address
null,
null,
+ HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE,
HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
HornetQClient.DEFAULT_CONNECTION_TTL,
1000,
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeWithDiscoveryGroupStartTest.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeWithDiscoveryGroupStartTest.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeWithDiscoveryGroupStartTest.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -116,6 +116,7 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE,
HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
HornetQClient.DEFAULT_CONNECTION_TTL,
1000,
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeWithPagingTest.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeWithPagingTest.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeWithPagingTest.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -113,6 +113,7 @@
forwardAddress,
null,
null,
+ HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE,
HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
HornetQClient.DEFAULT_CONNECTION_TTL,
retryInterval,
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusterTestBase.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusterTestBase.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusterTestBase.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -1921,6 +1921,7 @@
ClusterConnectionConfiguration clusterConf = new ClusterConnectionConfiguration(name,
address,
connectorFrom.getName(),
+ HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE,
ConfigurationImpl.DEFAULT_CLUSTER_FAILURE_CHECK_PERIOD,
ConfigurationImpl.DEFAULT_CLUSTER_CONNECTION_TTL,
retryInterval,
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusteredRequestResponseTest.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusteredRequestResponseTest.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusteredRequestResponseTest.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -96,8 +96,10 @@
/*
* Don't wait for the response queue bindings to get to the other side
+ *
+ * TODO: I believe this test is invalid. I'm just ignoring it for now. It will probably go away
*/
- public void testRequestResponseNoWaitForBindings() throws Exception
+ public void invalidTest_testRequestResponseNoWaitForBindings() throws Exception
{
setupCluster();
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/MessageRedistributionTest.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/MessageRedistributionTest.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/MessageRedistributionTest.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -494,10 +494,11 @@
{
for (int i = 0; i < 10; i++)
{
+ log.info("#test " + i);
setupCluster(false);
startServers(0, 1, 2);
-
+
setupSessionFactory(0, isNetty());
setupSessionFactory(1, isNetty());
setupSessionFactory(2, isNetty());
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/TwoWayTwoNodeClusterTest.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/TwoWayTwoNodeClusterTest.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/distribution/TwoWayTwoNodeClusterTest.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -141,7 +141,6 @@
for (int i = 0; i < 10; i++)
{
- Thread.sleep(10);
log.info("Sleep #test " + i);
log.info("#stop #test #" + i);
stopServers(1);
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/topology/HAClientTopologyWithDiscoveryTest.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/topology/HAClientTopologyWithDiscoveryTest.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/topology/HAClientTopologyWithDiscoveryTest.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -33,7 +33,13 @@
{
return false;
}
+
+ protected boolean isFileStorage()
+ {
+ return false;
+ }
+
protected void setupCluster() throws Exception
{
setupCluster(false);
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/topology/IsolatedTopologyTest.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/topology/IsolatedTopologyTest.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/topology/IsolatedTopologyTest.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -28,7 +28,6 @@
import org.hornetq.core.remoting.impl.netty.TransportConstants;
import org.hornetq.core.server.HornetQServer;
import org.hornetq.tests.util.ServiceTestBase;
-import org.hornetq.tests.util.UnitTestCase;
/**
* A IsolatedTopologyTest
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/topology/TopologyClusterTestBase.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/topology/TopologyClusterTestBase.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/cluster/topology/TopologyClusterTestBase.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -17,6 +17,7 @@
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.CountDownLatch;
import org.hornetq.api.core.HornetQException;
@@ -70,6 +71,11 @@
abstract protected boolean isNetty() throws Exception;
+ protected boolean isFileStorage()
+ {
+ return false;
+ }
+
@Override
protected void setUp() throws Exception
{
@@ -114,11 +120,12 @@
{
ok = (ok && actual.contains(nodeIDs[expected[i]]));
}
- if (ok)
+ if (ok)
{
return;
}
- } while(System.currentTimeMillis() - start < 5000);
+ }
+ while (System.currentTimeMillis() - start < 5000);
fail("did not contain all expected node ID: " + actual);
}
@@ -145,8 +152,8 @@
{
if (e.getCode() == HornetQException.OBJECT_CLOSED || e.getCode() == HornetQException.UNBLOCKED)
{
- ClientSessionFactory sf = locator.createSessionFactory();
- return sf.createSession();
+ ClientSessionFactory sf = locator.createSessionFactory();
+ return sf.createSession();
}
else
{
@@ -174,7 +181,14 @@
for (ClusterConnection clusterConn : clusterManager.getClusterConnections())
{
- nodesCount += clusterConn.getNodes().size();
+ Map<String, String> nodes = clusterConn.getNodes();
+ for (String id : nodes.keySet())
+ {
+ if (clusterConn.isNodeActive(id))
+ {
+ nodesCount++;
+ }
+ }
}
if (nodesCount == count)
@@ -185,85 +199,92 @@
Thread.sleep(10);
}
while (System.currentTimeMillis() - start < WAIT_TIMEOUT);
-
+
log.error(clusterDescription(servers[node]));
throw new IllegalStateException("Timed out waiting for cluster connections ");
}
+
public void testReceiveNotificationsWhenOtherNodesAreStartedAndStopped() throws Throwable
{
startServers(0);
ServerLocator locator = createHAServerLocator();
-
- ((ServerLocatorImpl)locator).getTopology().setOwner("testReceive");
- final List<String> nodes = new ArrayList<String>();
- final CountDownLatch upLatch = new CountDownLatch(5);
- final CountDownLatch downLatch = new CountDownLatch(4);
+ try
+ {
+ ((ServerLocatorImpl)locator).getTopology().setOwner("testReceive");
- locator.addClusterTopologyListener(new ClusterTopologyListener()
- {
- public void nodeUP(final long uniqueEventID,
- String nodeID,
- Pair<TransportConfiguration, TransportConfiguration> connectorPair,
- boolean last)
+ final List<String> nodes = new ArrayList<String>();
+ final CountDownLatch upLatch = new CountDownLatch(5);
+ final CountDownLatch downLatch = new CountDownLatch(4);
+
+ locator.addClusterTopologyListener(new ClusterTopologyListener()
{
- if(!nodes.contains(nodeID))
+ public void nodeUP(final long uniqueEventID,
+ String nodeID,
+ Pair<TransportConfiguration, TransportConfiguration> connectorPair,
+ boolean last)
{
- System.out.println("Node UP " + nodeID + " added");
- log.info("Node UP " + nodeID + " added");
- nodes.add(nodeID);
- upLatch.countDown();
+ if (!nodes.contains(nodeID))
+ {
+ System.out.println("Node UP " + nodeID + " added");
+ log.info("Node UP " + nodeID + " added");
+ nodes.add(nodeID);
+ upLatch.countDown();
+ }
+ else
+ {
+ System.out.println("Node UP " + nodeID + " was already here");
+ log.info("Node UP " + nodeID + " was already here");
+ }
}
- else
- {
- System.out.println("Node UP " + nodeID + " was already here");
- log.info("Node UP " + nodeID + " was already here");
- }
- }
- public void nodeDown(final long uniqueEventID, String nodeID)
- {
- if (nodes.contains(nodeID))
+ public void nodeDown(final long uniqueEventID, String nodeID)
{
- log.info("Node down " + nodeID + " accepted");
- System.out.println("Node down " + nodeID + " accepted");
- nodes.remove(nodeID);
- downLatch.countDown();
+ if (nodes.contains(nodeID))
+ {
+ log.info("Node down " + nodeID + " accepted");
+ System.out.println("Node down " + nodeID + " accepted");
+ nodes.remove(nodeID);
+ downLatch.countDown();
+ }
+ else
+ {
+ log.info("Node down " + nodeID + " already removed");
+ System.out.println("Node down " + nodeID + " already removed");
+ }
}
- else
- {
- log.info("Node down " + nodeID + " already removed");
- System.out.println("Node down " + nodeID + " already removed");
- }
- }
- });
+ });
- ClientSessionFactory sf = locator.createSessionFactory();
+ ClientSessionFactory sf = locator.createSessionFactory();
- startServers(1, 4, 3, 2);
- String[] nodeIDs = getNodeIDs(0, 1, 2, 3, 4);
+ startServers(1, 4, 3, 2);
+ String[] nodeIDs = getNodeIDs(0, 1, 2, 3, 4);
- assertTrue("Was not notified that all servers are UP", upLatch.await(10, SECONDS));
- checkContains(new int[] { 0, 1, 4, 3, 2 }, nodeIDs, nodes);
+ assertTrue("Was not notified that all servers are UP", upLatch.await(10, SECONDS));
+ checkContains(new int[] { 0, 1, 4, 3, 2 }, nodeIDs, nodes);
- waitForClusterConnections(0, 4);
- waitForClusterConnections(1, 4);
- waitForClusterConnections(2, 4);
- waitForClusterConnections(3, 4);
- waitForClusterConnections(4, 4);
+ waitForClusterConnections(0, 4);
+ waitForClusterConnections(1, 4);
+ waitForClusterConnections(2, 4);
+ waitForClusterConnections(3, 4);
+ waitForClusterConnections(4, 4);
- stopServers(2, 3, 1, 4);
+ stopServers(2, 3, 1, 4);
- assertTrue("Was not notified that all servers are DOWN", downLatch.await(10, SECONDS));
- checkContains(new int[] { 0 }, nodeIDs, nodes);
+ assertTrue("Was not notified that all servers are DOWN", downLatch.await(10, SECONDS));
+ checkContains(new int[] { 0 }, nodeIDs, nodes);
- sf.close();
-
- locator.close();
-
- stopServers(0);
+ sf.close();
+ }
+ finally
+ {
+ locator.close();
+
+ stopServers(0);
+ }
+
}
public void testReceiveNotifications() throws Throwable
@@ -273,73 +294,81 @@
ServerLocator locator = createHAServerLocator();
- final List<String> nodes = new ArrayList<String>();
- final CountDownLatch upLatch = new CountDownLatch(5);
- final CountDownLatch downLatch = new CountDownLatch(4);
+ try
+ {
- locator.addClusterTopologyListener(new ClusterTopologyListener()
- {
- public void nodeUP(final long uniqueEventID,
- String nodeID,
- Pair<TransportConfiguration, TransportConfiguration> connectorPair,
- boolean last)
+ waitForClusterConnections(0, 4);
+ waitForClusterConnections(1, 4);
+ waitForClusterConnections(2, 4);
+ waitForClusterConnections(3, 4);
+ waitForClusterConnections(4, 4);
+
+ final List<String> nodes = new ArrayList<String>();
+ final CountDownLatch upLatch = new CountDownLatch(5);
+ final CountDownLatch downLatch = new CountDownLatch(4);
+
+ locator.addClusterTopologyListener(new ClusterTopologyListener()
{
- if (!nodes.contains(nodeID))
+ public void nodeUP(final long uniqueEventID,
+ String nodeID,
+ Pair<TransportConfiguration, TransportConfiguration> connectorPair,
+ boolean last)
{
- nodes.add(nodeID);
- upLatch.countDown();
+ if (!nodes.contains(nodeID))
+ {
+ nodes.add(nodeID);
+ upLatch.countDown();
+ }
}
- }
- public void nodeDown(final long uniqueEventID, String nodeID)
- {
- if (nodes.contains(nodeID))
+ public void nodeDown(final long uniqueEventID, String nodeID)
{
- nodes.remove(nodeID);
- downLatch.countDown();
+ if (nodes.contains(nodeID))
+ {
+ nodes.remove(nodeID);
+ downLatch.countDown();
+ }
}
- }
- });
+ });
- ClientSessionFactory sf = locator.createSessionFactory();
+ ClientSessionFactory sf = locator.createSessionFactory();
- assertTrue("Was not notified that all servers are UP", upLatch.await(10, SECONDS));
- checkContains(new int[] { 0, 1, 2, 3, 4 }, nodeIDs, nodes);
+ assertTrue("Was not notified that all servers are UP", upLatch.await(10, SECONDS));
+ checkContains(new int[] { 0, 1, 2, 3, 4 }, nodeIDs, nodes);
- ClientSession session = sf.createSession();
-
- waitForClusterConnections(0, 4);
- waitForClusterConnections(1, 4);
- waitForClusterConnections(2, 4);
- waitForClusterConnections(3, 4);
- waitForClusterConnections(4, 4);
+ ClientSession session = sf.createSession();
- stopServers(0);
- session = checkSessionOrReconnect(session, locator);
- checkContains(new int[] { 1, 2, 3, 4 }, nodeIDs, nodes);
+ stopServers(0);
+ session = checkSessionOrReconnect(session, locator);
+ checkContains(new int[] { 1, 2, 3, 4 }, nodeIDs, nodes);
- stopServers(2);
- session = checkSessionOrReconnect(session, locator);
- checkContains(new int[] { 1, 3, 4 }, nodeIDs, nodes);
+ stopServers(2);
+ session = checkSessionOrReconnect(session, locator);
+ checkContains(new int[] { 1, 3, 4 }, nodeIDs, nodes);
- stopServers(4);
- session = checkSessionOrReconnect(session, locator);
- checkContains(new int[] { 1, 3 }, nodeIDs, nodes);
+ stopServers(4);
+ session = checkSessionOrReconnect(session, locator);
+ checkContains(new int[] { 1, 3 }, nodeIDs, nodes);
- stopServers(3);
- session = checkSessionOrReconnect(session, locator);
- checkContains(new int[] { 1 }, nodeIDs, nodes);
+ stopServers(3);
+ session = checkSessionOrReconnect(session, locator);
+ checkContains(new int[] { 1 }, nodeIDs, nodes);
- stopServers(1);
-
- assertTrue("Was not notified that all servers are DOWN", downLatch.await(10, SECONDS));
- checkContains(new int[] {}, nodeIDs, nodes);
+ stopServers(1);
- sf.close();
-
- locator.close();
+ assertTrue("Was not notified that all servers are DOWN", downLatch.await(10, SECONDS));
+ checkContains(new int[] {}, nodeIDs, nodes);
+
+ sf.close();
+ }
+ finally
+ {
+ locator.close();
+ }
+
}
+
public void testStopNodes() throws Throwable
{
startServers(0, 1, 2, 3, 4);
@@ -347,80 +376,87 @@
ServerLocator locator = createHAServerLocator();
- final List<String> nodes = new ArrayList<String>();
- final CountDownLatch upLatch = new CountDownLatch(5);
+ try
+ {
+ waitForClusterConnections(0, 4);
+ waitForClusterConnections(1, 4);
+ waitForClusterConnections(2, 4);
+ waitForClusterConnections(3, 4);
+ waitForClusterConnections(4, 4);
- locator.addClusterTopologyListener(new ClusterTopologyListener()
- {
- public void nodeUP(final long uniqueEventID, String nodeID,
- Pair<TransportConfiguration, TransportConfiguration> connectorPair,
- boolean last)
+ final List<String> nodes = new ArrayList<String>();
+ final CountDownLatch upLatch = new CountDownLatch(5);
+
+ locator.addClusterTopologyListener(new ClusterTopologyListener()
{
- if (!nodes.contains(nodeID))
+ public void nodeUP(final long uniqueEventID,
+ String nodeID,
+ Pair<TransportConfiguration, TransportConfiguration> connectorPair,
+ boolean last)
{
- nodes.add(nodeID);
- upLatch.countDown();
+ if (!nodes.contains(nodeID))
+ {
+ nodes.add(nodeID);
+ upLatch.countDown();
+ }
}
- }
- public void nodeDown(final long uniqueEventID, String nodeID)
- {
- if (nodes.contains(nodeID))
+ public void nodeDown(final long uniqueEventID, String nodeID)
{
- nodes.remove(nodeID);
+ if (nodes.contains(nodeID))
+ {
+ nodes.remove(nodeID);
+ }
}
- }
- });
+ });
- ClientSessionFactory sf = locator.createSessionFactory();
+ ClientSessionFactory sf = locator.createSessionFactory();
- assertTrue("Was not notified that all servers are UP", upLatch.await(10, SECONDS));
- checkContains(new int[] { 0, 1, 2, 3, 4 }, nodeIDs, nodes);
+ assertTrue("Was not notified that all servers are UP", upLatch.await(10, SECONDS));
+ checkContains(new int[] { 0, 1, 2, 3, 4 }, nodeIDs, nodes);
- waitForClusterConnections(0, 4);
- waitForClusterConnections(1, 4);
- waitForClusterConnections(2, 4);
- waitForClusterConnections(3, 4);
- waitForClusterConnections(4, 4);
+ ClientSession session = sf.createSession();
- ClientSession session = sf.createSession();
-
- stopServers(0);
- assertFalse(servers[0].isStarted());
- session = checkSessionOrReconnect(session, locator);
- checkContains(new int[] { 1, 2, 3, 4 }, nodeIDs, nodes);
+ stopServers(0);
+ assertFalse(servers[0].isStarted());
+ session = checkSessionOrReconnect(session, locator);
+ checkContains(new int[] { 1, 2, 3, 4 }, nodeIDs, nodes);
- stopServers(2);
- assertFalse(servers[2].isStarted());
- session = checkSessionOrReconnect(session, locator);
- checkContains(new int[] { 1, 3, 4 }, nodeIDs, nodes);
+ stopServers(2);
+ assertFalse(servers[2].isStarted());
+ session = checkSessionOrReconnect(session, locator);
+ checkContains(new int[] { 1, 3, 4 }, nodeIDs, nodes);
- stopServers(4);
- assertFalse(servers[4].isStarted());
- session = checkSessionOrReconnect(session, locator);
- checkContains(new int[] { 1, 3 }, nodeIDs, nodes);
+ stopServers(4);
+ assertFalse(servers[4].isStarted());
+ session = checkSessionOrReconnect(session, locator);
+ checkContains(new int[] { 1, 3 }, nodeIDs, nodes);
- stopServers(3);
- assertFalse(servers[3].isStarted());
+ stopServers(3);
+ assertFalse(servers[3].isStarted());
- session = checkSessionOrReconnect(session, locator);
- checkContains(new int[] { 1 }, nodeIDs, nodes);
+ session = checkSessionOrReconnect(session, locator);
+ checkContains(new int[] { 1 }, nodeIDs, nodes);
- stopServers(1);
- assertFalse(servers[1].isStarted());
- try
+ stopServers(1);
+ assertFalse(servers[1].isStarted());
+ try
+ {
+ session = checkSessionOrReconnect(session, locator);
+ fail();
+ }
+ catch (Exception e)
+ {
+
+ }
+ }
+ finally
{
- session = checkSessionOrReconnect(session, locator);
- fail();
+ locator.close();
}
- catch (Exception e)
- {
- }
-
- locator.close();
}
-
+
public void testMultipleClientSessionFactories() throws Throwable
{
startServers(0, 1, 2, 3, 4);
@@ -428,67 +464,75 @@
ServerLocator locator = createHAServerLocator();
- final List<String> nodes = new ArrayList<String>();
- final CountDownLatch upLatch = new CountDownLatch(5);
- final CountDownLatch downLatch = new CountDownLatch(4);
+ try
+ {
- locator.addClusterTopologyListener(new ClusterTopologyListener()
- {
- public void nodeUP(final long uniqueEventID, String nodeID,
- Pair<TransportConfiguration, TransportConfiguration> connectorPair,
- boolean last)
+ waitForClusterConnections(0, 4);
+ waitForClusterConnections(1, 4);
+ waitForClusterConnections(2, 4);
+ waitForClusterConnections(3, 4);
+ waitForClusterConnections(4, 4);
+
+ final List<String> nodes = new ArrayList<String>();
+ final CountDownLatch upLatch = new CountDownLatch(5);
+ final CountDownLatch downLatch = new CountDownLatch(4);
+
+ locator.addClusterTopologyListener(new ClusterTopologyListener()
{
- if (!nodes.contains(nodeID))
+ public void nodeUP(final long uniqueEventID,
+ String nodeID,
+ Pair<TransportConfiguration, TransportConfiguration> connectorPair,
+ boolean last)
{
- nodes.add(nodeID);
- upLatch.countDown();
+ if (!nodes.contains(nodeID))
+ {
+ nodes.add(nodeID);
+ upLatch.countDown();
+ }
}
- }
- public void nodeDown(final long uniqueEventID, String nodeID)
- {
- if (nodes.contains(nodeID))
+ public void nodeDown(final long uniqueEventID, String nodeID)
{
- nodes.remove(nodeID);
- downLatch.countDown();
+ if (nodes.contains(nodeID))
+ {
+ nodes.remove(nodeID);
+ downLatch.countDown();
+ }
}
- }
- });
+ });
- ClientSessionFactory[] sfs = new ClientSessionFactory[] {
- locator.createSessionFactory(),
- locator.createSessionFactory(),
- locator.createSessionFactory(),
- locator.createSessionFactory(),
- locator.createSessionFactory() };
- assertTrue("Was not notified that all servers are UP", upLatch.await(10, SECONDS));
- checkContains(new int[] { 0, 1, 2, 3, 4 }, nodeIDs, nodes);
+ ClientSessionFactory[] sfs = new ClientSessionFactory[] { locator.createSessionFactory(),
+ locator.createSessionFactory(),
+ locator.createSessionFactory(),
+ locator.createSessionFactory(),
+ locator.createSessionFactory() };
+ assertTrue("Was not notified that all servers are UP", upLatch.await(10, SECONDS));
+ checkContains(new int[] { 0, 1, 2, 3, 4 }, nodeIDs, nodes);
- waitForClusterConnections(0, 4);
- waitForClusterConnections(1, 4);
- waitForClusterConnections(2, 4);
- waitForClusterConnections(3, 4);
- waitForClusterConnections(4, 4);
- //we cant close all of the servers, we need to leave one up to notify us
- stopServers(4, 2, 3, 1);
+ // we cant close all of the servers, we need to leave one up to notify us
+ stopServers(4, 2, 3, 1);
- boolean ok = downLatch.await(10, SECONDS);
- if(!ok)
- {
- System.out.println("TopologyClusterTestBase.testMultipleClientSessionFactories");
+ boolean ok = downLatch.await(10, SECONDS);
+ if (!ok)
+ {
+ log.warn("TopologyClusterTestBase.testMultipleClientSessionFactories will fail");
+ }
+ assertTrue("Was not notified that all servers are Down", ok);
+ checkContains(new int[] { 0 }, nodeIDs, nodes);
+
+ for (int i = 0; i < sfs.length; i++)
+ {
+ ClientSessionFactory sf = sfs[i];
+ sf.close();
+ }
}
- assertTrue("Was not notified that all servers are Down", ok);
- checkContains(new int[] { 0 }, nodeIDs, nodes);
-
- for (int i = 0; i < sfs.length; i++)
+ finally
{
- ClientSessionFactory sf = sfs[i];
- sf.close();
+ locator.close();
+
+ stopServers(0);
}
-
- locator.close();
- stopServers(0);
}
// Private -------------------------------------------------------
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageTestBase.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageTestBase.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageTestBase.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -69,7 +69,18 @@
// Package protected ---------------------------------------------
// Protected -----------------------------------------------------
+
+ public LargeMessageTestBase(String test)
+ {
+ super(test);
+ }
+
+ public LargeMessageTestBase()
+ {
+ super();
+ }
+
@Override
protected void tearDown() throws Exception
{
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/management/BridgeControlTest.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/management/BridgeControlTest.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/management/BridgeControlTest.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -166,6 +166,7 @@
targetQueueConfig.getAddress(),
null,
null,
+ HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE,
HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
HornetQClient.DEFAULT_CONNECTION_TTL,
RandomUtil.randomPositiveLong(),
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/management/BridgeControlUsingCoreTest.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/management/BridgeControlUsingCoreTest.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/management/BridgeControlUsingCoreTest.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -144,6 +144,7 @@
targetQueueConfig.getAddress(),
null,
null,
+ HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE,
HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
HornetQClient.DEFAULT_CONNECTION_TTL,
RandomUtil.randomPositiveLong(),
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerSecurityTest.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerSecurityTest.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerSecurityTest.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -22,12 +22,11 @@
import org.hornetq.core.server.impl.QueueImpl;
import org.hornetq.ra.HornetQResourceAdapter;
import org.hornetq.ra.inflow.HornetQActivationSpec;
+import org.hornetq.tests.util.UnitTestCase;
-import javax.resource.ResourceException;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
/**
* @author <a href="mailto:andy.taylor@jboss.org">Andy Taylor</a>
@@ -44,6 +43,7 @@
public void testSimpleMessageReceivedOnQueueWithSecurityFails() throws Exception
{
HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter();
+ qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY);
MyBootstrapContext ctx = new MyBootstrapContext();
qResourceAdapter.start(ctx);
HornetQActivationSpec spec = new HornetQActivationSpec();
@@ -74,6 +74,7 @@
roles.add(role);
server.getSecurityRepository().addMatch(MDBQUEUEPREFIXED, roles);
HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter();
+ qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY);
MyBootstrapContext ctx = new MyBootstrapContext();
qResourceAdapter.start(ctx);
HornetQActivationSpec spec = new HornetQActivationSpec();
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerTest.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerTest.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerTest.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -20,6 +20,7 @@
import org.hornetq.core.postoffice.impl.LocalQueueBinding;
import org.hornetq.ra.HornetQResourceAdapter;
import org.hornetq.ra.inflow.HornetQActivationSpec;
+import org.hornetq.tests.util.UnitTestCase;
import javax.jms.Message;
import java.util.concurrent.CountDownLatch;
@@ -41,6 +42,7 @@
public void testSimpleMessageReceivedOnQueue() throws Exception
{
HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter();
+ qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY);
MyBootstrapContext ctx = new MyBootstrapContext();
qResourceAdapter.start(ctx);
HornetQActivationSpec spec = new HornetQActivationSpec();
@@ -72,6 +74,7 @@
public void testInvalidAckMode() throws Exception
{
HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter();
+ qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY);
MyBootstrapContext ctx = new MyBootstrapContext();
qResourceAdapter.start(ctx);
HornetQActivationSpec spec = new HornetQActivationSpec();
@@ -90,6 +93,7 @@
public void testSimpleMessageReceivedOnQueueInLocalTX() throws Exception
{
HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter();
+ qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY);
qResourceAdapter.setUseLocalTx(true);
MyBootstrapContext ctx = new MyBootstrapContext();
qResourceAdapter.start(ctx);
@@ -126,6 +130,7 @@
public void testSimpleMessageReceivedOnQueueWithSelector() throws Exception
{
HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter();
+ qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY);
MyBootstrapContext ctx = new MyBootstrapContext();
qResourceAdapter.start(ctx);
HornetQActivationSpec spec = new HornetQActivationSpec();
@@ -162,6 +167,7 @@
public void testEndpointDeactivated() throws Exception
{
HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter();
+ qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY);
MyBootstrapContext ctx = new MyBootstrapContext();
qResourceAdapter.start(ctx);
HornetQActivationSpec spec = new HornetQActivationSpec();
@@ -185,6 +191,7 @@
public void testMaxSessions() throws Exception
{
HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter();
+ qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY);
MyBootstrapContext ctx = new MyBootstrapContext();
qResourceAdapter.start(ctx);
HornetQActivationSpec spec = new HornetQActivationSpec();
@@ -207,6 +214,7 @@
public void testSimpleTopic() throws Exception
{
HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter();
+ qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY);
MyBootstrapContext ctx = new MyBootstrapContext();
qResourceAdapter.start(ctx);
HornetQActivationSpec spec = new HornetQActivationSpec();
@@ -237,6 +245,7 @@
public void testDurableSubscription() throws Exception
{
HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter();
+ qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY);
MyBootstrapContext ctx = new MyBootstrapContext();
qResourceAdapter.start(ctx);
HornetQActivationSpec spec = new HornetQActivationSpec();
@@ -293,6 +302,7 @@
public void testNonDurableSubscription() throws Exception
{
HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter();
+ qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY);
MyBootstrapContext ctx = new MyBootstrapContext();
qResourceAdapter.start(ctx);
HornetQActivationSpec spec = new HornetQActivationSpec();
@@ -340,6 +350,7 @@
public void testSelectorChangedWithTopic() throws Exception
{
HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter();
+ qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY);
MyBootstrapContext ctx = new MyBootstrapContext();
qResourceAdapter.start(ctx);
HornetQActivationSpec spec = new HornetQActivationSpec();
@@ -396,6 +407,7 @@
public void testSelectorNotChangedWithTopic() throws Exception
{
HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter();
+ qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY);
MyBootstrapContext ctx = new MyBootstrapContext();
qResourceAdapter.start(ctx);
HornetQActivationSpec spec = new HornetQActivationSpec();
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerXATest.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerXATest.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerXATest.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -18,6 +18,7 @@
import org.hornetq.core.transaction.impl.XidImpl;
import org.hornetq.ra.HornetQResourceAdapter;
import org.hornetq.ra.inflow.HornetQActivationSpec;
+import org.hornetq.tests.util.UnitTestCase;
import org.hornetq.utils.UUIDGenerator;
import javax.resource.ResourceException;
@@ -43,6 +44,7 @@
public void testXACommit() throws Exception
{
HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter();
+ qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY);
MyBootstrapContext ctx = new MyBootstrapContext();
qResourceAdapter.start(ctx);
HornetQActivationSpec spec = new HornetQActivationSpec();
@@ -74,6 +76,7 @@
public void testXARollback() throws Exception
{
HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter();
+ qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY);
MyBootstrapContext ctx = new MyBootstrapContext();
qResourceAdapter.start(ctx);
HornetQActivationSpec spec = new HornetQActivationSpec();
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/ResourceAdapterTest.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/ResourceAdapterTest.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/ra/ResourceAdapterTest.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -16,6 +16,7 @@
import org.hornetq.ra.HornetQResourceAdapter;
import org.hornetq.ra.inflow.HornetQActivation;
import org.hornetq.ra.inflow.HornetQActivationSpec;
+import org.hornetq.tests.util.UnitTestCase;
import javax.resource.ResourceException;
import javax.resource.spi.endpoint.MessageEndpoint;
@@ -31,6 +32,7 @@
public void testStartStop() throws Exception
{
HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter();
+ qResourceAdapter.setConnectorClassName(UnitTestCase.INVM_CONNECTOR_FACTORY);
HornetQRATestBase.MyBootstrapContext ctx = new HornetQRATestBase.MyBootstrapContext();
qResourceAdapter.setTransactionManagerLocatorClass("");
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/core/config/impl/FileConfigurationTest.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/core/config/impl/FileConfigurationTest.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/core/config/impl/FileConfigurationTest.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -199,6 +199,7 @@
{
Assert.assertEquals("bridge1", bc.getName());
Assert.assertEquals("queue1", bc.getQueueName());
+ Assert.assertEquals(4, bc.getMinLargeMessageSize());
Assert.assertEquals("bridge-forwarding-address1", bc.getForwardingAddress());
Assert.assertEquals("sku > 1", bc.getFilterString());
Assert.assertEquals("org.foo.BridgeTransformer", bc.getTransformerClassName());
@@ -227,6 +228,7 @@
if (ccc.getName().equals("cluster-connection1"))
{
Assert.assertEquals("cluster-connection1", ccc.getName());
+ Assert.assertEquals(321, ccc.getMinLargeMessageSize());
Assert.assertEquals("queues1", ccc.getAddress());
Assert.assertEquals(3, ccc.getRetryInterval());
Assert.assertEquals(true, ccc.isDuplicateDetection());
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/core/journal/impl/fakes/FakeSequentialFileFactory.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/core/journal/impl/fakes/FakeSequentialFileFactory.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/core/journal/impl/fakes/FakeSequentialFileFactory.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -609,7 +609,7 @@
{
}
- public SequentialFile copy()
+ public SequentialFile cloneFile()
{
return null; // To change body of implemented methods use File | Settings | File Templates.
}
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/core/persistence/impl/BatchIDGeneratorUnitTest.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/core/persistence/impl/BatchIDGeneratorUnitTest.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/core/persistence/impl/BatchIDGeneratorUnitTest.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -127,6 +127,8 @@
Assert.assertEquals("No Ids were generated, so the currentID was supposed to stay the same",
lastId,
batch.getCurrentID());
+
+ journal.stop();
}
Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/util/UnitTestCase.java
===================================================================
--- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/util/UnitTestCase.java 2011-11-07 04:32:51 UTC (rev 11665)
+++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/util/UnitTestCase.java 2011-11-08 00:45:05 UTC (rev 11666)
@@ -111,8 +111,23 @@
// Attributes ----------------------------------------------------
- private static final String testDir = System.getProperty("java.io.tmpdir", "/tmp") + "/hornetq-unit-test";
+ private final String baseDir = System.getProperty("java.io.tmpdir", "/tmp") + File.separator + "hornetq-unit-test";
+ private long timeStart = System.currentTimeMillis();
+
+ public long getTimeStart()
+ {
+ return timeStart;
+ }
+
+ private String testDir = baseDir + File.separator + timeStart;
+
+ public void setTimeStart(long time)
+ {
+ timeStart = time;
+ testDir = baseDir + File.separator + timeStart;
+ }
+
// There is a verification about thread leakages. We only fail a single thread when this happens
private static Set<Thread> alreadyFailedThread = new HashSet<Thread>();
@@ -617,7 +632,7 @@
{
// Need to delete the root
- File file = new File(testDir);
+ File file = new File(baseDir);
deleteDirectory(file);
file.mkdirs();
@@ -627,8 +642,6 @@
recreateDirectory(getLargeMessagesDir(testDir));
recreateDirectory(getClientLargeMessagesDir(testDir));
recreateDirectory(getTemporaryDir(testDir));
-
- System.out.println("deleted " + testDir);
}
/**
@@ -899,7 +912,7 @@
OperationContextImpl.clearContext();
- deleteDirectory(new File(getTestDir()));
+ clearData(getTestDir());
InVMRegistry.instance.clear();
@@ -914,7 +927,10 @@
protected void tearDown() throws Exception
{
cleanupPools();
+
+ deleteDirectory(new File(baseDir));
+
Map<Thread, StackTraceElement[]> threadMap = Thread.getAllStackTraces();
for (Thread thread : threadMap.keySet())
{
@@ -1162,16 +1178,43 @@
protected boolean deleteDirectory(final File directory)
{
+ if (!directory.exists())
+ {
+ return true;
+ }
+ else
if (directory.isDirectory())
{
String[] files = directory.list();
for (int j = 0; j < files.length; j++)
{
- if (!deleteDirectory(new File(directory, files[j])))
+ try
{
- return false;
+
+ File fileTmp = new File(directory, files[j]);
+ if (!deleteDirectory(fileTmp))
+ {
+ // This is because of Windows is dumb on releasing files
+ log.warn("could not delete " + fileTmp);
+ forceGC();
+ if (!deleteDirectory(fileTmp))
+ {
+ log.warn("**************************************************************");
+ log.warn("could not delete " + fileTmp + " even afer a retry on GC");
+ log.warn("**************************************************************");
+ }
+ else
+ {
+ log.info(fileTmp + " was deleted without a problem after a retry on GC");
+ }
+ return false;
+ }
}
+ catch (Throwable e)
+ {
+ e.printStackTrace();
+ }
}
}
12 years, 7 months
- 1
- 0
JBoss hornetq SVN: r11665 - in branches: STOMP11_PERF and 1 other directory.
by do-not-reply＠jboss.org
Author: gaohoward
Date: 2011-11-06 23:32:51 -0500 (Sun, 06 Nov 2011)
New Revision: 11665
Added:
branches/STOMP11_PERF/
Log:
stomp perf work
Property changes on: branches/STOMP11_PERF
___________________________________________________________________
Added: svn:ignore
+ build
eclipse-output
thirdparty
logs
ObjectStore
tmp
data
junit*.properties
target
.metadata
Added: svn:mergeinfo
+ /branches/HORNETQ-720_Replication:10878-11528
/branches/STOMP11:11225-11517
12 years, 7 months
- 1
- 0
JBoss hornetq SVN: r11664 - branches.
by do-not-reply＠jboss.org
Author: gaohoward
Date: 2011-11-06 23:31:21 -0500 (Sun, 06 Nov 2011)
New Revision: 11664
Removed:
branches/STOMP11_PERF/
Log:
obsolete branch
12 years, 7 months
- 1
- 0