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

Add a rollback test that rolls back repeatedly using new connections.

This commit is contained in:
Timothy Bish 2016-06-22 13:02:14 -04:00
parent a95d6fd4eb
commit 94ca7039b1
1 changed files with 57 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

60
activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientTransactionTest.java
View File

@ -22,6 +22,7 @@ import static org.junit.Assert.assertTrue;
import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicInteger;
import javax.jms.Connection;
import javax.jms.DeliveryMode; import javax.jms.DeliveryMode;
import javax.jms.Destination; import javax.jms.Destination;
import javax.jms.JMSException; import javax.jms.JMSException;
@ -107,7 +108,7 @@ public class JMSClientTransactionTest extends JMSClientTestSupport {
TextMessage message = (TextMessage) messageConsumer.receive(1000); TextMessage message = (TextMessage) messageConsumer.receive(1000);
if (message != null) { if (message != null) {
counter++; counter++;
LOG.info("Message n. {} with content '{}' has been recieved.", counter,message.getText()); LOG.info("Message n. {} with content '{}' has been recieved.", counter, message.getText());
session.commit(); session.commit();
LOG.info("Transaction has been committed."); LOG.info("Transaction has been committed.");
} }
@ -144,7 +145,7 @@ public class JMSClientTransactionTest extends JMSClientTestSupport {
TextMessage message = (TextMessage) messageConsumer.receive(1000); TextMessage message = (TextMessage) messageConsumer.receive(1000);
if (message != null) { if (message != null) {
counter++; counter++;
LOG.info("Message n. {} with content '{}' has been recieved.", counter,message.getText()); LOG.info("Message n. {} with content '{}' has been recieved.", counter, message.getText());
} }
} while (counter < MSG_COUNT); } while (counter < MSG_COUNT);
@ -199,7 +200,7 @@ public class JMSClientTransactionTest extends JMSClientTestSupport {
LOG.info("Trying to receive message: {}", i); LOG.info("Trying to receive message: {}", i);
TextMessage message = (TextMessage) consumer.receive(1000); TextMessage message = (TextMessage) consumer.receive(1000);
assertNotNull("Message " + i + " should be available", message); assertNotNull("Message " + i + " should be available", message);
assertEquals("Should get message: " + i, i , message.getIntProperty("MessageSequence")); assertEquals("Should get message: " + i, i, message.getIntProperty("MessageSequence"));
} }
session.commit(); session.commit();
@ -295,4 +296,57 @@ public class JMSClientTransactionTest extends JMSClientTestSupport {
} }
})); }));
} }
@Test
public void testMessageOrderAfterRollback() throws Exception {
sendMessages(5);
int counter = 0;
while (counter++ < 10) {
connection = createConnection();
connection.start();
Session session = connection.createSession(true, -1);
Queue queue = session.createQueue(getDestinationName());
MessageConsumer consumer = session.createConsumer(queue);
Message message = consumer.receive(5000);
assertNotNull(message);
assertTrue(message instanceof TextMessage);
int sequenceID = message.getIntProperty("sequenceID");
assertEquals(0, sequenceID);
LOG.info("Read message = {}", ((TextMessage) message).getText());
session.rollback();
session.close();
connection.close();
}
}
public void sendMessages(int messageCount) throws JMSException {
Connection connection = null;
try {
connection = createConnection();
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Queue queue = session.createQueue(getDestinationName());
for (int i = 0; i < messageCount; ++i) {
MessageProducer messageProducer = session.createProducer(queue);
TextMessage message = session.createTextMessage("(" + i + ")");
message.setIntProperty("sequenceID", i);
messageProducer.send(message);
LOG.info("Sent message = {}", message.getText());
}
} catch (Exception exp) {
exp.printStackTrace(System.out);
} finally {
if (connection != null) {
connection.close();
}
}
}
} }