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

Fixed AMQ-5160, restored previous DurableSubscription behaviour of only recovering messages when cursor is empty, retained messages are always recovered

This commit is contained in:
Dhiraj Bokde 2014-05-13 13:15:04 -07:00 committed by Dejan Bosanac
parent 8644090377
commit 42ad1039cb
2 changed files with 14 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
activemq-broker/src/main/java/org/apache/activemq/broker/region/DurableTopicSubscription.java
View File

@ -137,6 +137,11 @@ public class DurableTopicSubscription extends PrefetchSubscription implements Us
dispatchPending();
}
// used by RetaineMessageSubscriptionRecoveryPolicy
public boolean isEmpty(Topic topic) {
return pending.isEmpty(topic);
}
public void activate(SystemUsage memoryManager, ConnectionContext context, ConsumerInfo info, RegionBroker regionBroker) throws Exception {
if (!active.get()) {
this.context = context;

10
activemq-broker/src/main/java/org/apache/activemq/broker/region/policy/RetainedMessageSubscriptionRecoveryPolicy.java
View File

@ -22,6 +22,7 @@ import java.util.List;
import org.apache.activemq.broker.Broker;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.region.DurableTopicSubscription;
import org.apache.activemq.broker.region.MessageReference;
import org.apache.activemq.broker.region.SubscriptionRecovery;
import org.apache.activemq.broker.region.Topic;
@ -74,7 +75,14 @@ public class RetainedMessageSubscriptionRecoveryPolicy implements SubscriptionRe
sub.addRecoveredMessage(context, retainedMessage);
}
if (wrapped != null) {
wrapped.recover(context, topic, sub);
// retain default ActiveMQ behaviour of recovering messages only for empty durable subscriptions
boolean recover = true;
if (sub instanceof DurableTopicSubscription && !((DurableTopicSubscription)sub).isEmpty(topic)) {
recover = false;
}
if (recover) {
wrapped.recover(context, topic, sub);
}
}
}