You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ma...@apache.org on 2016/09/30 15:26:13 UTC
[13/52] [partial] activemq-artemis git commit: ARTEMIS-765 Improve
Checkstyle
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ec48f9ed/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsRollbackRedeliveryTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsRollbackRedeliveryTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsRollbackRedeliveryTest.java
index c57845d..23e7781 100644
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsRollbackRedeliveryTest.java
+++ b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsRollbackRedeliveryTest.java
@@ -6,7 +6,7 @@
* (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
+ * 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,
@@ -17,10 +17,6 @@
package org.apache.activemq;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicInteger;
-
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
@@ -29,17 +25,23 @@ import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.activemq.broker.BrokerService;
import org.junit.After;
import org.junit.Before;
-import org.junit.Test;
import org.junit.Rule;
+import org.junit.Test;
import org.junit.rules.TestName;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
public class JmsRollbackRedeliveryTest {
@@ -103,8 +105,7 @@ public class JmsRollbackRedeliveryTest {
if (interleaveProducer) {
populateDestinationWithInterleavedProducer(nbMessages, destinationName, connection);
- }
- else {
+ } else {
populateDestination(nbMessages, destinationName, connection);
}
@@ -123,8 +124,7 @@ public class JmsRollbackRedeliveryTest {
assertTrue(msg.getJMSRedelivered());
assertEquals(2, msg.getLongProperty("JMSXDeliveryCount"));
session.commit();
- }
- else {
+ } else {
LOG.info("Rollback message " + msg.getText() + " id: " + msg.getJMSMessageID());
assertFalse("should not have redelivery flag set, id: " + msg.getJMSMessageID(), msg.getJMSRedelivered());
session.rollback();
@@ -159,8 +159,7 @@ public class JmsRollbackRedeliveryTest {
LOG.info("Received message " + msg.getText() + " (" + received.getAndIncrement() + ")" + msg.getJMSMessageID());
assertTrue(msg.getJMSRedelivered());
session.commit();
- }
- else {
+ } else {
LOG.info("Rollback message " + msg.getText() + " id: " + msg.getJMSMessageID());
session.rollback();
}
@@ -194,8 +193,7 @@ public class JmsRollbackRedeliveryTest {
LOG.info("Received message " + msg.getText() + " (" + received.getAndIncrement() + ")" + msg.getJMSMessageID());
assertTrue(msg.getJMSRedelivered());
session.commit();
- }
- else {
+ } else {
LOG.info("Rollback message " + msg.getText() + " id: " + msg.getJMSMessageID());
session.rollback();
}
@@ -340,8 +338,7 @@ public class JmsRollbackRedeliveryTest {
for (int i = 1; i <= nbMessages; i++) {
if (i % 2 == 0) {
producer1.send(session1.createTextMessage("<hello id='" + i + "'/>"));
- }
- else {
+ } else {
producer2.send(session2.createTextMessage("<hello id='" + i + "'/>"));
}
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ec48f9ed/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsSendReceiveTestSupport.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsSendReceiveTestSupport.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsSendReceiveTestSupport.java
index 6fd36cc..04cbe1c 100644
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsSendReceiveTestSupport.java
+++ b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsSendReceiveTestSupport.java
@@ -6,7 +6,7 @@
* (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
+ * 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,
@@ -16,13 +16,6 @@
*/
package org.apache.activemq;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.Date;
-import java.util.Iterator;
-import java.util.List;
-
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
@@ -32,6 +25,12 @@ import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Date;
+import java.util.Iterator;
+import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -181,8 +180,7 @@ public class JmsSendReceiveTestSupport extends TestSupport implements MessageLis
while (messages.size() < data.length && waitTime >= 0) {
try {
lock.wait(200);
- }
- catch (InterruptedException e) {
+ } catch (InterruptedException e) {
e.printStackTrace();
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ec48f9ed/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsSendReceiveWithMessageExpirationTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsSendReceiveWithMessageExpirationTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsSendReceiveWithMessageExpirationTest.java
index 6e4e214..dffd722 100644
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsSendReceiveWithMessageExpirationTest.java
+++ b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsSendReceiveWithMessageExpirationTest.java
@@ -6,7 +6,7 @@
* (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
+ * 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,
@@ -16,10 +16,6 @@
*/
package org.apache.activemq;
-import java.util.Date;
-import java.util.Vector;
-import java.util.concurrent.TimeUnit;
-
import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
@@ -29,6 +25,9 @@ import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.Topic;
+import java.util.Date;
+import java.util.Vector;
+import java.util.concurrent.TimeUnit;
import org.apache.activemq.broker.BrokerRegistry;
import org.apache.activemq.broker.region.DestinationStatistics;
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ec48f9ed/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsSendWithAsyncCallbackTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsSendWithAsyncCallbackTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsSendWithAsyncCallbackTest.java
index f2717d2..d88fd28 100644
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsSendWithAsyncCallbackTest.java
+++ b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsSendWithAsyncCallbackTest.java
@@ -6,7 +6,7 @@
* (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
+ * 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,
@@ -16,8 +16,6 @@
*/
package org.apache.activemq;
-import java.util.concurrent.CountDownLatch;
-
import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.JMSException;
@@ -26,6 +24,7 @@ import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.Session;
+import java.util.concurrent.CountDownLatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ec48f9ed/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsSessionRecoverTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsSessionRecoverTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsSessionRecoverTest.java
index e083332..04bcc37 100644
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsSessionRecoverTest.java
+++ b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsSessionRecoverTest.java
@@ -6,7 +6,7 @@
* (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
+ * 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,
@@ -16,9 +16,6 @@
*/
package org.apache.activemq;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-
import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
@@ -29,6 +26,8 @@ import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import junit.framework.TestCase;
@@ -202,8 +201,7 @@ public class JmsSessionRecoverTest extends TestCase {
errorMessage[0] = "Got too many messages: " + counter;
doneCountDownLatch.countDown();
}
- }
- catch (Throwable e) {
+ } catch (Throwable e) {
e.printStackTrace();
errorMessage[0] = "Got exception: " + e;
doneCountDownLatch.countDown();
@@ -216,8 +214,7 @@ public class JmsSessionRecoverTest extends TestCase {
if (errorMessage[0] != null) {
fail(errorMessage[0]);
}
- }
- else {
+ } else {
fail("Timeout waiting for async message delivery to complete.");
}
@@ -273,8 +270,7 @@ public class JmsSessionRecoverTest extends TestCase {
errorMessage[0] = "Got too many messages: " + counter;
doneCountDownLatch.countDown();
}
- }
- catch (Throwable e) {
+ } catch (Throwable e) {
e.printStackTrace();
errorMessage[0] = "Got exception: " + e;
doneCountDownLatch.countDown();
@@ -287,8 +283,7 @@ public class JmsSessionRecoverTest extends TestCase {
if (errorMessage[0] != null) {
fail(errorMessage[0]);
}
- }
- else {
+ } else {
fail("Timeout waiting for async message delivery to complete.");
}
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ec48f9ed/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTempDestinationTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTempDestinationTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTempDestinationTest.java
index 866ab63..060853c 100644
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTempDestinationTest.java
+++ b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTempDestinationTest.java
@@ -6,7 +6,7 @@
* (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
+ * 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,
@@ -16,12 +16,6 @@
*/
package org.apache.activemq;
-import java.net.URISyntaxException;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.List;
-
import javax.jms.BytesMessage;
import javax.jms.Connection;
import javax.jms.DeliveryMode;
@@ -34,6 +28,11 @@ import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TemporaryQueue;
import javax.jms.TextMessage;
+import java.net.URISyntaxException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
import org.apache.activemq.util.Wait;
import org.junit.After;
@@ -70,8 +69,7 @@ public class JmsTempDestinationTest {
Connection conn = iter.next();
try {
conn.close();
- }
- catch (Throwable e) {
+ } catch (Throwable e) {
}
iter.remove();
}
@@ -107,8 +105,7 @@ public class JmsTempDestinationTest {
try {
consumer = otherSession.createConsumer(queue);
Assert.fail("Send should fail since temp destination should be used from another connection");
- }
- catch (InvalidDestinationException e) {
+ } catch (InvalidDestinationException e) {
Assert.assertTrue("failed to throw an exception", true);
}
@@ -249,8 +246,7 @@ public class JmsTempDestinationTest {
message = session.createTextMessage("Hello");
producer.send(message);
Assert.fail("Send should fail since temp destination should not exist anymore.");
- }
- catch (JMSException e) {
+ } catch (JMSException e) {
e.printStackTrace();
}
}
@@ -302,8 +298,7 @@ public class JmsTempDestinationTest {
message = session.createTextMessage("Hello");
producer.send(message);
Assert.fail("Send should fail since temp destination should not exist anymore.");
- }
- catch (JMSException e) {
+ } catch (JMSException e) {
Assert.assertTrue("failed to throw an exception", true);
}
}
@@ -329,8 +324,7 @@ public class JmsTempDestinationTest {
try {
queue.delete();
Assert.fail("Should fail as Subscribers are active");
- }
- catch (JMSException e) {
+ } catch (JMSException e) {
Assert.assertTrue("failed to throw an exception", true);
}
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ec48f9ed/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTestSupport.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTestSupport.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTestSupport.java
index 03bf32c..0fca303 100644
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTestSupport.java
+++ b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTestSupport.java
@@ -6,7 +6,7 @@
* (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
+ * 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,
@@ -16,6 +16,13 @@
*/
package org.apache.activemq;
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
import java.io.File;
import java.io.IOException;
import java.net.URI;
@@ -25,14 +32,6 @@ import java.util.Iterator;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
-import javax.jms.Connection;
-import javax.jms.ConnectionFactory;
-import javax.jms.Destination;
-import javax.jms.JMSException;
-import javax.jms.MessageConsumer;
-import javax.jms.MessageProducer;
-import javax.jms.Session;
-
import org.apache.activemq.broker.BrokerFactory;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.command.ActiveMQDestination;
@@ -134,8 +133,7 @@ public class JmsTestSupport extends CombinationTestSupport {
Connection conn = iter.next();
try {
conn.close();
- }
- catch (Throwable e) {
+ } catch (Throwable e) {
}
iter.remove();
}
@@ -146,32 +144,28 @@ public class JmsTestSupport extends CombinationTestSupport {
protected void safeClose(Connection c) {
try {
c.close();
- }
- catch (Throwable e) {
+ } catch (Throwable e) {
}
}
protected void safeClose(Session s) {
try {
s.close();
- }
- catch (Throwable e) {
+ } catch (Throwable e) {
}
}
protected void safeClose(MessageConsumer c) {
try {
c.close();
- }
- catch (Throwable e) {
+ } catch (Throwable e) {
}
}
protected void safeClose(MessageProducer p) {
try {
p.close();
- }
- catch (Throwable e) {
+ } catch (Throwable e) {
}
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ec48f9ed/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicCompositeSendReceiveTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicCompositeSendReceiveTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicCompositeSendReceiveTest.java
index eb9a06d..8b5bc51 100644
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicCompositeSendReceiveTest.java
+++ b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicCompositeSendReceiveTest.java
@@ -6,7 +6,7 @@
* (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
+ * 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,
@@ -47,8 +47,7 @@ public class JmsTopicCompositeSendReceiveTest extends JmsTopicSendReceiveTest {
if (durable) {
LOG.info("Creating durable consumer");
consumer2 = consumeSession.createDurableSubscriber((Topic) consumerDestination2, getName());
- }
- else {
+ } else {
consumer2 = consumeSession.createConsumer(consumerDestination2);
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ec48f9ed/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicRedeliverTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicRedeliverTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicRedeliverTest.java
index 3bb2df9..d862a4a 100644
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicRedeliverTest.java
+++ b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicRedeliverTest.java
@@ -6,7 +6,7 @@
* (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
+ * 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,
@@ -75,8 +75,7 @@ public class JmsTopicRedeliverTest extends TestSupport {
if (topic) {
consumerDestination = session.createTopic(getConsumerSubject());
producerDestination = session.createTopic(getProducerSubject());
- }
- else {
+ } else {
consumerDestination = session.createQueue(getConsumerSubject());
producerDestination = session.createQueue(getProducerSubject());
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ec48f9ed/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicRequestReplyTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicRequestReplyTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicRequestReplyTest.java
index ad94f09..72913cf 100644
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicRequestReplyTest.java
+++ b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicRequestReplyTest.java
@@ -6,7 +6,7 @@
* (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
+ * 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,
@@ -16,9 +16,6 @@
*/
package org.apache.activemq;
-import java.util.List;
-import java.util.Vector;
-
import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
@@ -30,6 +27,8 @@ import javax.jms.Session;
import javax.jms.TemporaryQueue;
import javax.jms.TemporaryTopic;
import javax.jms.TextMessage;
+import java.util.List;
+import java.util.Vector;
import org.apache.activemq.test.TestSupport;
import org.slf4j.Logger;
@@ -91,8 +90,7 @@ public class JmsTopicRequestReplyTest extends TestSupport implements MessageList
LOG.info("Received reply.");
LOG.info(replyMessage.toString());
assertEquals("Wrong message content", "Hello: Olivier", replyMessage.getText());
- }
- else {
+ } else {
fail("Should have received a reply by now");
}
replyConsumer.close();
@@ -133,15 +131,13 @@ public class JmsTopicRequestReplyTest extends TestSupport implements MessageList
if (dynamicallyCreateProducer) {
replyProducer = serverSession.createProducer(replyDestination);
replyProducer.send(replyMessage);
- }
- else {
+ } else {
replyProducer.send(replyDestination, replyMessage);
}
LOG.info("Sent reply.");
LOG.info(replyMessage.toString());
- }
- catch (JMSException e) {
+ } catch (JMSException e) {
onException(e);
}
}
@@ -154,12 +150,10 @@ public class JmsTopicRequestReplyTest extends TestSupport implements MessageList
Message message = requestConsumer.receive(5000);
if (message != null) {
onMessage(message);
- }
- else {
+ } else {
LOG.error("No message received");
}
- }
- catch (JMSException e) {
+ } catch (JMSException e) {
onException(e);
}
}
@@ -180,8 +174,7 @@ public class JmsTopicRequestReplyTest extends TestSupport implements MessageList
final MessageConsumer requestConsumer = serverSession.createConsumer(requestDestination);
if (useAsyncConsume) {
requestConsumer.setMessageListener(this);
- }
- else {
+ } else {
Thread thread = new Thread(new Runnable() {
@Override
public void run() {
@@ -225,8 +218,7 @@ public class JmsTopicRequestReplyTest extends TestSupport implements MessageList
protected void deleteTemporaryDestination(Destination dest) throws JMSException {
if (topic) {
((TemporaryTopic) dest).delete();
- }
- else {
+ } else {
((TemporaryQueue) dest).delete();
}
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ec48f9ed/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSelectorTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSelectorTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSelectorTest.java
index 8968d31..97dd1fc 100644
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSelectorTest.java
+++ b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSelectorTest.java
@@ -6,7 +6,7 @@
* (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
+ * 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,
@@ -67,8 +67,7 @@ public class JmsTopicSelectorTest extends TestSupport {
if (topic) {
consumerDestination = session.createTopic(getConsumerSubject());
producerDestination = session.createTopic(getProducerSubject());
- }
- else {
+ } else {
consumerDestination = session.createQueue(getConsumerSubject());
producerDestination = session.createQueue(getProducerSubject());
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ec48f9ed/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSendReceiveSubscriberTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSendReceiveSubscriberTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSendReceiveSubscriberTest.java
index 3f120a6..c96baa8 100644
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSendReceiveSubscriberTest.java
+++ b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSendReceiveSubscriberTest.java
@@ -6,7 +6,7 @@
* (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
+ * 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,
@@ -30,8 +30,7 @@ public class JmsTopicSendReceiveSubscriberTest extends JmsTopicSendReceiveTest {
protected MessageConsumer createConsumer() throws JMSException {
if (durable) {
return super.createConsumer();
- }
- else {
+ } else {
TopicSession topicSession = (TopicSession) session;
return topicSession.createSubscriber((Topic) consumerDestination, null, false);
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ec48f9ed/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSendReceiveTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSendReceiveTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSendReceiveTest.java
index 71ddcd5..c34b492 100644
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSendReceiveTest.java
+++ b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSendReceiveTest.java
@@ -6,7 +6,7 @@
* (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
+ * 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,
@@ -58,8 +58,7 @@ public class JmsTopicSendReceiveTest extends JmsSendReceiveTestSupport {
if (topic) {
consumerDestination = session.createTopic(getConsumerSubject());
producerDestination = session.createTopic(getProducerSubject());
- }
- else {
+ } else {
consumerDestination = session.createQueue(getConsumerSubject());
producerDestination = session.createQueue(getProducerSubject());
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ec48f9ed/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSendReceiveWithTwoConnectionsTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSendReceiveWithTwoConnectionsTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSendReceiveWithTwoConnectionsTest.java
index 216ed10..7bcd850 100644
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSendReceiveWithTwoConnectionsTest.java
+++ b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSendReceiveWithTwoConnectionsTest.java
@@ -6,7 +6,7 @@
* (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
+ * 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,
@@ -16,8 +16,6 @@
*/
package org.apache.activemq;
-import org.apache.activemq.transport.tcp.TcpTransportFactory;
-
import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
@@ -25,6 +23,8 @@ import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;
+import org.apache.activemq.transport.tcp.TcpTransportFactory;
+
/**
* @version
*/
@@ -65,8 +65,7 @@ public class JmsTopicSendReceiveWithTwoConnectionsTest extends JmsSendReceiveTes
if (topic) {
consumerDestination = session.createTopic(getConsumerSubject());
producerDestination = session.createTopic(getProducerSubject());
- }
- else {
+ } else {
consumerDestination = session.createQueue(getConsumerSubject());
producerDestination = session.createQueue(getProducerSubject());
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ec48f9ed/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSendReceiveWithTwoConnectionsWithJMXTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSendReceiveWithTwoConnectionsWithJMXTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSendReceiveWithTwoConnectionsWithJMXTest.java
index 5acaece..f622086 100644
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSendReceiveWithTwoConnectionsWithJMXTest.java
+++ b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSendReceiveWithTwoConnectionsWithJMXTest.java
@@ -6,7 +6,7 @@
* (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
+ * 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,
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ec48f9ed/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSendSameMessageTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSendSameMessageTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSendSameMessageTest.java
index 7b84a15..9929774 100644
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSendSameMessageTest.java
+++ b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicSendSameMessageTest.java
@@ -6,7 +6,7 @@
* (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
+ * 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,
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ec48f9ed/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicTransactionTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicTransactionTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicTransactionTest.java
index b9ec483..0fe34d1 100644
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicTransactionTest.java
+++ b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicTransactionTest.java
@@ -6,7 +6,7 @@
* (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
+ * 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,
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ec48f9ed/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicWildcardSendReceiveTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicWildcardSendReceiveTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicWildcardSendReceiveTest.java
index 8fd1c00..9a647b4 100644
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicWildcardSendReceiveTest.java
+++ b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTopicWildcardSendReceiveTest.java
@@ -6,7 +6,7 @@
* (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
+ * 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,
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ec48f9ed/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTransactionTestSupport.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTransactionTestSupport.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTransactionTestSupport.java
index 0171bfa..16a85eb 100755
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTransactionTestSupport.java
+++ b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/JmsTransactionTestSupport.java
@@ -6,7 +6,7 @@
* (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
+ * 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,
@@ -44,678 +44,677 @@ import org.slf4j.LoggerFactory;
*/
public abstract class JmsTransactionTestSupport extends TestSupport implements MessageListener {
- private static final Logger LOG = LoggerFactory.getLogger(JmsTransactionTestSupport.class);
- private static final int MESSAGE_COUNT = 5;
- private static final String MESSAGE_TEXT = "message";
-
- protected ConnectionFactory connectionFactory;
- protected Connection connection;
- protected Session session;
- protected MessageConsumer consumer;
- protected MessageProducer producer;
- protected JmsResourceProvider resourceProvider;
- protected Destination destination;
- protected int batchCount = 10;
- protected int batchSize = 20;
- protected BrokerService broker;
-
- // for message listener test
- private final List<Message> unackMessages = new ArrayList<>(MESSAGE_COUNT);
- private final List<Message> ackMessages = new ArrayList<>(MESSAGE_COUNT);
- private boolean resendPhase;
-
- public JmsTransactionTestSupport() {
- super();
- }
-
- public JmsTransactionTestSupport(String name) {
- super(name);
- }
-
- /*
- * (non-Javadoc)
- *
- * @see junit.framework.TestCase#setUp()
- */
- @Override
- protected void setUp() throws Exception {
- broker = createBroker();
- broker.start();
- broker.waitUntilStarted();
-
- resourceProvider = getJmsResourceProvider();
- topic = resourceProvider.isTopic();
- // We will be using transacted sessions.
- setSessionTransacted();
- connectionFactory = newConnectionFactory();
- reconnect();
- }
-
- protected void setSessionTransacted() {
- resourceProvider.setTransacted(true);
- }
-
- protected ConnectionFactory newConnectionFactory() throws Exception {
- return resourceProvider.createConnectionFactory();
- }
-
- protected void beginTx() throws Exception {
- //no-op for local tx
- }
-
- protected void commitTx() throws Exception {
- session.commit();
- }
-
- protected void rollbackTx() throws Exception {
- session.rollback();
- }
-
- /**
- */
- protected BrokerService createBroker() throws Exception, URISyntaxException {
- return BrokerFactory.createBroker(new URI("broker://()/localhost?persistent=false"));
- }
-
- /*
- * (non-Javadoc)
- *
- * @see junit.framework.TestCase#tearDown()
- */
- @Override
- protected void tearDown() throws Exception {
- LOG.info("Closing down connection");
-
- try {
- session.close();
- session = null;
- connection.close();
- connection = null;
- } catch (Exception e) {
- LOG.info("Caught exception while closing resources.");
- }
-
- try {
- broker.stop();
- broker.waitUntilStopped();
- broker = null;
- } catch (Exception e) {
- LOG.info("Caught exception while shutting down the Broker", e);
- }
-
- LOG.info("Connection closed.");
- }
-
- protected abstract JmsResourceProvider getJmsResourceProvider();
-
- /**
- * Sends a batch of messages and validates that the messages are received.
- *
- * @throws Exception
- */
- public void testSendReceiveTransactedBatches() throws Exception {
-
- TextMessage message = session.createTextMessage("Batch Message");
- for (int j = 0; j < batchCount; j++) {
- LOG.info("Producing bacth " + j + " of " + batchSize + " messages");
-
- beginTx();
- for (int i = 0; i < batchSize; i++) {
- producer.send(message);
+ private static final Logger LOG = LoggerFactory.getLogger(JmsTransactionTestSupport.class);
+ private static final int MESSAGE_COUNT = 5;
+ private static final String MESSAGE_TEXT = "message";
+
+ protected ConnectionFactory connectionFactory;
+ protected Connection connection;
+ protected Session session;
+ protected MessageConsumer consumer;
+ protected MessageProducer producer;
+ protected JmsResourceProvider resourceProvider;
+ protected Destination destination;
+ protected int batchCount = 10;
+ protected int batchSize = 20;
+ protected BrokerService broker;
+
+ // for message listener test
+ private final List<Message> unackMessages = new ArrayList<>(MESSAGE_COUNT);
+ private final List<Message> ackMessages = new ArrayList<>(MESSAGE_COUNT);
+ private boolean resendPhase;
+
+ public JmsTransactionTestSupport() {
+ super();
+ }
+
+ public JmsTransactionTestSupport(String name) {
+ super(name);
+ }
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see junit.framework.TestCase#setUp()
+ */
+ @Override
+ protected void setUp() throws Exception {
+ broker = createBroker();
+ broker.start();
+ broker.waitUntilStarted();
+
+ resourceProvider = getJmsResourceProvider();
+ topic = resourceProvider.isTopic();
+ // We will be using transacted sessions.
+ setSessionTransacted();
+ connectionFactory = newConnectionFactory();
+ reconnect();
+ }
+
+ protected void setSessionTransacted() {
+ resourceProvider.setTransacted(true);
+ }
+
+ protected ConnectionFactory newConnectionFactory() throws Exception {
+ return resourceProvider.createConnectionFactory();
+ }
+
+ protected void beginTx() throws Exception {
+ //no-op for local tx
+ }
+
+ protected void commitTx() throws Exception {
+ session.commit();
+ }
+
+ protected void rollbackTx() throws Exception {
+ session.rollback();
+ }
+
+ /**
+ */
+ protected BrokerService createBroker() throws Exception, URISyntaxException {
+ return BrokerFactory.createBroker(new URI("broker://()/localhost?persistent=false"));
+ }
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see junit.framework.TestCase#tearDown()
+ */
+ @Override
+ protected void tearDown() throws Exception {
+ LOG.info("Closing down connection");
+
+ try {
+ session.close();
+ session = null;
+ connection.close();
+ connection = null;
+ } catch (Exception e) {
+ LOG.info("Caught exception while closing resources.");
+ }
+
+ try {
+ broker.stop();
+ broker.waitUntilStopped();
+ broker = null;
+ } catch (Exception e) {
+ LOG.info("Caught exception while shutting down the Broker", e);
+ }
+
+ LOG.info("Connection closed.");
+ }
+
+ protected abstract JmsResourceProvider getJmsResourceProvider();
+
+ /**
+ * Sends a batch of messages and validates that the messages are received.
+ *
+ * @throws Exception
+ */
+ public void testSendReceiveTransactedBatches() throws Exception {
+
+ TextMessage message = session.createTextMessage("Batch Message");
+ for (int j = 0; j < batchCount; j++) {
+ LOG.info("Producing bacth " + j + " of " + batchSize + " messages");
+
+ beginTx();
+ for (int i = 0; i < batchSize; i++) {
+ producer.send(message);
+ }
+ messageSent();
+ commitTx();
+ LOG.info("Consuming bacth " + j + " of " + batchSize + " messages");
+
+ beginTx();
+ for (int i = 0; i < batchSize; i++) {
+ message = (TextMessage) consumer.receive(1000 * 5);
+ assertNotNull("Received only " + i + " messages in batch " + j, message);
+ assertEquals("Batch Message", message.getText());
+ }
+
+ commitTx();
+ }
+ }
+
+ protected void messageSent() throws Exception {
+ }
+
+ /**
+ * Sends a batch of messages and validates that the rollbacked message was
+ * not consumed.
+ *
+ * @throws Exception
+ */
+ public void testSendRollback() throws Exception {
+ Message[] outbound = new Message[]{session.createTextMessage("First Message"), session.createTextMessage("Second Message")};
+
+ // sends a message
+ beginTx();
+ producer.send(outbound[0]);
+ commitTx();
+
+ // sends a message that gets rollbacked
+ beginTx();
+ producer.send(session.createTextMessage("I'm going to get rolled back."));
+ rollbackTx();
+
+ // sends a message
+ beginTx();
+ producer.send(outbound[1]);
+ commitTx();
+
+ // receives the first message
+ beginTx();
+ ArrayList<Message> messages = new ArrayList<>();
+ LOG.info("About to consume message 1");
+ Message message = consumer.receive(1000);
+ messages.add(message);
+ LOG.info("Received: " + message);
+
+ // receives the second message
+ LOG.info("About to consume message 2");
+ message = consumer.receive(4000);
+ messages.add(message);
+ LOG.info("Received: " + message);
+
+ // validates that the rollbacked was not consumed
+ commitTx();
+ Message inbound[] = new Message[messages.size()];
+ messages.toArray(inbound);
+ assertTextMessagesEqual("Rollback did not work.", outbound, inbound);
+ }
+
+ /**
+ * spec section 3.6 acking a message with automation acks has no effect.
+ * @throws Exception
+ */
+ public void testAckMessageInTx() throws Exception {
+ Message[] outbound = new Message[]{session.createTextMessage("First Message")};
+
+ // sends a message
+ beginTx();
+ producer.send(outbound[0]);
+ outbound[0].acknowledge();
+ commitTx();
+ outbound[0].acknowledge();
+
+ // receives the first message
+ beginTx();
+ ArrayList<Message> messages = new ArrayList<>();
+ LOG.info("About to consume message 1");
+ Message message = consumer.receive(1000);
+ messages.add(message);
+ LOG.info("Received: " + message);
+
+ // validates that the rollbacked was not consumed
+ commitTx();
+ Message inbound[] = new Message[messages.size()];
+ messages.toArray(inbound);
+ assertTextMessagesEqual("Message not delivered.", outbound, inbound);
+ }
+
+ /**
+ * Sends a batch of messages and validates that the message sent before
+ * session close is not consumed.
+ *
+ * This test only works with local transactions, not xa.
+ * @throws Exception
+ */
+ public void testSendSessionClose() throws Exception {
+ Message[] outbound = new Message[]{session.createTextMessage("First Message"), session.createTextMessage("Second Message")};
+
+ // sends a message
+ beginTx();
+ producer.send(outbound[0]);
+ commitTx();
+
+ // sends a message that gets rollbacked
+ beginTx();
+ producer.send(session.createTextMessage("I'm going to get rolled back."));
+ consumer.close();
+
+ reconnectSession();
+
+ // sends a message
+ producer.send(outbound[1]);
+ commitTx();
+
+ // receives the first message
+ ArrayList<Message> messages = new ArrayList<>();
+ LOG.info("About to consume message 1");
+ beginTx();
+ Message message = consumer.receive(1000);
+ messages.add(message);
+ LOG.info("Received: " + message);
+
+ // receives the second message
+ LOG.info("About to consume message 2");
+ message = consumer.receive(4000);
+ messages.add(message);
+ LOG.info("Received: " + message);
+
+ // validates that the rollbacked was not consumed
+ commitTx();
+ Message inbound[] = new Message[messages.size()];
+ messages.toArray(inbound);
+ assertTextMessagesEqual("Rollback did not work.", outbound, inbound);
+ }
+
+ /**
+ * Sends a batch of messages and validates that the message sent before
+ * session close is not consumed.
+ *
+ * @throws Exception
+ */
+ public void testSendSessionAndConnectionClose() throws Exception {
+ Message[] outbound = new Message[]{session.createTextMessage("First Message"), session.createTextMessage("Second Message")};
+
+ // sends a message
+ beginTx();
+ producer.send(outbound[0]);
+ commitTx();
+
+ // sends a message that gets rollbacked
+ beginTx();
+ producer.send(session.createTextMessage("I'm going to get rolled back."));
+ consumer.close();
+ session.close();
+
+ reconnect();
+
+ // sends a message
+ beginTx();
+ producer.send(outbound[1]);
+ commitTx();
+
+ // receives the first message
+ ArrayList<Message> messages = new ArrayList<>();
+ LOG.info("About to consume message 1");
+ beginTx();
+ Message message = consumer.receive(1000);
+ messages.add(message);
+ LOG.info("Received: " + message);
+
+ // receives the second message
+ LOG.info("About to consume message 2");
+ message = consumer.receive(4000);
+ messages.add(message);
+ LOG.info("Received: " + message);
+
+ // validates that the rollbacked was not consumed
+ commitTx();
+ Message inbound[] = new Message[messages.size()];
+ messages.toArray(inbound);
+ assertTextMessagesEqual("Rollback did not work.", outbound, inbound);
+ }
+
+ /**
+ * Sends a batch of messages and validates that the rollbacked message was
+ * redelivered.
+ *
+ * @throws Exception
+ */
+ public void testReceiveRollback() throws Exception {
+ Message[] outbound = new Message[]{session.createTextMessage("First Message"), session.createTextMessage("Second Message")};
+
+ // lets consume any outstanding messages from prev test runs
+ beginTx();
+ while (consumer.receive(1000) != null) {
+ }
+ commitTx();
+
+ // sent both messages
+ beginTx();
+ producer.send(outbound[0]);
+ producer.send(outbound[1]);
+ commitTx();
+
+ LOG.info("Sent 0: " + outbound[0]);
+ LOG.info("Sent 1: " + outbound[1]);
+
+ ArrayList<Message> messages = new ArrayList<>();
+ beginTx();
+ Message message = consumer.receive(1000);
+ messages.add(message);
+ assertEquals(outbound[0], message);
+ commitTx();
+
+ // rollback so we can get that last message again.
+ beginTx();
+ message = consumer.receive(1000);
+ assertNotNull(message);
+ assertEquals(outbound[1], message);
+ rollbackTx();
+
+ // Consume again.. the prev message should
+ // get redelivered.
+ beginTx();
+ message = consumer.receive(5000);
+ assertNotNull("Should have re-received the message again!", message);
+ messages.add(message);
+ commitTx();
+
+ Message inbound[] = new Message[messages.size()];
+ messages.toArray(inbound);
+ assertTextMessagesEqual("Rollback did not work", outbound, inbound);
+ }
+
+ /**
+ * Sends a batch of messages and validates that the rollbacked message was
+ * redelivered.
+ *
+ * @throws Exception
+ */
+ public void testReceiveTwoThenRollback() throws Exception {
+ Message[] outbound = new Message[]{session.createTextMessage("First Message"), session.createTextMessage("Second Message")};
+
+ // lets consume any outstanding messages from prev test runs
+ beginTx();
+ while (consumer.receive(1000) != null) {
+ }
+ commitTx();
+
+ //
+ beginTx();
+ producer.send(outbound[0]);
+ producer.send(outbound[1]);
+ commitTx();
+
+ LOG.info("Sent 0: " + outbound[0]);
+ LOG.info("Sent 1: " + outbound[1]);
+
+ ArrayList<Message> messages = new ArrayList<>();
+ beginTx();
+ Message message = consumer.receive(1000);
+ assertEquals(outbound[0], message);
+
+ message = consumer.receive(1000);
+ assertNotNull(message);
+ assertEquals(outbound[1], message);
+ rollbackTx();
+
+ // Consume again.. the prev message should
+ // get redelivered.
+ beginTx();
+ message = consumer.receive(5000);
+ assertNotNull("Should have re-received the first message again!", message);
+ messages.add(message);
+ assertEquals(outbound[0], message);
+ message = consumer.receive(5000);
+ assertNotNull("Should have re-received the second message again!", message);
+ messages.add(message);
+ assertEquals(outbound[1], message);
+
+ assertNull(consumer.receiveNoWait());
+ commitTx();
+
+ Message inbound[] = new Message[messages.size()];
+ messages.toArray(inbound);
+ assertTextMessagesEqual("Rollback did not work", outbound, inbound);
+ }
+
+ /**
+ * Sends a batch of messages and validates that the rollbacked message was
+ * not consumed.
+ *
+ * @throws Exception
+ */
+ public void testSendReceiveWithPrefetchOne() throws Exception {
+ setPrefetchToOne();
+ Message[] outbound = new Message[]{session.createTextMessage("First Message"), session.createTextMessage("Second Message"), session.createTextMessage("Third Message"), session.createTextMessage("Fourth Message")};
+
+ beginTx();
+ for (int i = 0; i < outbound.length; i++) {
+ // sends a message
+ producer.send(outbound[i]);
+ }
+ commitTx();
+
+ // receives the first message
+ beginTx();
+ for (int i = 0; i < outbound.length; i++) {
+ LOG.info("About to consume message 1");
+ Message message = consumer.receive(1000);
+ assertNotNull(message);
+ LOG.info("Received: " + message);
+ }
+
+ // validates that the rollbacked was not consumed
+ commitTx();
+ }
+
+ /**
+ * Perform the test that validates if the rollbacked message was redelivered
+ * multiple times.
+ *
+ * @throws Exception
+ */
+ public void testReceiveTwoThenRollbackManyTimes() throws Exception {
+ for (int i = 0; i < 5; i++) {
+ testReceiveTwoThenRollback();
+ }
+ }
+
+ /**
+ * Sends a batch of messages and validates that the rollbacked message was
+ * not consumed. This test differs by setting the message prefetch to one.
+ *
+ * @throws Exception
+ */
+ public void testSendRollbackWithPrefetchOfOne() throws Exception {
+ setPrefetchToOne();
+ testSendRollback();
+ }
+
+ /**
+ * Sends a batch of messages and and validates that the rollbacked message
+ * was redelivered. This test differs by setting the message prefetch to
+ * one.
+ *
+ * @throws Exception
+ */
+ public void testReceiveRollbackWithPrefetchOfOne() throws Exception {
+ setPrefetchToOne();
+ testReceiveRollback();
+ }
+
+ /**
+ * Tests if the messages can still be received if the consumer is closed
+ * (session is not closed).
+ *
+ * @throws Exception see http://jira.codehaus.org/browse/AMQ-143
+ */
+ public void testCloseConsumerBeforeCommit() throws Exception {
+ TextMessage[] outbound = new TextMessage[]{session.createTextMessage("First Message"), session.createTextMessage("Second Message")};
+
+ // lets consume any outstanding messages from prev test runs
+ beginTx();
+ while (consumer.receiveNoWait() != null) {
+ }
+
+ commitTx();
+
+ // sends the messages
+ beginTx();
+ producer.send(outbound[0]);
+ producer.send(outbound[1]);
+ commitTx();
+ LOG.info("Sent 0: " + outbound[0]);
+ LOG.info("Sent 1: " + outbound[1]);
+
+ beginTx();
+ TextMessage message = (TextMessage) consumer.receive(1000);
+ assertEquals(outbound[0].getText(), message.getText());
+ // Close the consumer before the commit. This should not cause the
+ // received message
+ // to rollback.
+ consumer.close();
+ commitTx();
+
+ // Create a new consumer
+ consumer = resourceProvider.createConsumer(session, destination);
+ LOG.info("Created consumer: " + consumer);
+
+ beginTx();
+ message = (TextMessage) consumer.receive(1000);
+ assertEquals(outbound[1].getText(), message.getText());
+ commitTx();
+ }
+
+ public void testChangeMutableObjectInObjectMessageThenRollback() throws Exception {
+ ArrayList<String> list = new ArrayList<>();
+ list.add("First");
+ Message outbound = session.createObjectMessage(list);
+ outbound.setStringProperty("foo", "abc");
+
+ beginTx();
+ producer.send(outbound);
+ commitTx();
+
+ LOG.info("About to consume message 1");
+ beginTx();
+ Message message = consumer.receive(5000);
+
+ List<String> body = assertReceivedObjectMessageWithListBody(message);
+
+ // now lets try mutate it
+ try {
+ message.setStringProperty("foo", "def");
+ fail("Cannot change properties of the object!");
+ } catch (JMSException e) {
+ LOG.info("Caught expected exception: " + e, e);
+ }
+ body.clear();
+ body.add("This should never be seen!");
+ rollbackTx();
+
+ beginTx();
+ message = consumer.receive(5000);
+ List<String> secondBody = assertReceivedObjectMessageWithListBody(message);
+ assertNotSame("Second call should return a different body", secondBody, body);
+ commitTx();
+ }
+
+ @SuppressWarnings("unchecked")
+ protected List<String> assertReceivedObjectMessageWithListBody(Message message) throws JMSException {
+ assertNotNull("Should have received a message!", message);
+ assertEquals("foo header", "abc", message.getStringProperty("foo"));
+
+ assertTrue("Should be an object message but was: " + message, message instanceof ObjectMessage);
+ ObjectMessage objectMessage = (ObjectMessage) message;
+ List<String> body = (List<String>) objectMessage.getObject();
+ LOG.info("Received body: " + body);
+
+ assertEquals("Size of list should be 1", 1, body.size());
+ assertEquals("element 0 of list", "First", body.get(0));
+ return body;
+ }
+
+ /**
+ * Recreates the connection.
+ *
+ * @throws javax.jms.JMSException
+ */
+ protected void reconnect() throws Exception {
+
+ if (connection != null) {
+ // Close the prev connection.
+ connection.close();
+ }
+ session = null;
+ connection = resourceProvider.createConnection(connectionFactory);
+ reconnectSession();
+ connection.start();
+ }
+
+ /**
+ * Recreates the connection.
+ *
+ * @throws javax.jms.JMSException
+ */
+ protected void reconnectSession() throws JMSException {
+ if (session != null) {
+ session.close();
+ }
+
+ session = resourceProvider.createSession(connection);
+ destination = resourceProvider.createDestination(session, getSubject());
+ producer = resourceProvider.createProducer(session, destination);
+ consumer = resourceProvider.createConsumer(session, destination);
+ }
+
+ /**
+ * Sets the prefeftch policy to one.
+ */
+ protected void setPrefetchToOne() {
+ ActiveMQPrefetchPolicy prefetchPolicy = getPrefetchPolicy();
+ prefetchPolicy.setQueuePrefetch(1);
+ prefetchPolicy.setTopicPrefetch(1);
+ prefetchPolicy.setDurableTopicPrefetch(1);
+ prefetchPolicy.setOptimizeDurableTopicPrefetch(1);
+ }
+
+ protected ActiveMQPrefetchPolicy getPrefetchPolicy() {
+ return ((ActiveMQConnection) connection).getPrefetchPolicy();
+ }
+
+ //This test won't work with xa tx so no beginTx() has been added.
+ public void testMessageListener() throws Exception {
+ // send messages
+ for (int i = 0; i < MESSAGE_COUNT; i++) {
+ producer.send(session.createTextMessage(MESSAGE_TEXT + i));
+ }
+ commitTx();
+ consumer.setMessageListener(this);
+ // wait receive
+ waitReceiveUnack();
+ assertEquals(unackMessages.size(), MESSAGE_COUNT);
+ // resend phase
+ waitReceiveAck();
+ assertEquals(ackMessages.size(), MESSAGE_COUNT);
+ // should no longer re-receive
+ consumer.setMessageListener(null);
+ assertNull(consumer.receive(500));
+ reconnect();
+ }
+
+ @Override
+ public void onMessage(Message message) {
+ if (!resendPhase) {
+ unackMessages.add(message);
+ if (unackMessages.size() == MESSAGE_COUNT) {
+ try {
+ rollbackTx();
+ resendPhase = true;
+ } catch (Exception e) {
+ e.printStackTrace();
}
- messageSent();
- commitTx();
- LOG.info("Consuming bacth " + j + " of " + batchSize + " messages");
-
- beginTx();
- for (int i = 0; i < batchSize; i++) {
- message = (TextMessage)consumer.receive(1000 * 5);
- assertNotNull("Received only " + i + " messages in batch " + j, message);
- assertEquals("Batch Message", message.getText());
+ }
+ } else {
+ ackMessages.add(message);
+ if (ackMessages.size() == MESSAGE_COUNT) {
+ try {
+ commitTx();
+ } catch (Exception e) {
+ e.printStackTrace();
}
-
- commitTx();
- }
- }
-
- protected void messageSent() throws Exception {
- }
-
- /**
- * Sends a batch of messages and validates that the rollbacked message was
- * not consumed.
- *
- * @throws Exception
- */
- public void testSendRollback() throws Exception {
- Message[] outbound = new Message[] {session.createTextMessage("First Message"), session.createTextMessage("Second Message")};
-
- // sends a message
- beginTx();
- producer.send(outbound[0]);
- commitTx();
-
- // sends a message that gets rollbacked
- beginTx();
- producer.send(session.createTextMessage("I'm going to get rolled back."));
- rollbackTx();
-
- // sends a message
- beginTx();
- producer.send(outbound[1]);
- commitTx();
-
- // receives the first message
- beginTx();
- ArrayList<Message> messages = new ArrayList<>();
- LOG.info("About to consume message 1");
- Message message = consumer.receive(1000);
- messages.add(message);
- LOG.info("Received: " + message);
-
- // receives the second message
- LOG.info("About to consume message 2");
- message = consumer.receive(4000);
- messages.add(message);
- LOG.info("Received: " + message);
-
- // validates that the rollbacked was not consumed
- commitTx();
- Message inbound[] = new Message[messages.size()];
- messages.toArray(inbound);
- assertTextMessagesEqual("Rollback did not work.", outbound, inbound);
- }
-
- /**
- * spec section 3.6 acking a message with automation acks has no effect.
- * @throws Exception
- */
- public void testAckMessageInTx() throws Exception {
- Message[] outbound = new Message[] {session.createTextMessage("First Message")};
-
- // sends a message
- beginTx();
- producer.send(outbound[0]);
- outbound[0].acknowledge();
- commitTx();
- outbound[0].acknowledge();
-
- // receives the first message
- beginTx();
- ArrayList<Message> messages = new ArrayList<>();
- LOG.info("About to consume message 1");
- Message message = consumer.receive(1000);
- messages.add(message);
- LOG.info("Received: " + message);
-
- // validates that the rollbacked was not consumed
- commitTx();
- Message inbound[] = new Message[messages.size()];
- messages.toArray(inbound);
- assertTextMessagesEqual("Message not delivered.", outbound, inbound);
- }
-
- /**
- * Sends a batch of messages and validates that the message sent before
- * session close is not consumed.
- *
- * This test only works with local transactions, not xa.
- * @throws Exception
- */
- public void testSendSessionClose() throws Exception {
- Message[] outbound = new Message[] {session.createTextMessage("First Message"), session.createTextMessage("Second Message")};
-
- // sends a message
- beginTx();
- producer.send(outbound[0]);
- commitTx();
-
- // sends a message that gets rollbacked
- beginTx();
- producer.send(session.createTextMessage("I'm going to get rolled back."));
- consumer.close();
-
- reconnectSession();
-
- // sends a message
- producer.send(outbound[1]);
- commitTx();
-
- // receives the first message
- ArrayList<Message> messages = new ArrayList<>();
- LOG.info("About to consume message 1");
- beginTx();
- Message message = consumer.receive(1000);
- messages.add(message);
- LOG.info("Received: " + message);
-
- // receives the second message
- LOG.info("About to consume message 2");
- message = consumer.receive(4000);
- messages.add(message);
- LOG.info("Received: " + message);
-
- // validates that the rollbacked was not consumed
- commitTx();
- Message inbound[] = new Message[messages.size()];
- messages.toArray(inbound);
- assertTextMessagesEqual("Rollback did not work.", outbound, inbound);
- }
-
- /**
- * Sends a batch of messages and validates that the message sent before
- * session close is not consumed.
- *
- * @throws Exception
- */
- public void testSendSessionAndConnectionClose() throws Exception {
- Message[] outbound = new Message[] {session.createTextMessage("First Message"), session.createTextMessage("Second Message")};
-
- // sends a message
- beginTx();
- producer.send(outbound[0]);
- commitTx();
-
- // sends a message that gets rollbacked
- beginTx();
- producer.send(session.createTextMessage("I'm going to get rolled back."));
- consumer.close();
- session.close();
-
- reconnect();
-
- // sends a message
- beginTx();
- producer.send(outbound[1]);
- commitTx();
-
- // receives the first message
- ArrayList<Message> messages = new ArrayList<>();
- LOG.info("About to consume message 1");
- beginTx();
- Message message = consumer.receive(1000);
- messages.add(message);
- LOG.info("Received: " + message);
-
- // receives the second message
- LOG.info("About to consume message 2");
- message = consumer.receive(4000);
- messages.add(message);
- LOG.info("Received: " + message);
-
- // validates that the rollbacked was not consumed
- commitTx();
- Message inbound[] = new Message[messages.size()];
- messages.toArray(inbound);
- assertTextMessagesEqual("Rollback did not work.", outbound, inbound);
- }
-
- /**
- * Sends a batch of messages and validates that the rollbacked message was
- * redelivered.
- *
- * @throws Exception
- */
- public void testReceiveRollback() throws Exception {
- Message[] outbound = new Message[] {session.createTextMessage("First Message"), session.createTextMessage("Second Message")};
-
- // lets consume any outstanding messages from prev test runs
- beginTx();
- while (consumer.receive(1000) != null) {
- }
- commitTx();
-
- // sent both messages
- beginTx();
- producer.send(outbound[0]);
- producer.send(outbound[1]);
- commitTx();
-
- LOG.info("Sent 0: " + outbound[0]);
- LOG.info("Sent 1: " + outbound[1]);
-
- ArrayList<Message> messages = new ArrayList<>();
- beginTx();
- Message message = consumer.receive(1000);
- messages.add(message);
- assertEquals(outbound[0], message);
- commitTx();
-
- // rollback so we can get that last message again.
- beginTx();
- message = consumer.receive(1000);
- assertNotNull(message);
- assertEquals(outbound[1], message);
- rollbackTx();
-
- // Consume again.. the prev message should
- // get redelivered.
- beginTx();
- message = consumer.receive(5000);
- assertNotNull("Should have re-received the message again!", message);
- messages.add(message);
- commitTx();
-
- Message inbound[] = new Message[messages.size()];
- messages.toArray(inbound);
- assertTextMessagesEqual("Rollback did not work", outbound, inbound);
- }
-
- /**
- * Sends a batch of messages and validates that the rollbacked message was
- * redelivered.
- *
- * @throws Exception
- */
- public void testReceiveTwoThenRollback() throws Exception {
- Message[] outbound = new Message[] {session.createTextMessage("First Message"), session.createTextMessage("Second Message")};
-
- // lets consume any outstanding messages from prev test runs
- beginTx();
- while (consumer.receive(1000) != null) {
- }
- commitTx();
-
- //
- beginTx();
- producer.send(outbound[0]);
- producer.send(outbound[1]);
- commitTx();
-
- LOG.info("Sent 0: " + outbound[0]);
- LOG.info("Sent 1: " + outbound[1]);
-
- ArrayList<Message> messages = new ArrayList<>();
- beginTx();
- Message message = consumer.receive(1000);
- assertEquals(outbound[0], message);
-
- message = consumer.receive(1000);
- assertNotNull(message);
- assertEquals(outbound[1], message);
- rollbackTx();
-
- // Consume again.. the prev message should
- // get redelivered.
- beginTx();
- message = consumer.receive(5000);
- assertNotNull("Should have re-received the first message again!", message);
- messages.add(message);
- assertEquals(outbound[0], message);
- message = consumer.receive(5000);
- assertNotNull("Should have re-received the second message again!", message);
- messages.add(message);
- assertEquals(outbound[1], message);
-
- assertNull(consumer.receiveNoWait());
- commitTx();
-
- Message inbound[] = new Message[messages.size()];
- messages.toArray(inbound);
- assertTextMessagesEqual("Rollback did not work", outbound, inbound);
- }
-
- /**
- * Sends a batch of messages and validates that the rollbacked message was
- * not consumed.
- *
- * @throws Exception
- */
- public void testSendReceiveWithPrefetchOne() throws Exception {
- setPrefetchToOne();
- Message[] outbound = new Message[] {session.createTextMessage("First Message"), session.createTextMessage("Second Message"), session.createTextMessage("Third Message"),
- session.createTextMessage("Fourth Message")};
-
- beginTx();
- for (int i = 0; i < outbound.length; i++) {
- // sends a message
- producer.send(outbound[i]);
- }
- commitTx();
-
- // receives the first message
- beginTx();
- for (int i = 0; i < outbound.length; i++) {
- LOG.info("About to consume message 1");
- Message message = consumer.receive(1000);
- assertNotNull(message);
- LOG.info("Received: " + message);
- }
-
- // validates that the rollbacked was not consumed
- commitTx();
- }
-
- /**
- * Perform the test that validates if the rollbacked message was redelivered
- * multiple times.
- *
- * @throws Exception
- */
- public void testReceiveTwoThenRollbackManyTimes() throws Exception {
- for (int i = 0; i < 5; i++) {
- testReceiveTwoThenRollback();
- }
- }
-
- /**
- * Sends a batch of messages and validates that the rollbacked message was
- * not consumed. This test differs by setting the message prefetch to one.
- *
- * @throws Exception
- */
- public void testSendRollbackWithPrefetchOfOne() throws Exception {
- setPrefetchToOne();
- testSendRollback();
- }
-
- /**
- * Sends a batch of messages and and validates that the rollbacked message
- * was redelivered. This test differs by setting the message prefetch to
- * one.
- *
- * @throws Exception
- */
- public void testReceiveRollbackWithPrefetchOfOne() throws Exception {
- setPrefetchToOne();
- testReceiveRollback();
- }
-
- /**
- * Tests if the messages can still be received if the consumer is closed
- * (session is not closed).
- *
- * @throws Exception see http://jira.codehaus.org/browse/AMQ-143
- */
- public void testCloseConsumerBeforeCommit() throws Exception {
- TextMessage[] outbound = new TextMessage[] {session.createTextMessage("First Message"), session.createTextMessage("Second Message")};
-
- // lets consume any outstanding messages from prev test runs
- beginTx();
- while (consumer.receiveNoWait() != null) {
- }
-
- commitTx();
-
- // sends the messages
- beginTx();
- producer.send(outbound[0]);
- producer.send(outbound[1]);
- commitTx();
- LOG.info("Sent 0: " + outbound[0]);
- LOG.info("Sent 1: " + outbound[1]);
-
- beginTx();
- TextMessage message = (TextMessage)consumer.receive(1000);
- assertEquals(outbound[0].getText(), message.getText());
- // Close the consumer before the commit. This should not cause the
- // received message
- // to rollback.
- consumer.close();
- commitTx();
-
- // Create a new consumer
- consumer = resourceProvider.createConsumer(session, destination);
- LOG.info("Created consumer: " + consumer);
-
- beginTx();
- message = (TextMessage)consumer.receive(1000);
- assertEquals(outbound[1].getText(), message.getText());
- commitTx();
- }
-
- public void testChangeMutableObjectInObjectMessageThenRollback() throws Exception {
- ArrayList<String> list = new ArrayList<>();
- list.add("First");
- Message outbound = session.createObjectMessage(list);
- outbound.setStringProperty("foo", "abc");
-
- beginTx();
- producer.send(outbound);
- commitTx();
-
- LOG.info("About to consume message 1");
- beginTx();
- Message message = consumer.receive(5000);
-
- List<String> body = assertReceivedObjectMessageWithListBody(message);
-
- // now lets try mutate it
- try {
- message.setStringProperty("foo", "def");
- fail("Cannot change properties of the object!");
- } catch (JMSException e) {
- LOG.info("Caught expected exception: " + e, e);
- }
- body.clear();
- body.add("This should never be seen!");
- rollbackTx();
-
- beginTx();
- message = consumer.receive(5000);
- List<String> secondBody = assertReceivedObjectMessageWithListBody(message);
- assertNotSame("Second call should return a different body", secondBody, body);
- commitTx();
- }
-
- @SuppressWarnings("unchecked")
- protected List<String> assertReceivedObjectMessageWithListBody(Message message) throws JMSException {
- assertNotNull("Should have received a message!", message);
- assertEquals("foo header", "abc", message.getStringProperty("foo"));
-
- assertTrue("Should be an object message but was: " + message, message instanceof ObjectMessage);
- ObjectMessage objectMessage = (ObjectMessage)message;
- List<String> body = (List<String>)objectMessage.getObject();
- LOG.info("Received body: " + body);
-
- assertEquals("Size of list should be 1", 1, body.size());
- assertEquals("element 0 of list", "First", body.get(0));
- return body;
- }
-
- /**
- * Recreates the connection.
- *
- * @throws javax.jms.JMSException
- */
- protected void reconnect() throws Exception {
-
- if (connection != null) {
- // Close the prev connection.
- connection.close();
- }
- session = null;
- connection = resourceProvider.createConnection(connectionFactory);
- reconnectSession();
- connection.start();
- }
-
- /**
- * Recreates the connection.
- *
- * @throws javax.jms.JMSException
- */
- protected void reconnectSession() throws JMSException {
- if (session != null) {
- session.close();
- }
-
- session = resourceProvider.createSession(connection);
- destination = resourceProvider.createDestination(session, getSubject());
- producer = resourceProvider.createProducer(session, destination);
- consumer = resourceProvider.createConsumer(session, destination);
- }
-
- /**
- * Sets the prefeftch policy to one.
- */
- protected void setPrefetchToOne() {
- ActiveMQPrefetchPolicy prefetchPolicy = getPrefetchPolicy();
- prefetchPolicy.setQueuePrefetch(1);
- prefetchPolicy.setTopicPrefetch(1);
- prefetchPolicy.setDurableTopicPrefetch(1);
- prefetchPolicy.setOptimizeDurableTopicPrefetch(1);
- }
-
- protected ActiveMQPrefetchPolicy getPrefetchPolicy() {
- return ((ActiveMQConnection)connection).getPrefetchPolicy();
- }
-
- //This test won't work with xa tx so no beginTx() has been added.
- public void testMessageListener() throws Exception {
- // send messages
- for (int i = 0; i < MESSAGE_COUNT; i++) {
- producer.send(session.createTextMessage(MESSAGE_TEXT + i));
- }
- commitTx();
- consumer.setMessageListener(this);
- // wait receive
- waitReceiveUnack();
- assertEquals(unackMessages.size(), MESSAGE_COUNT);
- // resend phase
- waitReceiveAck();
- assertEquals(ackMessages.size(), MESSAGE_COUNT);
- // should no longer re-receive
- consumer.setMessageListener(null);
- assertNull(consumer.receive(500));
- reconnect();
- }
-
- @Override
- public void onMessage(Message message) {
- if (!resendPhase) {
- unackMessages.add(message);
- if (unackMessages.size() == MESSAGE_COUNT) {
- try {
- rollbackTx();
- resendPhase = true;
- } catch (Exception e) {
- e.printStackTrace();
- }
- }
- } else {
- ackMessages.add(message);
- if (ackMessages.size() == MESSAGE_COUNT) {
- try {
- commitTx();
- } catch (Exception e) {
- e.printStackTrace();
- }
- }
- }
- }
-
- private void waitReceiveUnack() throws Exception {
- for (int i = 0; i < 100 && !resendPhase; i++) {
- Thread.sleep(100);
- }
- assertTrue(resendPhase);
- }
-
- private void waitReceiveAck() throws Exception {
- for (int i = 0; i < 100 && ackMessages.size() < MESSAGE_COUNT; i++) {
- Thread.sleep(100);
- }
- assertFalse(ackMessages.size() < MESSAGE_COUNT);
- }
+ }
+ }
+ }
+
+ private void waitReceiveUnack() throws Exception {
+ for (int i = 0; i < 100 && !resendPhase; i++) {
+ Thread.sleep(100);
+ }
+ assertTrue(resendPhase);
+ }
+
+ private void waitReceiveAck() throws Exception {
+ for (int i = 0; i < 100 && ackMessages.size() < MESSAGE_COUNT; i++) {
+ Thread.sleep(100);
+ }
+ assertFalse(ackMessages.size() < MESSAGE_COUNT);
+ }
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ec48f9ed/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/LargeMessageTestSupport.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/LargeMessageTestSupport.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/LargeMessageTestSupport.java
index 889ec76..b20b6d5 100644
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/LargeMessageTestSupport.java
+++ b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/LargeMessageTestSupport.java
@@ -6,7 +6,7 @@
* (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
+ * 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,
@@ -17,8 +17,6 @@
package org.apache.activemq;
-import java.util.concurrent.atomic.AtomicInteger;
-
import javax.jms.BytesMessage;
import javax.jms.Connection;
import javax.jms.DeliveryMode;
@@ -30,6 +28,7 @@ import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.Topic;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.activemq.command.ActiveMQMessage;
import org.apache.activemq.command.ActiveMQQueue;
@@ -69,8 +68,7 @@ public class LargeMessageTestSupport extends ClientTestSupport implements Messag
String subject = getClass().getName();
if (isTopic) {
return new ActiveMQTopic(subject);
- }
- else {
+ } else {
return new ActiveMQQueue(subject);
}
}
@@ -78,8 +76,7 @@ public class LargeMessageTestSupport extends ClientTestSupport implements Messag
protected MessageConsumer createConsumer() throws JMSException {
if (isTopic && isDurable) {
return consumerSession.createDurableSubscriber((Topic) destination, idGen.generateId());
- }
- else {
+ } else {
return consumerSession.createConsumer(destination);
}
}
@@ -96,8 +93,7 @@ public class LargeMessageTestSupport extends ClientTestSupport implements Messag
for (int i = 0; i < LARGE_MESSAGE_SIZE; i++) {
if (i % 2 == 0) {
largeMessageData[i] = 'a';
- }
- else {
+ } else {
largeMessageData[i] = 'z';
}
}
@@ -105,8 +101,7 @@ public class LargeMessageTestSupport extends ClientTestSupport implements Messag
try {
// allow the broker to start
Thread.sleep(1000);
- }
- catch (InterruptedException e) {
+ } catch (InterruptedException e) {
throw new JMSException(e.getMessage());
}
@@ -176,8 +171,7 @@ public class LargeMessageTestSupport extends ClientTestSupport implements Messag
if (messageCount.get() % 50 == 0) {
LOG.info("count = " + messageCount);
}
- }
- catch (Exception e) {
+ } catch (Exception e) {
e.printStackTrace();
}
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/ec48f9ed/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/LoadTestBurnIn.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/LoadTestBurnIn.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/LoadTestBurnIn.java
index bedfd47..3a4acfd 100644
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/LoadTestBurnIn.java
+++ b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/LoadTestBurnIn.java
@@ -6,7 +6,7 @@
* (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
+ * 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,
@@ -16,12 +16,6 @@
*/
package org.apache.activemq;
-import java.io.IOException;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-
import javax.jms.BytesMessage;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
@@ -32,6 +26,11 @@ import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.Topic;
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import junit.framework.Test;
@@ -71,11 +70,9 @@ public class LoadTestBurnIn extends JmsTestSupport {
protected void tearDown() throws Exception {
try {
super.tearDown();
- }
- catch (Throwable e) {
+ } catch (Throwable e) {
e.printStackTrace(System.out);
- }
- finally {
+ } finally {
LOG.info("End: " + getName());
}
}
@@ -119,8 +116,7 @@ public class LoadTestBurnIn extends JmsTestSupport {
MessageConsumer consumer;
if (durableConsumer) {
consumer = session.createDurableSubscriber((Topic) destination, "sub1:" + System.currentTimeMillis());
- }
- else {
+ } else {
consumer = session.createConsumer(destination);
}
profilerPause("Ready: ");
@@ -143,11 +139,9 @@ public class LoadTestBurnIn extends JmsTestSupport {
producer.send(m);
}
producer.close();
- }
- catch (JMSException e) {
+ } catch (JMSException e) {
e.printStackTrace();
- }
- finally {
+ } finally {
safeClose(connection2);
producerDoneLatch.countDown();
}