You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by an...@apache.org on 2014/11/11 12:00:43 UTC
[13/51] [partial] activemq-6 git commit: ACTIVEMQ6-2 Update to HQ
master
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/ConcurrentCreateDeleteProduceTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/ConcurrentCreateDeleteProduceTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/ConcurrentCreateDeleteProduceTest.java
index c2eed39..5923fc5 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/ConcurrentCreateDeleteProduceTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/ConcurrentCreateDeleteProduceTest.java
@@ -58,10 +58,9 @@ public class ConcurrentCreateDeleteProduceTest extends ServiceTestBase
{
super.setUp();
- Configuration config = createDefaultConfig(false);
-
- config.setJournalSyncNonTransactional(false);
- config.setJournalSyncTransactional(false);
+ Configuration config = createDefaultConfig(false)
+ .setJournalSyncNonTransactional(false)
+ .setJournalSyncTransactional(false);
server =
createServer(true, config,
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/ConsumerCloseTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/ConsumerCloseTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/ConsumerCloseTest.java
index 7975279..f3d10db 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/ConsumerCloseTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/ConsumerCloseTest.java
@@ -303,9 +303,9 @@ public class ConsumerCloseTest extends ServiceTestBase
{
super.setUp();
- Configuration config = createDefaultConfig();
- config.getAcceptorConfigurations().add(new TransportConfiguration(InVMAcceptorFactory.class.getCanonicalName()));
- config.setSecurityEnabled(false);
+ Configuration config = createDefaultConfig()
+ .addAcceptorConfiguration(new TransportConfiguration(InVMAcceptorFactory.class.getCanonicalName()))
+ .setSecurityEnabled(false);
server = addServer(HornetQServers.newHornetQServer(config, false));
server.start();
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/ConsumerStuckTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/ConsumerStuckTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/ConsumerStuckTest.java
new file mode 100644
index 0000000..e9b1a38
--- /dev/null
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/ConsumerStuckTest.java
@@ -0,0 +1,305 @@
+/*
+ * Copyright 2005-2014 Red Hat, Inc.
+ * Red Hat licenses this file to you under the Apache License, version
+ * 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ * implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+package org.hornetq.tests.integration.client;
+
+import org.hornetq.api.core.SimpleString;
+import org.hornetq.api.core.TransportConfiguration;
+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.client.impl.ClientSessionFactoryImpl;
+import org.hornetq.core.protocol.core.impl.RemotingConnectionImpl;
+import org.hornetq.core.remoting.impl.netty.NettyConnection;
+import org.hornetq.core.server.HornetQServer;
+import org.hornetq.tests.util.ServiceTestBase;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a>
+ */
+public class ConsumerStuckTest extends ServiceTestBase
+{
+ private HornetQServer server;
+
+ private final SimpleString QUEUE = new SimpleString("ConsumerTestQueue");
+
+ protected boolean isNetty()
+ {
+ return true;
+ }
+
+ @Before
+ @Override
+ public void setUp() throws Exception
+ {
+ super.setUp();
+
+ server = createServer(false, isNetty());
+
+ server.start();
+ }
+
+ @Test
+ public void testClientStuckTest() throws Exception
+ {
+
+ ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(NETTY_CONNECTOR_FACTORY));
+ locator.setConnectionTTL(1000);
+ locator.setClientFailureCheckPeriod(100);
+ locator.setConsumerWindowSize(10 * 1024 * 1024);
+ ClientSessionFactory sf = locator.createSessionFactory();
+ ((ClientSessionFactoryImpl) sf).stopPingingAfterOne();
+
+ RemotingConnectionImpl remotingConnection = (RemotingConnectionImpl) sf.getConnection();
+ ClientSession session = sf.createSession(false, true, true, true);
+
+ session.createQueue(QUEUE, QUEUE, null, false);
+
+ ClientProducer producer = session.createProducer(QUEUE);
+
+ final int numMessages = 10000;
+
+ for (int i = 0; i < numMessages; i++)
+ {
+ ClientMessage message = createTextMessage(session, "m" + i);
+ producer.send(message);
+ }
+
+
+ final ClientConsumer consumer = session.createConsumer(QUEUE);
+ session.start();
+
+ final NettyConnection nettyConnection = (NettyConnection) remotingConnection.getTransportConnection();
+
+
+ Thread tReceive = new Thread()
+ {
+ public void run()
+ {
+ boolean first = true;
+ try
+ {
+ while (!Thread.interrupted())
+ {
+ ClientMessage received = consumer.receive(500);
+ System.out.println("Received " + received);
+ if (first)
+ {
+ first = false;
+ nettyConnection.getNettyChannel().config().setAutoRead(false);
+ }
+ if (received != null)
+ {
+ received.acknowledge();
+ }
+ }
+ }
+ catch (Throwable e)
+ {
+ Thread.currentThread().interrupt();
+ e.printStackTrace();
+ }
+ }
+ };
+
+ tReceive.start();
+
+ try
+ {
+
+ assertEquals(1, server.getSessions().size());
+
+ System.out.println("sessions = " + server.getSessions().size());
+
+ assertEquals(1, server.getConnectionCount());
+
+ long timeout = System.currentTimeMillis() + 20000;
+
+ while (System.currentTimeMillis() < timeout && server.getSessions().size() != 0)
+ {
+ Thread.sleep(10);
+ }
+
+ System.out.println("Size = " + server.getConnectionCount());
+
+ System.out.println("sessions = " + server.getSessions().size());
+
+
+
+ if (server.getSessions().size() != 0)
+ {
+ System.out.println(threadDump("Thread dump"));
+ fail("The cleanup wasn't able to finish cleaning the session. It's probably stuck, look at the thread dump generated by the test for more information");
+ }
+
+
+ timeout = System.currentTimeMillis() + 20000;
+
+ while (System.currentTimeMillis() < timeout && server.getConnectionCount() != 0)
+ {
+ Thread.sleep(10);
+ }
+
+ assertEquals(0, server.getConnectionCount());
+ }
+ finally
+ {
+ nettyConnection.getNettyChannel().config().setAutoRead(true);
+ tReceive.interrupt();
+ tReceive.join();
+ }
+ }
+
+ @Test
+ public void testClientStuckTestWithDirectDelivery() throws Exception
+ {
+
+ ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(NETTY_CONNECTOR_FACTORY));
+ locator.setConnectionTTL(1000);
+ locator.setClientFailureCheckPeriod(100);
+ locator.setConsumerWindowSize(10 * 1024 * 1024);
+ ClientSessionFactory sf = locator.createSessionFactory();
+ ((ClientSessionFactoryImpl) sf).stopPingingAfterOne();
+
+ RemotingConnectionImpl remotingConnection = (RemotingConnectionImpl) sf.getConnection();
+ ClientSession session = sf.createSession(false, true, true, true);
+
+ session.createQueue(QUEUE, QUEUE, null, false);
+
+
+ final int numMessages = 10000;
+
+ final ClientConsumer consumer = session.createConsumer(QUEUE);
+ session.start();
+
+ final NettyConnection nettyConnection = (NettyConnection) remotingConnection.getTransportConnection();
+
+
+ Thread tReceive = new Thread()
+ {
+ public void run()
+ {
+ boolean first = true;
+ try
+ {
+ while (!Thread.interrupted())
+ {
+ ClientMessage received = consumer.receive(500);
+ System.out.println("Received " + received);
+ if (first)
+ {
+ first = false;
+ nettyConnection.getNettyChannel().config().setAutoRead(false);
+ }
+ if (received != null)
+ {
+ received.acknowledge();
+ }
+ }
+ }
+ catch (Throwable e)
+ {
+ Thread.currentThread().interrupt();
+ e.printStackTrace();
+ }
+ }
+ };
+
+ tReceive.start();
+
+ Thread sender = new Thread()
+ {
+ public void run()
+ {
+ try (
+ ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(NETTY_CONNECTOR_FACTORY));
+ ClientSessionFactory factory = locator.createSessionFactory();
+ ClientSession session = factory.createSession(false, true, true, true);
+ ClientProducer producer = session.createProducer(QUEUE);
+ )
+ {
+ for (int i = 0; i < numMessages; i++)
+ {
+ ClientMessage message = createTextMessage(session, "m" + i);
+ producer.send(message);
+ }
+ }
+ catch (Exception e)
+ {
+ e.printStackTrace();
+ }
+ }
+ };
+
+
+ sender.start();
+
+ try
+ {
+
+ long timeout = System.currentTimeMillis() + 20000;
+
+ while (System.currentTimeMillis() < timeout && server.getSessions().size() != 2)
+ {
+ Thread.sleep(10);
+ }
+
+ assertEquals(2, server.getSessions().size());
+
+ System.out.println("sessions = " + server.getSessions().size());
+
+ assertEquals(2, server.getConnectionCount());
+
+ timeout = System.currentTimeMillis() + 20000;
+
+ while (System.currentTimeMillis() < timeout && server.getSessions().size() != 1)
+ {
+ Thread.sleep(10);
+ }
+
+ System.out.println("Size = " + server.getConnectionCount());
+
+ System.out.println("sessions = " + server.getSessions().size());
+
+
+
+ if (server.getSessions().size() != 1)
+ {
+ System.out.println(threadDump("Thread dump"));
+ fail("The cleanup wasn't able to finish cleaning the session. It's probably stuck, look at the thread dump generated by the test for more information");
+ }
+
+ sender.join();
+
+ timeout = System.currentTimeMillis() + 20000;
+
+ while (System.currentTimeMillis() < timeout && server.getConnectionCount() != 0)
+ {
+ Thread.sleep(10);
+ }
+ assertEquals(0, server.getConnectionCount());
+ }
+ finally
+ {
+ nettyConnection.getNettyChannel().config().setAutoRead(true);
+ tReceive.interrupt();
+ tReceive.join();
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/ConsumerTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/ConsumerTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/ConsumerTest.java
index 07d9586..96fca27 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/ConsumerTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/ConsumerTest.java
@@ -12,6 +12,8 @@
*/
package org.hornetq.tests.integration.client;
+import java.util.Arrays;
+import java.util.Collection;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -38,13 +40,32 @@ import org.hornetq.utils.ConcurrentHashSet;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
/**
* @author <a href="mailto:andy.taylor@jboss.org">Andy Taylor</a>
* @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a>
*/
+
+@RunWith(value = Parameterized.class)
public class ConsumerTest extends ServiceTestBase
{
+ @Parameterized.Parameters(name = "isNetty={0}")
+ public static Collection getParameters()
+ {
+ return Arrays.asList(new Object[][]{
+ {true},
+ {false}
+ });
+ }
+
+ public ConsumerTest(boolean netty)
+ {
+ this.netty = netty;
+ }
+
+ private final boolean netty;
private HornetQServer server;
private final SimpleString QUEUE = new SimpleString("ConsumerTestQueue");
@@ -53,7 +74,7 @@ public class ConsumerTest extends ServiceTestBase
protected boolean isNetty()
{
- return false;
+ return netty;
}
@Before
@@ -114,7 +135,7 @@ public class ConsumerTest extends ServiceTestBase
}
// assert that all the messages are there and none have been acked
Assert.assertEquals(0, ((Queue) server.getPostOffice().getBinding(QUEUE).getBindable()).getDeliveringCount());
- Assert.assertEquals(0, ((Queue) server.getPostOffice().getBinding(QUEUE).getBindable()).getMessageCount());
+ Assert.assertEquals(0, getMessageCount(((Queue) server.getPostOffice().getBinding(QUEUE).getBindable())));
session.close();
}
@@ -149,7 +170,7 @@ public class ConsumerTest extends ServiceTestBase
}
// assert that all the messages are there and none have been acked
Assert.assertEquals(0, ((Queue) server.getPostOffice().getBinding(QUEUE).getBindable()).getDeliveringCount());
- Assert.assertEquals(0, ((Queue) server.getPostOffice().getBinding(QUEUE).getBindable()).getMessageCount());
+ Assert.assertEquals(0, getMessageCount(((Queue) server.getPostOffice().getBinding(QUEUE).getBindable())));
session.close();
}
@@ -188,7 +209,7 @@ public class ConsumerTest extends ServiceTestBase
}
// assert that all the messages are there and none have been acked
Assert.assertEquals(0, ((Queue) server.getPostOffice().getBinding(QUEUE).getBindable()).getDeliveringCount());
- Assert.assertEquals(0, ((Queue) server.getPostOffice().getBinding(QUEUE).getBindable()).getMessageCount());
+ Assert.assertEquals(0, getMessageCount(((Queue) server.getPostOffice().getBinding(QUEUE).getBindable())));
session.close();
}
@@ -227,12 +248,12 @@ public class ConsumerTest extends ServiceTestBase
}
// assert that all the messages are there and none have been acked
Assert.assertEquals(0, ((Queue) server.getPostOffice().getBinding(QUEUE).getBindable()).getDeliveringCount());
- Assert.assertEquals(0, ((Queue) server.getPostOffice().getBinding(QUEUE).getBindable()).getMessageCount());
+ Assert.assertEquals(0, getMessageCount(((Queue) server.getPostOffice().getBinding(QUEUE).getBindable())));
session.close();
Assert.assertEquals(0, ((Queue) server.getPostOffice().getBinding(QUEUE).getBindable()).getDeliveringCount());
- Assert.assertEquals(0, ((Queue) server.getPostOffice().getBinding(QUEUE).getBindable()).getMessageCount());
+ Assert.assertEquals(0, getMessageCount(((Queue) server.getPostOffice().getBinding(QUEUE).getBindable())));
}
@Test
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/CoreClientTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/CoreClientTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/CoreClientTest.java
index 7521ce6..a87f180 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/CoreClientTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/CoreClientTest.java
@@ -63,11 +63,9 @@ public class CoreClientTest extends ServiceTestBase
{
final SimpleString QUEUE = new SimpleString("CoreClientTestQueue");
- Configuration conf = createDefaultConfig();
-
- conf.setSecurityEnabled(false);
-
- conf.getAcceptorConfigurations().add(new TransportConfiguration(acceptorFactoryClassName));
+ Configuration conf = createDefaultConfig()
+ .setSecurityEnabled(false)
+ .addAcceptorConfiguration(new TransportConfiguration(acceptorFactoryClassName));
HornetQServer server = addServer(HornetQServers.newHornetQServer(conf, false));
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/CreateQueueIdempotentTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/CreateQueueIdempotentTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/CreateQueueIdempotentTest.java
index 4f90e5d..ca11aa4 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/CreateQueueIdempotentTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/CreateQueueIdempotentTest.java
@@ -40,9 +40,9 @@ public class CreateQueueIdempotentTest extends ServiceTestBase
{
super.setUp();
- Configuration conf = createDefaultConfig();
- conf.setSecurityEnabled(false);
- conf.getAcceptorConfigurations().add(new TransportConfiguration(INVM_ACCEPTOR_FACTORY));
+ Configuration conf = createDefaultConfig()
+ .setSecurityEnabled(false)
+ .addAcceptorConfiguration(new TransportConfiguration(INVM_ACCEPTOR_FACTORY));
server = addServer(HornetQServers.newHornetQServer(conf, true));
server.start();
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/DeadLetterAddressTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/DeadLetterAddressTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/DeadLetterAddressTest.java
index 8c58227..e88ca83 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/DeadLetterAddressTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/DeadLetterAddressTest.java
@@ -327,12 +327,12 @@ public class DeadLetterAddressTest extends ServiceTestBase
long timeout = System.currentTimeMillis() + 5000;
// DLA transfer is asynchronous fired on the rollback
- while (System.currentTimeMillis() < timeout && ((Queue)server.getPostOffice().getBinding(qName).getBindable()).getMessageCount() != 0)
+ while (System.currentTimeMillis() < timeout && getMessageCount(((Queue)server.getPostOffice().getBinding(qName).getBindable())) != 0)
{
Thread.sleep(1);
}
- Assert.assertEquals(0, ((Queue)server.getPostOffice().getBinding(qName).getBindable()).getMessageCount());
+ Assert.assertEquals(0, getMessageCount(((Queue)server.getPostOffice().getBinding(qName).getBindable())));
ClientMessage m = clientConsumer.receiveImmediate();
Assert.assertNull(m);
// All the messages should now be in the DLQ
@@ -511,11 +511,11 @@ public class DeadLetterAddressTest extends ServiceTestBase
public void setUp() throws Exception
{
super.setUp();
-
- Configuration configuration = createDefaultConfig();
- configuration.setSecurityEnabled(false);
TransportConfiguration transportConfig = new TransportConfiguration(UnitTestCase.INVM_ACCEPTOR_FACTORY);
- configuration.getAcceptorConfigurations().add(transportConfig);
+
+ Configuration configuration = createDefaultConfig()
+ .setSecurityEnabled(false)
+ .addAcceptorConfiguration(transportConfig);
server = addServer(HornetQServers.newHornetQServer(configuration, false));
// start the server
server.start();
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/ExpiryAddressTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/ExpiryAddressTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/ExpiryAddressTest.java
index cc7a24a..edddb8e 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/ExpiryAddressTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/ExpiryAddressTest.java
@@ -11,6 +11,7 @@
* permissions and limitations under the License.
*/
package org.hornetq.tests.integration.client;
+import org.hornetq.core.message.impl.MessageImpl;
import org.junit.Before;
import org.junit.Test;
@@ -52,15 +53,16 @@ public class ExpiryAddressTest extends ServiceTestBase
public void testBasicSend() throws Exception
{
SimpleString ea = new SimpleString("EA");
+ SimpleString adSend = new SimpleString("a1");
SimpleString qName = new SimpleString("q1");
SimpleString eq = new SimpleString("EA1");
AddressSettings addressSettings = new AddressSettings();
addressSettings.setExpiryAddress(ea);
- server.getAddressSettingsRepository().addMatch(qName.toString(), addressSettings);
+ server.getAddressSettingsRepository().addMatch("#", addressSettings);
clientSession.createQueue(ea, eq, null, false);
- clientSession.createQueue(qName, qName, null, false);
+ clientSession.createQueue(adSend, qName, null, false);
- ClientProducer producer = clientSession.createProducer(qName);
+ ClientProducer producer = clientSession.createProducer(adSend);
ClientMessage clientMessage = createTextMessage(clientSession, "heyho!");
clientMessage.setExpiration(System.currentTimeMillis());
producer.send(clientMessage);
@@ -75,6 +77,9 @@ public class ExpiryAddressTest extends ServiceTestBase
clientConsumer = clientSession.createConsumer(eq);
m = clientConsumer.receive(500);
Assert.assertNotNull(m);
+ Assert.assertEquals(qName.toString(), m.getStringProperty(MessageImpl.HDR_ORIGINAL_QUEUE));
+ Assert.assertEquals(adSend.toString(), m.getStringProperty(MessageImpl.HDR_ORIGINAL_ADDRESS));
+ Assert.assertNotNull(m);
Assert.assertEquals(m.getBodyBuffer().readString(), "heyho!");
m.acknowledge();
}
@@ -165,6 +170,9 @@ public class ExpiryAddressTest extends ServiceTestBase
Assert.assertNotNull(m);
+ assertNotNull(m.getStringProperty(MessageImpl.HDR_ORIGINAL_ADDRESS));
+ assertNotNull(m.getStringProperty(MessageImpl.HDR_ORIGINAL_QUEUE));
+
ExpiryAddressTest.log.info("acking");
m.acknowledge();
@@ -178,6 +186,9 @@ public class ExpiryAddressTest extends ServiceTestBase
Assert.assertNotNull(m);
+ assertNotNull(m.getStringProperty(MessageImpl.HDR_ORIGINAL_ADDRESS));
+ assertNotNull(m.getStringProperty(MessageImpl.HDR_ORIGINAL_QUEUE));
+
ExpiryAddressTest.log.info("acking");
m.acknowledge();
@@ -206,6 +217,7 @@ public class ExpiryAddressTest extends ServiceTestBase
ClientConsumer clientConsumer = clientSession.createConsumer(qName);
ClientMessage m = clientConsumer.receiveImmediate();
Assert.assertNull(m);
+
clientConsumer.close();
}
@@ -375,11 +387,11 @@ public class ExpiryAddressTest extends ServiceTestBase
public void setUp() throws Exception
{
super.setUp();
-
- Configuration configuration = createDefaultConfig();
- configuration.setSecurityEnabled(false);
TransportConfiguration transportConfig = new TransportConfiguration(UnitTestCase.INVM_ACCEPTOR_FACTORY);
- configuration.getAcceptorConfigurations().add(transportConfig);
+
+ Configuration configuration = createDefaultConfig()
+ .setSecurityEnabled(false)
+ .addAcceptorConfiguration(transportConfig);
server = addServer(HornetQServers.newHornetQServer(configuration, false));
// start the server
server.start();
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/ExpiryLargeMessageTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/ExpiryLargeMessageTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/ExpiryLargeMessageTest.java
index 62d2b9a..0044d9d 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/ExpiryLargeMessageTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/ExpiryLargeMessageTest.java
@@ -143,14 +143,14 @@ public class ExpiryLargeMessageTest extends ServiceTestBase
Thread.sleep(1500);
long timeout = System.currentTimeMillis() + 5000;
- while (timeout > System.currentTimeMillis() && queueExpiry.getMessageCount() != numberOfMessages)
+ while (timeout > System.currentTimeMillis() && getMessageCount(queueExpiry) != numberOfMessages)
{
// What the Expiry Scan would be doing
myQueue.expireReferences();
Thread.sleep(50);
}
- assertEquals(50, queueExpiry.getMessageCount());
+ assertEquals(50, getMessageCount(queueExpiry));
session = sf.createSession(false, false);
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/FailureDeadlockTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/FailureDeadlockTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/FailureDeadlockTest.java
index 7c44021..8435a5e 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/FailureDeadlockTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/FailureDeadlockTest.java
@@ -62,9 +62,9 @@ public class FailureDeadlockTest extends ServiceTestBase
{
super.setUp();
- Configuration conf = createDefaultConfig();
- conf.setSecurityEnabled(false);
- conf.getAcceptorConfigurations().add(new TransportConfiguration(INVM_ACCEPTOR_FACTORY));
+ Configuration conf = createDefaultConfig()
+ .setSecurityEnabled(false)
+ .addAcceptorConfiguration(new TransportConfiguration(INVM_ACCEPTOR_FACTORY));
server = createServer(false, conf);
jmsServer = new JMSServerManagerImpl(server);
jmsServer.setContext(new NullInitialContext());
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/HangConsumerTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/HangConsumerTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/HangConsumerTest.java
index 0f0ae8f..cfe619f 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/HangConsumerTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/HangConsumerTest.java
@@ -12,9 +12,9 @@
*/
package org.hornetq.tests.integration.client;
import org.hornetq.core.server.MessageReference;
+import org.hornetq.core.server.ServerConsumer;
import org.junit.Before;
import org.junit.After;
-
import org.junit.Test;
import java.lang.management.ManagementFactory;
@@ -29,7 +29,6 @@ import java.util.concurrent.TimeUnit;
import javax.management.MBeanServer;
import org.junit.Assert;
-
import org.hornetq.api.core.HornetQException;
import org.hornetq.api.core.Interceptor;
import org.hornetq.api.core.SimpleString;
@@ -56,6 +55,7 @@ import org.hornetq.core.protocol.core.impl.wireformat.SessionReceiveMessage;
import org.hornetq.core.server.HornetQServer;
import org.hornetq.core.server.Queue;
import org.hornetq.core.server.ServerMessage;
+import org.hornetq.core.server.ServerSessionFactory;
import org.hornetq.core.server.impl.HornetQServerImpl;
import org.hornetq.core.server.impl.QueueFactoryImpl;
import org.hornetq.core.server.impl.QueueImpl;
@@ -93,9 +93,8 @@ public class HangConsumerTest extends ServiceTestBase
{
super.setUp();
- Configuration config = createDefaultConfig(false);
-
- config.setMessageExpiryScanPeriod(10);
+ Configuration config = createDefaultConfig(false)
+ .setMessageExpiryScanPeriod(10);
HornetQSecurityManager securityManager = new HornetQSecurityManagerImpl();
@@ -148,9 +147,9 @@ public class HangConsumerTest extends ServiceTestBase
producer.send(sessionProducer.createMessage(true));
sessionProducer.commit();
- // These two operations should finish without the test hanging
- queue.getMessagesAdded(1);
- queue.getMessageCount(1);
+ // These three operations should finish without the test hanging
+ getMessagesAdded(queue);
+ getMessageCount(queue);
releaseConsumers();
@@ -161,8 +160,8 @@ public class HangConsumerTest extends ServiceTestBase
// a flush to guarantee any pending task is finished on flushing out delivery and pending msgs
queue.flushExecutor();
- Assert.assertEquals(2, queue.getMessageCount());
- Assert.assertEquals(2, queue.getMessagesAdded());
+ Assert.assertEquals(2, getMessageCount(queue));
+ Assert.assertEquals(2, getMessagesAdded(queue));
ClientMessage msg = consumer.receive(5000);
Assert.assertNotNull(msg);
@@ -396,8 +395,8 @@ public class HangConsumerTest extends ServiceTestBase
producer.send(session.createMessage(true));
session.commit();
- long queueID = server.getStorageManager().generateUniqueID();
- long txID = server.getStorageManager().generateUniqueID();
+ long queueID = server.getStorageManager().generateID();
+ long txID = server.getStorageManager().generateID();
// Forcing a situation where the server would unexpectedly create a duplicated queue. The server should still start normally
@@ -523,6 +522,12 @@ public class HangConsumerTest extends ServiceTestBase
class MyCallback implements SessionCallback
{
+ @Override
+ public boolean hasCredits(ServerConsumer consumerID)
+ {
+ return true;
+ }
+
final SessionCallback targetCallback;
MyCallback(SessionCallback parameter)
@@ -548,7 +553,7 @@ public class HangConsumerTest extends ServiceTestBase
* @see org.hornetq.spi.core.protocol.SessionCallback#sendMessage(org.hornetq.core.server.ServerMessage, long, int)
*/
@Override
- public int sendMessage(ServerMessage message, long consumerID, int deliveryCount)
+ public int sendMessage(ServerMessage message, ServerConsumer consumer, int deliveryCount)
{
inCall.countDown();
try
@@ -563,7 +568,7 @@ public class HangConsumerTest extends ServiceTestBase
try
{
- return targetCallback.sendMessage(message, consumerID, deliveryCount);
+ return targetCallback.sendMessage(message, consumer, deliveryCount);
}
finally
{
@@ -576,18 +581,18 @@ public class HangConsumerTest extends ServiceTestBase
* @see org.hornetq.spi.core.protocol.SessionCallback#sendLargeMessage(org.hornetq.core.server.ServerMessage, long, long, int)
*/
@Override
- public int sendLargeMessage(ServerMessage message, long consumerID, long bodySize, int deliveryCount)
+ public int sendLargeMessage(ServerMessage message, ServerConsumer consumer, long bodySize, int deliveryCount)
{
- return targetCallback.sendLargeMessage(message, consumerID, bodySize, deliveryCount);
+ return targetCallback.sendLargeMessage(message, consumer, bodySize, deliveryCount);
}
/* (non-Javadoc)
* @see org.hornetq.spi.core.protocol.SessionCallback#sendLargeMessageContinuation(long, byte[], boolean, boolean)
*/
@Override
- public int sendLargeMessageContinuation(long consumerID, byte[] body, boolean continues, boolean requiresResponse)
+ public int sendLargeMessageContinuation(ServerConsumer consumer, byte[] body, boolean continues, boolean requiresResponse)
{
- return targetCallback.sendLargeMessageContinuation(consumerID, body, continues, requiresResponse);
+ return targetCallback.sendLargeMessageContinuation(consumer, body, continues, requiresResponse);
}
/* (non-Javadoc)
@@ -618,7 +623,7 @@ public class HangConsumerTest extends ServiceTestBase
}
@Override
- public void disconnect(long consumerId, String queueName)
+ public void disconnect(ServerConsumer consumerId, String queueName)
{
//To change body of implemented methods use File | Settings | File Templates.
}
@@ -639,7 +644,7 @@ public class HangConsumerTest extends ServiceTestBase
}
@Override
- protected ServerSessionImpl internalCreateSession(String name, String username, String password, int minLargeMessageSize, RemotingConnection connection, boolean autoCommitSends, boolean autoCommitAcks, boolean preAcknowledge, boolean xa, String defaultAddress, SessionCallback callback, OperationContext context) throws Exception
+ protected ServerSessionImpl internalCreateSession(String name, String username, String password, int minLargeMessageSize, RemotingConnection connection, boolean autoCommitSends, boolean autoCommitAcks, boolean preAcknowledge, boolean xa, String defaultAddress, SessionCallback callback, OperationContext context, ServerSessionFactory sessionFactory) throws Exception
{
return new ServerSessionImpl(name,
username,
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/HeuristicXATest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/HeuristicXATest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/HeuristicXATest.java
index b7f94bc..b737f83 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/HeuristicXATest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/HeuristicXATest.java
@@ -67,8 +67,8 @@ public class HeuristicXATest extends ServiceTestBase
@Test
public void testInvalidCall() throws Exception
{
- Configuration configuration = createDefaultConfig();
- configuration.setJMXManagementEnabled(true);
+ Configuration configuration = createDefaultConfig()
+ .setJMXManagementEnabled(true);
HornetQServer server = createServer(false, configuration, mbeanServer, new HashMap<String, AddressSettings>());
server.start();
@@ -92,8 +92,8 @@ public class HeuristicXATest extends ServiceTestBase
private void internalTest(final boolean isCommit) throws Exception
{
- Configuration configuration = createDefaultConfig();
- configuration.setJMXManagementEnabled(true);
+ Configuration configuration = createDefaultConfig()
+ .setJMXManagementEnabled(true);
HornetQServer server = createServer(false, configuration, mbeanServer, new HashMap<String, AddressSettings>());
server.start();
@@ -155,7 +155,7 @@ public class HeuristicXATest extends ServiceTestBase
if (isCommit)
{
- Assert.assertEquals(1, ((Queue)server.getPostOffice().getBinding(ADDRESS).getBindable()).getMessageCount());
+ Assert.assertEquals(1, getMessageCount(((Queue)server.getPostOffice().getBinding(ADDRESS).getBindable())));
session = sf.createSession(false, false, false);
@@ -170,7 +170,7 @@ public class HeuristicXATest extends ServiceTestBase
session.close();
}
- Assert.assertEquals(0, ((Queue)server.getPostOffice().getBinding(ADDRESS).getBindable()).getMessageCount());
+ Assert.assertEquals(0, getMessageCount(((Queue)server.getPostOffice().getBinding(ADDRESS).getBindable())));
}
@Test
@@ -187,8 +187,8 @@ public class HeuristicXATest extends ServiceTestBase
private void doHeuristicCompletionWithRestart(final boolean isCommit) throws Exception
{
- Configuration configuration = createDefaultConfig();
- configuration.setJMXManagementEnabled(true);
+ Configuration configuration = createDefaultConfig()
+ .setJMXManagementEnabled(true);
HornetQServer server = createServer(true, configuration, mbeanServer, new HashMap<String, AddressSettings>());
server.start();
@@ -237,7 +237,7 @@ public class HeuristicXATest extends ServiceTestBase
if (isCommit)
{
- Assert.assertEquals(1, ((Queue)server.getPostOffice().getBinding(ADDRESS).getBindable()).getMessageCount());
+ Assert.assertEquals(1, getMessageCount(((Queue)server.getPostOffice().getBinding(ADDRESS).getBindable())));
session = sf.createSession(false, false, false);
@@ -252,7 +252,7 @@ public class HeuristicXATest extends ServiceTestBase
session.close();
}
- Assert.assertEquals(0, ((Queue)server.getPostOffice().getBinding(ADDRESS).getBindable()).getMessageCount());
+ Assert.assertEquals(0, getMessageCount(((Queue)server.getPostOffice().getBinding(ADDRESS).getBindable())));
server.stop();
@@ -287,8 +287,8 @@ public class HeuristicXATest extends ServiceTestBase
private void doRecoverHeuristicCompletedTxWithRestart(final boolean heuristicCommit) throws Exception
{
- Configuration configuration = createDefaultConfig();
- configuration.setJMXManagementEnabled(true);
+ Configuration configuration = createDefaultConfig()
+ .setJMXManagementEnabled(true);
HornetQServer server = createServer(true, configuration, mbeanServer, new HashMap<String, AddressSettings>());
server.start();
@@ -337,7 +337,7 @@ public class HeuristicXATest extends ServiceTestBase
if (heuristicCommit)
{
- Assert.assertEquals(1, ((Queue)server.getPostOffice().getBinding(ADDRESS).getBindable()).getMessageCount());
+ Assert.assertEquals(1, getMessageCount(((Queue)server.getPostOffice().getBinding(ADDRESS).getBindable())));
session = sf.createSession(false, false, false);
@@ -352,7 +352,7 @@ public class HeuristicXATest extends ServiceTestBase
session.close();
}
- Assert.assertEquals(0, ((Queue)server.getPostOffice().getBinding(ADDRESS).getBindable()).getMessageCount());
+ Assert.assertEquals(0, getMessageCount(((Queue)server.getPostOffice().getBinding(ADDRESS).getBindable())));
server.stop();
@@ -396,8 +396,8 @@ public class HeuristicXATest extends ServiceTestBase
private void doForgetHeuristicCompletedTxAndRestart(final boolean heuristicCommit) throws Exception
{
- Configuration configuration = createDefaultConfig();
- configuration.setJMXManagementEnabled(true);
+ Configuration configuration = createDefaultConfig()
+ .setJMXManagementEnabled(true);
HornetQServer server = createServer(true, configuration, mbeanServer, new HashMap<String, AddressSettings>());
server.start();
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/HornetQCrashTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/HornetQCrashTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/HornetQCrashTest.java
index ace6eba..55dc38e 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/HornetQCrashTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/HornetQCrashTest.java
@@ -56,10 +56,10 @@ public class HornetQCrashTest extends UnitTestCase
@Test
public void testHang() throws Exception
{
- Configuration configuration = createDefaultConfig();
- configuration.setPersistenceEnabled(false);
- configuration.setSecurityEnabled(false);
- configuration.getAcceptorConfigurations().add(new TransportConfiguration(InVMAcceptorFactory.class.getName()));
+ Configuration configuration = createDefaultConfig()
+ .setPersistenceEnabled(false)
+ .setSecurityEnabled(false)
+ .addAcceptorConfiguration(new TransportConfiguration(InVMAcceptorFactory.class.getName()));
server = addServer(HornetQServers.newHornetQServer(configuration));
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/IncompatibleVersionTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/IncompatibleVersionTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/IncompatibleVersionTest.java
index aaee7fb..ec7f135 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/IncompatibleVersionTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/IncompatibleVersionTest.java
@@ -252,9 +252,9 @@ public class IncompatibleVersionTest extends ServiceTestBase
{
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"));
+ Configuration conf = new ConfigurationImpl()
+ .setSecurityEnabled(false)
+ .addAcceptorConfiguration(new TransportConfiguration("org.hornetq.core.remoting.impl.netty.NettyAcceptorFactory"));
HornetQServer server = HornetQServers.newHornetQServer(conf, false);
server.start();
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/InterruptedLargeMessageTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/InterruptedLargeMessageTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/InterruptedLargeMessageTest.java
index ece72e3..5365012 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/InterruptedLargeMessageTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/InterruptedLargeMessageTest.java
@@ -524,7 +524,7 @@ public class InterruptedLargeMessageTest extends LargeMessageTestBase
}
@Override
- protected void postAcknowledge(final MessageReference ref)
+ public void postAcknowledge(final MessageReference ref)
{
System.out.println("Ignoring postACK on message " + ref);
}
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/JournalCrashTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/JournalCrashTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/JournalCrashTest.java
index ecbb925..4ad3f26 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/JournalCrashTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/JournalCrashTest.java
@@ -59,11 +59,11 @@ public class JournalCrashTest extends ServiceTestBase
protected void startServer() throws Exception
{
- Configuration config = createDefaultConfig();
- config.setJournalFileSize(HornetQDefaultConfiguration.getDefaultJournalFileSize());
- config.setJournalCompactMinFiles(HornetQDefaultConfiguration.getDefaultJournalCompactMinFiles());
- config.setJournalCompactPercentage(HornetQDefaultConfiguration.getDefaultJournalCompactPercentage());
- config.setJournalMinFiles(2);
+ Configuration config = createDefaultConfig()
+ .setJournalFileSize(HornetQDefaultConfiguration.getDefaultJournalFileSize())
+ .setJournalCompactMinFiles(HornetQDefaultConfiguration.getDefaultJournalCompactMinFiles())
+ .setJournalCompactPercentage(HornetQDefaultConfiguration.getDefaultJournalCompactPercentage())
+ .setJournalMinFiles(2);
server = super.createServer(true, config);
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/LargeMessageTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/LargeMessageTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/LargeMessageTest.java
index 25f41a0..d3be7b8 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/LargeMessageTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/LargeMessageTest.java
@@ -301,9 +301,8 @@ public class LargeMessageTest extends LargeMessageTestBase
final int PAGE_SIZE = 10 * 1024;
final int MESSAGE_SIZE = 1024; // 1k
- Configuration config = createDefaultConfig();
-
- config.setJournalSyncNonTransactional(false);
+ Configuration config = createDefaultConfig()
+ .setJournalSyncNonTransactional(false);
HornetQServer server = createServer(true,
config,
@@ -380,11 +379,10 @@ public class LargeMessageTest extends LargeMessageTestBase
ClientSession session = null;
- Configuration config = createDefaultConfig(isNetty());
- config.setJournalFileSize(journalsize);
-
- config.setJournalBufferSize_AIO(10 * 1024);
- config.setJournalBufferSize_NIO(10 * 1024);
+ Configuration config = createDefaultConfig(isNetty())
+ .setJournalFileSize(journalsize)
+ .setJournalBufferSize_AIO(10 * 1024)
+ .setJournalBufferSize_NIO(10 * 1024);
HornetQServer server = createServer(true, config);
@@ -2504,7 +2502,7 @@ public class LargeMessageTest extends LargeMessageTestBase
Assert.assertEquals(0,
((Queue) server.getPostOffice().getBinding(ADDRESS).getBindable()).getDeliveringCount());
Assert.assertEquals(0,
- ((Queue) server.getPostOffice().getBinding(ADDRESS).getBindable()).getMessageCount());
+ getMessageCount(((Queue) server.getPostOffice().getBinding(ADDRESS).getBindable())));
}
finally
@@ -2609,7 +2607,7 @@ public class LargeMessageTest extends LargeMessageTestBase
Assert.assertEquals(0,
((Queue) server.getPostOffice().getBinding(ADDRESS).getBindable()).getDeliveringCount());
Assert.assertEquals(0,
- ((Queue) server.getPostOffice().getBinding(ADDRESS).getBindable()).getMessageCount());
+ getMessageCount(((Queue) server.getPostOffice().getBinding(ADDRESS).getBindable())));
}
finally
@@ -2942,7 +2940,7 @@ public class LargeMessageTest extends LargeMessageTestBase
Assert.assertEquals(0,
((Queue) server.getPostOffice().getBinding(ADDRESS).getBindable()).getDeliveringCount());
Assert.assertEquals(0,
- ((Queue) server.getPostOffice().getBinding(ADDRESS).getBindable()).getMessageCount());
+ getMessageCount(((Queue) server.getPostOffice().getBinding(ADDRESS).getBindable())));
}
finally
@@ -3028,7 +3026,7 @@ public class LargeMessageTest extends LargeMessageTestBase
Assert.assertEquals(0,
((Queue) server.getPostOffice().getBinding(ADDRESS).getBindable()).getDeliveringCount());
Assert.assertEquals(0,
- ((Queue) server.getPostOffice().getBinding(ADDRESS).getBindable()).getMessageCount());
+ getMessageCount(((Queue) server.getPostOffice().getBinding(ADDRESS).getBindable())));
log.debug("Thread done");
}
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/MessageExpirationTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/MessageExpirationTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/MessageExpirationTest.java
index 9e567da..2a0ea06 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/MessageExpirationTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/MessageExpirationTest.java
@@ -157,7 +157,7 @@ public class MessageExpirationTest extends ServiceTestBase
Thread.sleep(500);
Assert.assertEquals(0, ((Queue)server.getPostOffice().getBinding(queue).getBindable()).getDeliveringCount());
- Assert.assertEquals(0, ((Queue)server.getPostOffice().getBinding(queue).getBindable()).getMessageCount());
+ Assert.assertEquals(0, getMessageCount(((Queue)server.getPostOffice().getBinding(queue).getBindable())));
ClientMessage message2 = consumer.receiveImmediate();
Assert.assertNull(message2);
@@ -188,7 +188,7 @@ public class MessageExpirationTest extends ServiceTestBase
Assert.assertNull(message2);
Assert.assertEquals(0, ((Queue)server.getPostOffice().getBinding(queue).getBindable()).getDeliveringCount());
- Assert.assertEquals(0, ((Queue)server.getPostOffice().getBinding(queue).getBindable()).getMessageCount());
+ Assert.assertEquals(0, getMessageCount(((Queue)server.getPostOffice().getBinding(queue).getBindable())));
consumer.close();
session.deleteQueue(queue);
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/MessageGroupingConnectionFactoryTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/MessageGroupingConnectionFactoryTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/MessageGroupingConnectionFactoryTest.java
index b7b167d..93ee1b9 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/MessageGroupingConnectionFactoryTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/MessageGroupingConnectionFactoryTest.java
@@ -121,11 +121,11 @@ public class MessageGroupingConnectionFactoryTest extends UnitTestCase
public void setUp() throws Exception
{
super.setUp();
-
- Configuration configuration = createDefaultConfig();
- configuration.setSecurityEnabled(false);
TransportConfiguration transportConfig = new TransportConfiguration(UnitTestCase.INVM_ACCEPTOR_FACTORY);
- configuration.getAcceptorConfigurations().add(transportConfig);
+
+ Configuration configuration = createDefaultConfig()
+ .setSecurityEnabled(false)
+ .addAcceptorConfiguration(transportConfig);
server = addServer(HornetQServers.newHornetQServer(configuration, false));
// start the server
server.start();
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/MessageGroupingTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/MessageGroupingTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/MessageGroupingTest.java
index edcc2f2..eff418a 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/MessageGroupingTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/MessageGroupingTest.java
@@ -610,11 +610,11 @@ public class MessageGroupingTest extends UnitTestCase
public void setUp() throws Exception
{
super.setUp();
-
- Configuration configuration = createDefaultConfig();
- configuration.setSecurityEnabled(false);
TransportConfiguration transportConfig = new TransportConfiguration(UnitTestCase.INVM_ACCEPTOR_FACTORY);
- configuration.getAcceptorConfigurations().add(transportConfig);
+
+ Configuration configuration = createDefaultConfig()
+ .setSecurityEnabled(false)
+ .addAcceptorConfiguration(transportConfig);
server = addServer(HornetQServers.newHornetQServer(configuration, false));
// start the server
server.start();
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/MessagePriorityTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/MessagePriorityTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/MessagePriorityTest.java
index 2ba1786..14fb379 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/MessagePriorityTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/MessagePriorityTest.java
@@ -337,9 +337,9 @@ public class MessagePriorityTest extends UnitTestCase
{
super.setUp();
- Configuration config = createDefaultConfig();
- config.getAcceptorConfigurations().add(new TransportConfiguration(InVMAcceptorFactory.class.getCanonicalName()));
- config.setSecurityEnabled(false);
+ Configuration config = createDefaultConfig()
+ .addAcceptorConfiguration(new TransportConfiguration(InVMAcceptorFactory.class.getCanonicalName()))
+ .setSecurityEnabled(false);
server = addServer(HornetQServers.newHornetQServer(config, false));
server.start();
locator =
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/NIOvsOIOTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/NIOvsOIOTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/NIOvsOIOTest.java
index 3b26f2b..dd752a8 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/NIOvsOIOTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/NIOvsOIOTest.java
@@ -153,9 +153,8 @@ public class NIOvsOIOTest extends UnitTestCase
{
String acceptorFactoryClassName = "org.hornetq.core.remoting.impl.netty.NettyAcceptorFactory";
- Configuration conf = createDefaultConfig();
-
- conf.setSecurityEnabled(false);
+ Configuration conf = createDefaultConfig()
+ .setSecurityEnabled(false);
Map<String, Object> params = new HashMap<String, Object>();
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/NettyConsumerTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/NettyConsumerTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/NettyConsumerTest.java
deleted file mode 100644
index 6f21b7a..0000000
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/NettyConsumerTest.java
+++ /dev/null
@@ -1,29 +0,0 @@
-/*
- * Copyright 2005-2014 Red Hat, Inc.
- * Red Hat licenses this file to you under the Apache License, version
- * 2.0 (the "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- * http://www.apache.org/licenses/LICENSE-2.0
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
- * implied. See the License for the specific language governing
- * permissions and limitations under the License.
- */
-package org.hornetq.tests.integration.client;
-
-/**
- * A NettyConsumerTest
- *
- * @author clebertsuconic
- *
- *
- */
-public class NettyConsumerTest extends ConsumerTest
-{
- protected boolean isNetty()
- {
- return true;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/NewDeadLetterAddressTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/NewDeadLetterAddressTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/NewDeadLetterAddressTest.java
index 67c0c03..99ea6ba 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/NewDeadLetterAddressTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/NewDeadLetterAddressTest.java
@@ -75,11 +75,11 @@ public class NewDeadLetterAddressTest extends UnitTestCase
public void setUp() throws Exception
{
super.setUp();
-
- Configuration configuration = createDefaultConfig();
- configuration.setSecurityEnabled(false);
TransportConfiguration transportConfig = new TransportConfiguration(UnitTestCase.INVM_ACCEPTOR_FACTORY);
- configuration.getAcceptorConfigurations().add(transportConfig);
+
+ Configuration configuration = createDefaultConfig()
+ .setSecurityEnabled(false)
+ .addAcceptorConfiguration(transportConfig);
server = addServer(HornetQServers.newHornetQServer(configuration, false));
// start the server
server.start();
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/OrderTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/OrderTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/OrderTest.java
index fac7099..621073b 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/OrderTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/OrderTest.java
@@ -12,6 +12,9 @@
*/
package org.hornetq.tests.integration.client;
+import java.util.Arrays;
+import java.util.Collection;
+
import org.hornetq.api.core.client.ClientConsumer;
import org.hornetq.api.core.client.ClientMessage;
import org.hornetq.api.core.client.ClientProducer;
@@ -24,19 +27,38 @@ import org.hornetq.tests.util.ServiceTestBase;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
/**
* A OrderTest
*
* @author <mailto:clebert.suconic@jboss.org">Clebert Suconic</a>
*/
+@RunWith(Parameterized.class)
public class OrderTest extends ServiceTestBase
{
+ private boolean persistent;
+
private HornetQServer server;
private ServerLocator locator;
+ public OrderTest(boolean persistent)
+ {
+ this.persistent = persistent;
+ }
+ @Parameterized.Parameters(name = "persistent={0}")
+ public static Collection<Object[]> getParams()
+ {
+ return Arrays.asList(new Object[][]{
+ {true},
+ {false}
+ });
+ }
+
+
@Override
@Before
public void setUp() throws Exception
@@ -53,18 +75,7 @@ public class OrderTest extends ServiceTestBase
// Public --------------------------------------------------------
@Test
- public void testSimpleOrderNoStorage() throws Exception
- {
- doTestSimpleOrder(false);
- }
-
- @Test
- public void testSimpleOrderPersistence() throws Exception
- {
- doTestSimpleOrder(true);
- }
-
- public void doTestSimpleOrder(final boolean persistent) throws Exception
+ public void testSimpleStorage() throws Exception
{
server = createServer(persistent, true);
server.start();
@@ -138,18 +149,7 @@ public class OrderTest extends ServiceTestBase
}
@Test
- public void testOrderOverSessionClosePersistent() throws Exception
- {
- doTestOverCancel(true);
- }
-
- @Test
- public void testOrderOverSessionCloseNonPersistent() throws Exception
- {
- doTestOverCancel(false);
- }
-
- public void doTestOverCancel(final boolean persistent) throws Exception
+ public void testOrderOverSessionClose() throws Exception
{
server = createServer(persistent, true);
@@ -210,19 +210,7 @@ public class OrderTest extends ServiceTestBase
}
@Test
- public void testOrderOverSessionClosePersistentWithRedeliveryDelay() throws Exception
- {
- doTestOverCancelWithRedelivery(true);
- }
-
- @Test
- public void testOrderOverSessionCloseNonPersistentWithRedeliveryDelay() throws Exception
- {
- doTestOverCancelWithRedelivery(false);
- }
-
-
- public void doTestOverCancelWithRedelivery(final boolean persistent) throws Exception
+ public void testOrderOverSessionCloseWithRedeliveryDelay() throws Exception
{
server = createServer(persistent, true);
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/PagingOrderTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/PagingOrderTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/PagingOrderTest.java
index 5d8f379..6da8ca3 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/PagingOrderTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/PagingOrderTest.java
@@ -92,9 +92,8 @@ public class PagingOrderTest extends ServiceTestBase
{
boolean persistentMessages = true;
- Configuration config = createDefaultConfig();
-
- config.setJournalSyncNonTransactional(false);
+ Configuration config = createDefaultConfig()
+ .setJournalSyncNonTransactional(false);
HornetQServer server = createServer(true, config, PAGE_SIZE, PAGE_MAX, new HashMap<String, AddressSettings>());
@@ -202,9 +201,8 @@ public class PagingOrderTest extends ServiceTestBase
{
boolean persistentMessages = true;
- Configuration config = createDefaultConfig();
-
- config.setJournalSyncNonTransactional(false);
+ Configuration config = createDefaultConfig()
+ .setJournalSyncNonTransactional(false);
HornetQServer server = createServer(true, config, PAGE_SIZE, PAGE_MAX, new HashMap<String, AddressSettings>());
@@ -304,10 +302,10 @@ public class PagingOrderTest extends ServiceTestBase
assertEquals(0, errors.get());
- assertEquals(numberOfMessages, q2.getMessageCount());
- assertEquals(numberOfMessages, q2.getMessagesAdded());
- assertEquals(0, q1.getMessageCount());
- assertEquals(numberOfMessages, q1.getMessagesAdded());
+ assertEquals(numberOfMessages, getMessageCount(q2));
+ assertEquals(numberOfMessages, getMessagesAdded(q2));
+ assertEquals(0, getMessageCount(q1));
+ assertEquals(numberOfMessages, getMessagesAdded(q1));
session.close();
sf.close();
@@ -343,11 +341,11 @@ public class PagingOrderTest extends ServiceTestBase
assertNotNull(q2);
- assertEquals("q2 msg count", numberOfMessages, q2.getMessageCount());
- assertEquals("q2 msgs added", numberOfMessages, q2.getMessagesAdded());
- assertEquals("q1 msg count", 0, q1.getMessageCount());
+ assertEquals("q2 msg count", numberOfMessages, getMessageCount(q2));
+ assertEquals("q2 msgs added", numberOfMessages, getMessagesAdded(q2));
+ assertEquals("q1 msg count", 0, getMessageCount(q1));
// 0, since nothing was sent to the queue after the server was restarted
- assertEquals("q1 msgs added", 0, q1.getMessagesAdded());
+ assertEquals("q1 msgs added", 0, getMessagesAdded(q1));
}
@@ -356,9 +354,8 @@ public class PagingOrderTest extends ServiceTestBase
{
boolean persistentMessages = true;
- Configuration config = createDefaultConfig();
-
- config.setJournalSyncNonTransactional(false);
+ Configuration config = createDefaultConfig()
+ .setJournalSyncNonTransactional(false);
HornetQServer server = createServer(true, config, PAGE_SIZE, PAGE_MAX, new HashMap<String, AddressSettings>());
@@ -455,16 +452,15 @@ public class PagingOrderTest extends ServiceTestBase
assertEquals(0, errors.get());
long timeout = System.currentTimeMillis() + 10000;
- while (numberOfMessages - 100 != q1.getMessageCount() && System.currentTimeMillis() < timeout)
+ while (numberOfMessages - 100 != getMessageCount(q1) && System.currentTimeMillis() < timeout)
{
Thread.sleep(500);
}
- assertEquals(numberOfMessages, q2.getMessageCount());
- assertEquals(numberOfMessages, q2.getMessagesAdded());
- assertEquals(numberOfMessages - 100, q1.getMessageCount());
- assertEquals(numberOfMessages, q2.getMessagesAdded());
+ assertEquals(numberOfMessages, getMessageCount(q2));
+ assertEquals(numberOfMessages, getMessagesAdded(q2));
+ assertEquals(numberOfMessages - 100, getMessageCount(q1));
}
@Test
@@ -472,9 +468,8 @@ public class PagingOrderTest extends ServiceTestBase
{
boolean persistentMessages = true;
- Configuration config = createDefaultConfig();
-
- config.setJournalSyncNonTransactional(false);
+ Configuration config = createDefaultConfig()
+ .setJournalSyncNonTransactional(false);
HornetQServer server = createServer(true, config, PAGE_SIZE, PAGE_MAX, new HashMap<String, AddressSettings>());
@@ -573,9 +568,8 @@ public class PagingOrderTest extends ServiceTestBase
{
boolean persistentMessages = true;
- Configuration config = createDefaultConfig();
-
- config.setJournalSyncNonTransactional(false);
+ Configuration config = createDefaultConfig()
+ .setJournalSyncNonTransactional(false);
HornetQServer server = createServer(true, config, PAGE_SIZE, PAGE_MAX, new HashMap<String, AddressSettings>());
@@ -721,9 +715,8 @@ public class PagingOrderTest extends ServiceTestBase
public void testPagingOverCreatedDestinationTopics() throws Exception
{
- Configuration config = createDefaultConfig();
-
- config.setJournalSyncNonTransactional(false);
+ Configuration config = createDefaultConfig()
+ .setJournalSyncNonTransactional(false);
HornetQServer server = createServer(true, config, PAGE_SIZE, -1, new HashMap<String, AddressSettings>());
@@ -748,7 +741,10 @@ public class PagingOrderTest extends ServiceTestBase
1000,
0,
false,
- "PAGE");
+ "PAGE",
+ -1,
+ 10,
+ "KILL");
HornetQJMSConnectionFactory cf = (HornetQJMSConnectionFactory) HornetQJMSClient.createConnectionFactoryWithoutHA(JMSFactoryType.CF,
new TransportConfiguration(INVM_CONNECTOR_FACTORY));
@@ -796,9 +792,8 @@ public class PagingOrderTest extends ServiceTestBase
public void testPagingOverCreatedDestinationQueues() throws Exception
{
- Configuration config = createDefaultConfig();
-
- config.setJournalSyncNonTransactional(false);
+ Configuration config = createDefaultConfig()
+ .setJournalSyncNonTransactional(false);
HornetQServer server = createServer(true, config, -1, -1, AddressFullMessagePolicy.BLOCK, new HashMap<String, AddressSettings>());
@@ -821,7 +816,10 @@ public class PagingOrderTest extends ServiceTestBase
1000,
0,
false,
- "PAGE");
+ "PAGE",
+ -1,
+ 10,
+ "KILL");
jmsServer.createQueue(true, "Q1", null, true, "/queue/Q1");
http://git-wip-us.apache.org/repos/asf/activemq-6/blob/177e6820/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/PagingSyncTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/PagingSyncTest.java b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/PagingSyncTest.java
index a038863..28fb180 100644
--- a/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/PagingSyncTest.java
+++ b/tests/integration-tests/src/test/java/org/hornetq/tests/integration/client/PagingSyncTest.java
@@ -53,9 +53,8 @@ public class PagingSyncTest extends ServiceTestBase
{
boolean persistentMessages = true;
- Configuration config = createDefaultConfig();
-
- config.setJournalSyncNonTransactional(false);
+ Configuration config = createDefaultConfig()
+ .setJournalSyncNonTransactional(false);
HornetQServer server = createServer(true, config, PAGE_SIZE, PAGE_MAX, new HashMap<String, AddressSettings>());