You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2015/10/12 23:04:20 UTC
[4/4] activemq-artemis git commit: ARTEMIS-252 retryMessages retrying
to topic subscriptions + some ammends to #193
ARTEMIS-252 retryMessages retrying to topic subscriptions + some ammends to #193
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/2a81a5f1
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/2a81a5f1
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/2a81a5f1
Branch: refs/heads/master
Commit: 2a81a5f146d6bb8b9dcef352fc68978464db85aa
Parents: 8848c96
Author: Clebert Suconic <cl...@apache.org>
Authored: Mon Oct 12 16:34:23 2015 -0400
Committer: Clebert Suconic <cl...@apache.org>
Committed: Mon Oct 12 17:03:53 2015 -0400
----------------------------------------------------------------------
.../core/management/impl/QueueControlImpl.java | 40 ++++-----
.../activemq/artemis/core/server/Queue.java | 2 +
.../artemis/core/server/impl/QueueImpl.java | 56 +++++++++++-
.../impl/ScheduledDeliveryHandlerTest.java | 5 ++
.../server/management/JMSQueueControlTest.java | 93 +++++++++++++++++---
.../unit/core/postoffice/impl/FakeQueue.java | 5 ++
6 files changed, 161 insertions(+), 40 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/2a81a5f1/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/QueueControlImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/QueueControlImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/QueueControlImpl.java
index 1260169..cb362ec 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/QueueControlImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/QueueControlImpl.java
@@ -21,7 +21,6 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
-import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -37,11 +36,12 @@ import org.apache.activemq.artemis.core.messagecounter.impl.MessageCounterHelper
import org.apache.activemq.artemis.core.persistence.StorageManager;
import org.apache.activemq.artemis.core.postoffice.Binding;
import org.apache.activemq.artemis.core.postoffice.PostOffice;
-import org.apache.activemq.artemis.core.server.Consumer;
import org.apache.activemq.artemis.core.server.ActiveMQMessageBundle;
+import org.apache.activemq.artemis.core.server.Consumer;
import org.apache.activemq.artemis.core.server.MessageReference;
import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.core.server.ServerConsumer;
+import org.apache.activemq.artemis.core.server.ServerMessage;
import org.apache.activemq.artemis.core.settings.HierarchicalRepository;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.utils.LinkedListIterator;
@@ -558,16 +558,19 @@ public class QueueControlImpl extends AbstractControl implements QueueControl {
clearIO();
try {
- MessageReference message = queue.getReference(messageID);
- if ( message == null ) {
- return false;
- }
- else {
- final String originalAddress = message.getMessage().getStringProperty(Message.HDR_ORIGINAL_ADDRESS);
- if (originalAddress != null) {
- return queue.moveReference(messageID, new SimpleString(originalAddress));
+ Filter singleMessageFilter = new Filter() {
+ @Override
+ public boolean match(ServerMessage message) {
+ return message.getMessageID() == messageID;
}
- }
+
+ @Override
+ public SimpleString getFilterString() {
+ return new SimpleString("custom filter for MESSAGEID= messageID");
+ }
+ };
+
+ queue.retryMessages(singleMessageFilter);
}
finally {
blockOnIO();
@@ -580,25 +583,12 @@ public class QueueControlImpl extends AbstractControl implements QueueControl {
checkStarted();
clearIO();
- int retriedMessages = 0;
try {
- Iterator<MessageReference> messageIterator = queue.totalIterator();
- while (messageIterator.hasNext()) {
- MessageReference message = messageIterator.next();
- // Will only try messages with Message.HDR_ORIGINAL_ADDRESS set.
- final String originalAddress = message.getMessage().getStringProperty(Message.HDR_ORIGINAL_ADDRESS);
- final long messageID = message.getMessage().getMessageID();
- if ( originalAddress != null) {
- if ( queue.moveReference(messageID, new SimpleString(originalAddress))) {
- retriedMessages++;
- }
- }
- }
+ return queue.retryMessages(null);
}
finally {
blockOnIO();
}
- return retriedMessages;
}
public boolean moveMessage(final long messageID, final String otherQueueName) throws Exception {
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/2a81a5f1/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java
index ba12a55..f5a19a8 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/Queue.java
@@ -164,6 +164,8 @@ public interface Queue extends Bindable {
SimpleString toAddress,
boolean rejectDuplicates) throws Exception;
+ int retryMessages(Filter filter) throws Exception;
+
void addRedistributor(long delay);
void cancelRedistributor() throws Exception;
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/2a81a5f1/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
index 2ac5c8a..95a7ba6 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
@@ -55,6 +55,7 @@ import org.apache.activemq.artemis.core.postoffice.Binding;
import org.apache.activemq.artemis.core.postoffice.Bindings;
import org.apache.activemq.artemis.core.postoffice.DuplicateIDCache;
import org.apache.activemq.artemis.core.postoffice.PostOffice;
+import org.apache.activemq.artemis.core.postoffice.impl.LocalQueueBinding;
import org.apache.activemq.artemis.core.postoffice.impl.PostOfficeImpl;
import org.apache.activemq.artemis.core.remoting.server.RemotingService;
import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
@@ -1525,6 +1526,50 @@ public class QueueImpl implements Queue {
});
}
+ public int retryMessages(Filter filter) throws Exception {
+
+ final HashMap<SimpleString, Long> queues = new HashMap<>();
+
+ return iterQueue(DEFAULT_FLUSH_LIMIT, null, new QueueIterateAction() {
+ @Override
+ public void actMessage(Transaction tx, MessageReference ref) throws Exception {
+
+ SimpleString originalMessageAddress = ref.getMessage().getSimpleStringProperty(MessageImpl.HDR_ORIGINAL_ADDRESS);
+ SimpleString originalMessageQueue = ref.getMessage().getSimpleStringProperty(MessageImpl.HDR_ORIGINAL_QUEUE);
+
+ if (originalMessageAddress != null) {
+
+ incDelivering();
+
+ Long targetQueue = null;
+ if (originalMessageQueue != null && !originalMessageQueue.equals(originalMessageAddress)) {
+ targetQueue = queues.get(originalMessageQueue);
+ if (targetQueue == null) {
+ Binding binding = postOffice.getBinding(originalMessageQueue);
+
+ if (binding != null && binding instanceof LocalQueueBinding) {
+ targetQueue = ((LocalQueueBinding)binding).getID();
+ queues.put(originalMessageQueue, targetQueue);
+ }
+ }
+ }
+
+ if (targetQueue != null) {
+ move(originalMessageAddress, tx, ref, false, false, targetQueue.longValue());
+ }
+ else {
+ move(originalMessageAddress, tx, ref, false, false);
+
+ }
+
+
+ }
+ }
+ });
+
+
+ }
+
public synchronized boolean changeReferencePriority(final long messageID, final byte newPriority) throws Exception {
LinkedListIterator<MessageReference> iter = iterator();
@@ -2057,11 +2102,20 @@ public class QueueImpl implements Queue {
final Transaction tx,
final MessageReference ref,
final boolean expiry,
- final boolean rejectDuplicate) throws Exception {
+ final boolean rejectDuplicate,
+ final long ... queueIDs) throws Exception {
ServerMessage copyMessage = makeCopy(ref, expiry);
copyMessage.setAddress(toAddress);
+ if (queueIDs != null && queueIDs.length > 0) {
+ ByteBuffer buffer = ByteBuffer.allocate(8 * queueIDs.length);
+ for (long id : queueIDs) {
+ buffer.putLong(id);
+ }
+ copyMessage.putBytesProperty(MessageImpl.HDR_ROUTE_TO_IDS, buffer.array());
+ }
+
postOffice.route(copyMessage, null, tx, false, rejectDuplicate);
acknowledge(tx, ref);
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/2a81a5f1/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerTest.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerTest.java
index d42c89d..b18be4a 100644
--- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerTest.java
+++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerTest.java
@@ -908,6 +908,11 @@ public class ScheduledDeliveryHandlerTest extends Assert {
}
@Override
+ public int retryMessages(Filter filter) throws Exception {
+ return 0;
+ }
+
+ @Override
public int getConsumerCount() {
return 0;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/2a81a5f1/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/server/management/JMSQueueControlTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/server/management/JMSQueueControlTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/server/management/JMSQueueControlTest.java
index b5183ca..ac9b28e 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/server/management/JMSQueueControlTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/server/management/JMSQueueControlTest.java
@@ -16,6 +16,20 @@
*/
package org.apache.activemq.artemis.tests.integration.jms.server.management;
+import javax.jms.Connection;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.management.Notification;
+import javax.naming.Context;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.TransportConfiguration;
import org.apache.activemq.artemis.api.core.client.ClientConsumer;
@@ -39,6 +53,7 @@ import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
import org.apache.activemq.artemis.jms.client.ActiveMQDestination;
import org.apache.activemq.artemis.jms.client.ActiveMQJMSConnectionFactory;
import org.apache.activemq.artemis.jms.client.ActiveMQQueue;
+import org.apache.activemq.artemis.jms.client.ActiveMQTopic;
import org.apache.activemq.artemis.jms.server.impl.JMSServerManagerImpl;
import org.apache.activemq.artemis.jms.server.management.JMSNotificationType;
import org.apache.activemq.artemis.tests.integration.management.ManagementControlHelper;
@@ -51,20 +66,6 @@ import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
-import javax.jms.Connection;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.MessageConsumer;
-import javax.jms.MessageProducer;
-import javax.jms.Session;
-import javax.management.Notification;
-import javax.naming.Context;
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
/**
* A QueueControlTest
* <br>
@@ -808,6 +809,16 @@ public class JMSQueueControlTest extends ManagementTestBase {
return testQueue;
}
+ protected ActiveMQTopic createTestTopicWithDLQ(final String queueName, final ActiveMQQueue dlq) throws Exception {
+ serverManager.createTopic(false, queueName);
+ ActiveMQTopic testQueue = (ActiveMQTopic) ActiveMQJMSClient.createTopic(queueName);
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setDeadLetterAddress(new SimpleString(dlq.getAddress()));
+ addressSettings.setMaxDeliveryAttempts(1);
+ server.getAddressSettingsRepository().addMatch(testQueue.getAddress(), addressSettings);
+ return testQueue;
+ }
+
/**
* Test retrying all messages put on DLQ - i.e. they should appear on the original queue.
* @throws Exception
@@ -834,10 +845,64 @@ public class JMSQueueControlTest extends ManagementTestBase {
Assert.assertEquals(0, getMessageCount(testQueueControl));
Assert.assertEquals(numMessagesToTest,getMessageCount(dlqQueueControl));
+ Assert.assertEquals(10,getMessageCount(dlqQueueControl));
+
dlqQueueControl.retryMessages();
Assert.assertEquals(numMessagesToTest, getMessageCount(testQueueControl));
Assert.assertEquals(0,getMessageCount(dlqQueueControl));
+
+ connection.close();
+ }
+
+ /**
+ * Test retrying all messages put on DLQ - i.e. they should appear on the original queue.
+ * @throws Exception
+ */
+ @Test
+ public void testRetryMessagesOnTopic() throws Exception {
+ ActiveMQQueue dlq = createDLQ(RandomUtil.randomString());
+ ActiveMQTopic testTopic = createTestTopicWithDLQ(RandomUtil.randomString(), dlq);
+
+ Connection connectionConsume = createConnection();
+ connectionConsume.setClientID("ID");
+ Session sessionConsume = connectionConsume.createSession(true, Session.SESSION_TRANSACTED);
+ MessageConsumer cons1 = sessionConsume.createDurableSubscriber(testTopic, "sub1");
+ MessageConsumer cons2 = sessionConsume.createDurableSubscriber(testTopic, "sub2");
+
+
+ final int numMessagesToTest = 10;
+ JMSUtil.sendMessages(testTopic, numMessagesToTest);
+
+
+ connectionConsume.start();
+ for (int i = 0; i < numMessagesToTest; i++) {
+ Assert.assertNotNull(cons1.receive(500));
+ }
+ sessionConsume.commit();
+
+ Assert.assertNull(cons1.receiveNoWait());
+
+ connectionConsume.start();
+ for (int i = 0; i < numMessagesToTest; i++) {
+ cons2.receive(500);
+ }
+ sessionConsume.rollback();
+ Assert.assertNull(cons2.receiveNoWait());
+
+ JMSQueueControl dlqQueueControl = createManagementControl(dlq);
+ dlqQueueControl.retryMessages();
+
+ Assert.assertNull("Retry is sending back to cons1 even though it succeeded", cons1.receiveNoWait());
+
+ for (int i = 0; i < numMessagesToTest; i++) {
+ Assert.assertNotNull(cons2.receive(500));
+ }
+ sessionConsume.commit();
+ Assert.assertNull(cons1.receiveNoWait());
+
+ connectionConsume.close();
+
}
/**
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/2a81a5f1/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/FakeQueue.java
----------------------------------------------------------------------
diff --git a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/FakeQueue.java b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/FakeQueue.java
index 970e89d..4aefc9a 100644
--- a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/FakeQueue.java
+++ b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/postoffice/impl/FakeQueue.java
@@ -50,6 +50,11 @@ public class FakeQueue implements Queue {
}
@Override
+ public int retryMessages(Filter filter) throws Exception {
+ return 0;
+ }
+
+ @Override
public void setConsumersRefCount(ReferenceCounter referenceCounter) {
}