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

This closes #3419

This commit is contained in:
Clebert Suconic 2021-01-27 20:41:33 -05:00
parent 0845ff2353 88b21f9039
commit 2cf8d5c181
2 changed files with 32 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
View File

@ -3919,12 +3919,16 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
return;
}
// if the queue is non-destructive then any ack is ignored so no need to add messages back onto the queue
if (!isNonDestructive()) {
if (sorted) {
addSorted(refs, false);
} else {
addHead(refs, false);
}
}
}
private void purgeAfterRollback(LinkedList<MessageReference> refs) {
try {

25
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/JMSNonDestructiveTest.java
View File

@ -16,7 +16,6 @@
*/
package org.apache.activemq.artemis.tests.integration.amqp;
import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
@ -329,6 +328,30 @@ public class JMSNonDestructiveTest extends JMSClientTestSupport {
}
@Test
public void testMessageCount() throws Exception {
sendMessage(CoreConnection, NON_DESTRUCTIVE_QUEUE_NAME);
QueueBinding queueBinding = (QueueBinding) server.getPostOffice().getBinding(SimpleString.toSimpleString(NON_DESTRUCTIVE_QUEUE_NAME));
assertEquals("Ensure Message count", 1, queueBinding.getQueue().getMessageCount());
//Consume Once
receive(CoreConnection, NON_DESTRUCTIVE_QUEUE_NAME);
assertEquals("Ensure Message count", 1, queueBinding.getQueue().getMessageCount());
sendMessage(CoreConnection, NON_DESTRUCTIVE_QUEUE_NAME);
assertEquals("Ensure Message count", 2, queueBinding.getQueue().getMessageCount());
//Consume Again as should be non-destructive
receive(CoreConnection, NON_DESTRUCTIVE_QUEUE_NAME);
assertEquals("Ensure Message count", 2, queueBinding.getQueue().getMessageCount());
QueueControl control = (QueueControl) server.getManagementService().getResource(ResourceNames.QUEUE + NON_DESTRUCTIVE_QUEUE_NAME);
control.removeAllMessages();
assertEquals("Message count after clearing queue via queue control should be 0", 0, queueBinding.getQueue().getMessageCount());
}
private void receive(ConnectionSupplier consumerConnectionSupplier, String queueName, int i) throws JMSException {
try (Connection consumerConnection = consumerConnectionSupplier.createConnection()) {