From 0d5de13537dad45a044668948d5c14325b867716 Mon Sep 17 00:00:00 2001
From: James Strachan <jstrachan@apache.org>
Date: Thu, 2 Mar 2006 15:56:30 +0000
Subject: [PATCH] added the ability to discard old messages for non-durable
 topics if a maximum number of pending messages is reached

git-svn-id: https://svn.apache.org/repos/asf/incubator/activemq/trunk@382421 13f79535-47bb-0310-9956-ffa450edef68
---
 .../broker/region/TopicSubscription.java      | 28 +++++++++++++++++--
 1 file changed, 25 insertions(+), 3 deletions(-)

diff --git a/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java b/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java
index a9b860bf69..78a6792576 100755
--- a/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java
+++ b/activemq-core/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java
@@ -42,6 +42,7 @@ public class TopicSubscription extends AbstractSubscription {
     final protected UsageManager usageManager;
     protected int dispatched=0;
     protected int delivered=0;
+    private int maximumPendingMessages = 0;
     
     public TopicSubscription(Broker broker,ConnectionContext context, ConsumerInfo info, UsageManager usageManager) throws InvalidSelectorException {
         super(broker,context, info);
@@ -50,11 +51,20 @@ public class TopicSubscription extends AbstractSubscription {
 
     public void add(MessageReference node) throws InterruptedException, IOException {
         node.incrementReferenceCount();
-        if( !isFull() && !isSlaveBroker() ) {
+        if( !isFull() && !isSlaveBroker()) {
+            // TODO - if we have already dispatched too many messages to this slow consumer
+            // should we avoid dispatching and just discard old messages as shown below
             dispatch(node);
         } else {
-            synchronized(matched){
-            matched.addLast(node);
+            synchronized (matched) {
+                matched.addLast(node);
+                if (maximumPendingMessages > 0) {
+                    // lets discard old messages as we are a slow consumer
+                    while (matched.size() > maximumPendingMessages) {
+                        MessageReference oldMessage = (MessageReference) matched.removeFirst();
+                        oldMessage.decrementReferenceCount();
+                    }
+                }
             }
         }        
     }
@@ -122,6 +132,18 @@ public class TopicSubscription extends AbstractSubscription {
         return delivered;
     }
     
+    public int getMaximumPendingMessages() {
+        return maximumPendingMessages;
+    }
+
+    /**
+     * Sets the maximum number of pending messages that can be matched against this consumer
+     * before old messages are discarded.
+     */
+    public void setMaximumPendingMessages(int maximumPendingMessages) {
+        this.maximumPendingMessages = maximumPendingMessages;
+    }
+
     private boolean isFull() {
         return dispatched-delivered >= info.getPrefetchSize();
     }