Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

https://issues.apache.org/jira/browse/AMQ-4887 

Update test case to show other broken scenarios.
This commit is contained in:
Timothy Bish 2013-12-05 15:52:05 -05:00
parent 30f3a9b7f3
commit d0f1dd6668
1 changed files with 105 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

126
activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4887Test.java
View File

@ -16,6 +16,19 @@
*/ */
package org.apache.activemq.bugs; package org.apache.activemq.bugs;
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 org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.ActiveMQConnectionFactory;
import org.junit.Rule; import org.junit.Rule;
@ -24,40 +37,44 @@ import org.junit.rules.TestName;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import javax.jms.BytesMessage;
import javax.jms.Destination;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import static org.junit.Assert.*;
public class AMQ4887Test { public class AMQ4887Test {
private static final transient Logger LOG = LoggerFactory.getLogger(AMQ4887Test.class); private static final transient Logger LOG = LoggerFactory.getLogger(AMQ4887Test.class);
private static final Integer ITERATIONS = 10; private static final Integer ITERATIONS = 10;
@Rule @Rule
public TestName name = new TestName(); public TestName name = new TestName();
@Test @Test
public void testSetPropertyBeforeCopy() throws Exception { public void testBytesMessageSetPropertyBeforeCopy() throws Exception {
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://localhost"); ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://localhost");
ActiveMQConnection connection = (ActiveMQConnection) connectionFactory.createConnection(); ActiveMQConnection connection = (ActiveMQConnection) connectionFactory.createConnection();
connection.start(); 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); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createQueue(name.toString()); Destination destination = session.createQueue(name.toString());
MessageConsumer consumer = session.createConsumer(destination); MessageConsumer consumer = session.createConsumer(destination);
MessageProducer producer = session.createProducer(destination); MessageProducer producer = session.createProducer(destination);
byte[] messageContent = "bytes message".getBytes(); BytesMessage message = session.createBytesMessage();
BytesMessage bytesMessage = session.createBytesMessage();
bytesMessage.writeBytes(messageContent);
for (int i=0; i < ITERATIONS; i++) { for (int i=0; i < ITERATIONS; i++) {
long sendTime = System.currentTimeMillis(); long sendTime = System.currentTimeMillis();
bytesMessage.setLongProperty("sendTime", sendTime); message.setLongProperty("sendTime", sendTime);
producer.send(bytesMessage); producer.send(message);
LOG.debug("Receiving message " + i); LOG.debug("Receiving message " + i);
Message receivedMessage = consumer.receive(5000); Message receivedMessage = consumer.receive(5000);
@ -65,16 +82,83 @@ public class AMQ4887Test {
assertTrue("On message " + i, receivedMessage instanceof BytesMessage); assertTrue("On message " + i, receivedMessage instanceof BytesMessage);
BytesMessage receivedBytesMessage = (BytesMessage) receivedMessage; BytesMessage receivedBytesMessage = (BytesMessage) receivedMessage;
byte[] receivedBytes = new byte[(int) receivedBytesMessage.getBodyLength()];
receivedBytesMessage.readBytes(receivedBytes); int numElements = 0;
LOG.debug("Message " + i + " content [" + new String(receivedBytes) + "]"); try {
assertEquals("On message " + i, messageContent.length, receivedBytes.length); while (true) {
assertArrayEquals("On message " + i, messageContent, receivedBytes); receivedBytesMessage.readBoolean();
numElements++;
}
} catch (Exception ex) {
}
LOG.info("Iteration [{}]: Received Message contained {} boolean values.", i, numElements);
assertEquals(i, numElements);
long receivedSendTime = receivedBytesMessage.getLongProperty("sendTime"); long receivedSendTime = receivedBytesMessage.getLongProperty("sendTime");
assertEquals("On message " + i, receivedSendTime, 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);
} }
} }