You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ta...@apache.org on 2013/12/05 21:52:56 UTC
git commit: https://issues.apache.org/jira/browse/AMQ-4887
Updated Branches:
refs/heads/trunk 30f3a9b7f -> d0f1dd666
https://issues.apache.org/jira/browse/AMQ-4887
Update test case to show other broken scenarios.
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/d0f1dd66
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/d0f1dd66
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/d0f1dd66
Branch: refs/heads/trunk
Commit: d0f1dd6668ed56c2369525818031096cd87cee98
Parents: 30f3a9b
Author: Timothy Bish <ta...@gmai.com>
Authored: Thu Dec 5 15:52:05 2013 -0500
Committer: Timothy Bish <ta...@gmai.com>
Committed: Thu Dec 5 15:52:27 2013 -0500
----------------------------------------------------------------------
.../org/apache/activemq/bugs/AMQ4887Test.java | 124 ++++++++++++++++---
1 file changed, 104 insertions(+), 20 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/d0f1dd66/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4887Test.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4887Test.java b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4887Test.java
index f29f605..cf33ece 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4887Test.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4887Test.java
@@ -16,48 +16,65 @@
*/
package org.apache.activemq.bugs;
-import org.apache.activemq.ActiveMQConnection;
-import org.apache.activemq.ActiveMQConnectionFactory;
-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.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
import javax.jms.BytesMessage;
+import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
+import javax.jms.StreamMessage;
-import static org.junit.Assert.*;
+import org.apache.activemq.ActiveMQConnection;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TestName;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class AMQ4887Test {
private static final transient Logger LOG = LoggerFactory.getLogger(AMQ4887Test.class);
private static final Integer ITERATIONS = 10;
+
@Rule
public TestName name = new TestName();
@Test
- public void testSetPropertyBeforeCopy() throws Exception {
+ public void testBytesMessageSetPropertyBeforeCopy() throws Exception {
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://localhost");
ActiveMQConnection connection = (ActiveMQConnection) connectionFactory.createConnection();
connection.start();
+ doTestBytesMessageSetPropertyBeforeCopy(connection);
+ }
+
+ @Test
+ public void testBytesMessageSetPropertyBeforeCopyCompressed() throws Exception {
+ ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://localhost");
+ connectionFactory.setUseCompression(true);
+ ActiveMQConnection connection = (ActiveMQConnection) connectionFactory.createConnection();
+ connection.start();
+ doTestBytesMessageSetPropertyBeforeCopy(connection);
+ }
+
+ public void doTestBytesMessageSetPropertyBeforeCopy(Connection connection) throws Exception {
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createQueue(name.toString());
MessageConsumer consumer = session.createConsumer(destination);
MessageProducer producer = session.createProducer(destination);
- byte[] messageContent = "bytes message".getBytes();
- BytesMessage bytesMessage = session.createBytesMessage();
- bytesMessage.writeBytes(messageContent);
+ BytesMessage message = session.createBytesMessage();
for (int i=0; i < ITERATIONS; i++) {
+
long sendTime = System.currentTimeMillis();
- bytesMessage.setLongProperty("sendTime", sendTime);
- producer.send(bytesMessage);
+ message.setLongProperty("sendTime", sendTime);
+ producer.send(message);
LOG.debug("Receiving message " + i);
Message receivedMessage = consumer.receive(5000);
@@ -65,16 +82,83 @@ public class AMQ4887Test {
assertTrue("On message " + i, receivedMessage instanceof BytesMessage);
BytesMessage receivedBytesMessage = (BytesMessage) receivedMessage;
- byte[] receivedBytes = new byte[(int) receivedBytesMessage.getBodyLength()];
- receivedBytesMessage.readBytes(receivedBytes);
- LOG.debug("Message " + i + " content [" + new String(receivedBytes) + "]");
- assertEquals("On message " + i, messageContent.length, receivedBytes.length);
- assertArrayEquals("On message " + i, messageContent, receivedBytes);
+
+ int numElements = 0;
+ try {
+ while (true) {
+ receivedBytesMessage.readBoolean();
+ numElements++;
+ }
+ } catch (Exception ex) {
+ }
+
+ LOG.info("Iteration [{}]: Received Message contained {} boolean values.", i, numElements);
+ assertEquals(i, numElements);
long receivedSendTime = receivedBytesMessage.getLongProperty("sendTime");
assertEquals("On message " + i, receivedSendTime, sendTime);
- Thread.sleep(10);
+ // Add a new bool value on each iteration.
+ message.writeBoolean(true);
+ }
+ }
+
+ @Test
+ public void testStreamMessageSetPropertyBeforeCopy() throws Exception {
+ ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://localhost");
+ ActiveMQConnection connection = (ActiveMQConnection) connectionFactory.createConnection();
+ connection.start();
+ doTestStreamMessageSetPropertyBeforeCopy(connection);
+ }
+
+ @Test
+ public void testStreamMessageSetPropertyBeforeCopyCompressed() throws Exception {
+ ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://localhost");
+ connectionFactory.setUseCompression(true);
+ ActiveMQConnection connection = (ActiveMQConnection) connectionFactory.createConnection();
+ connection.start();
+ doTestStreamMessageSetPropertyBeforeCopy(connection);
+ }
+
+ public void doTestStreamMessageSetPropertyBeforeCopy(Connection connection) throws Exception {
+
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Destination destination = session.createQueue(name.toString());
+ MessageConsumer consumer = session.createConsumer(destination);
+ MessageProducer producer = session.createProducer(destination);
+
+ StreamMessage message = session.createStreamMessage();
+
+ for (int i=0; i < ITERATIONS; i++) {
+
+ long sendTime = System.currentTimeMillis();
+ message.setLongProperty("sendTime", sendTime);
+ producer.send(message);
+
+ LOG.debug("Receiving message " + i);
+ Message receivedMessage = consumer.receive(5000);
+ assertNotNull("On message " + i, receivedMessage);
+ assertTrue("On message " + i, receivedMessage instanceof StreamMessage);
+
+ StreamMessage receivedStreamMessage = (StreamMessage) receivedMessage;
+
+ int numElements = 0;
+ try {
+ while (true) {
+ receivedStreamMessage.readBoolean();
+ numElements++;
+ }
+ } catch (Exception ex) {
+ }
+
+ LOG.info("Iteration [{}]: Received Message contained {} boolean values.", i, numElements);
+ assertEquals(i, numElements);
+
+ long receivedSendTime = receivedStreamMessage.getLongProperty("sendTime");
+ assertEquals("On message " + i, receivedSendTime, sendTime);
+
+ // Add a new bool value on each iteration.
+ message.writeBoolean(true);
}
}