added better MBeans for topic subscriptions so that they can see the discarded and matched statistics. Also added a configurable MessageEvictionStrategy

git-svn-id: https://svn.apache.org/repos/asf/incubator/activemq/trunk@382816 13f79535-47bb-0310-9956-ffa450edef68

activemq-core/src/main/java/org/apache/activemq/broker/jmx/ManagedRegionBroker.java

@@ -43,6 +43,7 @@ import org.apache.activemq.broker.region.Region;
 import org.apache.activemq.broker.region.RegionBroker;
 import org.apache.activemq.broker.region.Subscription;
 import org.apache.activemq.broker.region.Topic;
+import org.apache.activemq.broker.region.TopicSubscription;
 import org.apache.activemq.broker.region.policy.PolicyMap;
 import org.apache.activemq.command.ActiveMQDestination;
 import org.apache.activemq.command.ActiveMQMessage;
@@ -159,7 +160,12 @@ public class ManagedRegionBroker extends RegionBroker{
             if(sub.getConsumerInfo().isDurable()){
                 view=new DurableSubscriptionView(this,context.getClientId(),sub);
             }else{
-                view=new SubscriptionView(context.getClientId(),sub);
+                if (sub instanceof TopicSubscription) {
+                    view = new TopicSubscriptionView(context.getClientId(),(TopicSubscription) sub);
+                }
+                else {
+                    view=new SubscriptionView(context.getClientId(),sub);
+                }
             }
             subscriptionMap.put(sub,objectName);
             registerSubscription(objectName,sub.getConsumerInfo(),key,view);

activemq-core/src/main/java/org/apache/activemq/broker/jmx/TopicSubscriptionView.java

@@ -0,0 +1,52 @@
+/**
+ *
+ * Copyright 2005-2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.broker.jmx;
+
+import org.apache.activemq.broker.region.TopicSubscription;
+
+/**
+ * 
+ * @version $Revision$
+ */
+public class TopicSubscriptionView extends SubscriptionView implements TopicSubscriptionViewMBean {
+
+    public TopicSubscriptionView(String clientId, TopicSubscription subs) {
+        super(clientId, subs);
+    }
+
+    protected TopicSubscription getTopicSubscription() {
+        return (TopicSubscription) subscription;
+    }
+
+    /**
+     * @return the number of messages discarded due to being a slow consumer
+     */
+    public int getDiscarded() {
+        TopicSubscription topicSubscription = getTopicSubscription();
+        return topicSubscription != null ? topicSubscription.discarded() : 0;
+    }
+
+    /**
+     * @return the number of matched messages (messages targeted for the
+     *         subscription but not yet able to be dispatched due to the
+     *         prefetch buffer being full).
+     */
+    public int getMatched() {
+        TopicSubscription topicSubscription = getTopicSubscription();
+        return topicSubscription != null ? topicSubscription.matched() : 0;
+    }
+}

activemq-core/src/main/java/org/apache/activemq/broker/jmx/TopicSubscriptionViewMBean.java

@@ -0,0 +1,36 @@
+/**
+ *
+ * Copyright 2005-2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.broker.jmx;
+
+/**
+ *
+ * @version $Revision$
+ */
+public interface TopicSubscriptionViewMBean extends SubscriptionViewMBean {
+
+    /**
+     * @return the number of messages discarded due to being a slow consumer
+     */
+    public int getDiscarded();
+    
+    /**
+     * @return the number of matched messages (messages targeted for the subscription but not
+     * yet able to be dispatched due to the prefetch buffer being full).
+     */
+    public int getMatched();
+
+}

activemq-core/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java

@@ -20,6 +20,8 @@ import javax.jms.InvalidSelectorException;
 import javax.jms.JMSException;
 import org.apache.activemq.broker.Broker;
 import org.apache.activemq.broker.ConnectionContext;
+import org.apache.activemq.broker.region.policy.MessageEvictionStrategy;
+import org.apache.activemq.broker.region.policy.OldestMessageEvictionStrategy;
 import org.apache.activemq.command.ActiveMQDestination;
 import org.apache.activemq.command.ActiveMQQueue;
 import org.apache.activemq.command.ConsumerInfo;
@@ -42,6 +44,8 @@ public class TopicSubscription extends AbstractSubscription{
     protected AtomicInteger dispatched=new AtomicInteger();
     protected AtomicInteger delivered=new AtomicInteger();
     private int maximumPendingMessages=-1;
+    private MessageEvictionStrategy messageEvictionStrategy = new OldestMessageEvictionStrategy();
+    private int discarded = 0;
     private final Object matchedListMutex=new Object();
 
     public TopicSubscription(Broker broker,ConnectionContext context,ConsumerInfo info,UsageManager usageManager)
@@ -64,8 +68,9 @@ public class TopicSubscription extends AbstractSubscription{
                     if(maximumPendingMessages>0){
                         // lets discard old messages as we are a slow consumer
                         while(!matched.isEmpty()&&matched.size()>maximumPendingMessages){
-                            MessageReference oldMessage=(MessageReference) matched.removeFirst();
+                            MessageReference oldMessage=messageEvictionStrategy.evictMessage(matched);
                             oldMessage.decrementReferenceCount();
+                            discarded++;
                             if (log.isDebugEnabled()){
                                 log.debug("Discarding message " + oldMessage);
                             }
@@ -122,7 +127,7 @@ public class TopicSubscription extends AbstractSubscription{
     }
 
     public int pending(){
-        return matched.size()-dispatched.get();
+        return matched()-dispatched();
     }
 
     public int dispatched(){
@@ -137,6 +142,26 @@ public class TopicSubscription extends AbstractSubscription{
         return maximumPendingMessages;
     }
 
+    /**
+     * @return the number of messages discarded due to being a slow consumer
+     */
+    public int discarded() {
+        synchronized(matchedListMutex) {
+            return discarded;
+        }
+    }
+
+    /**
+     * @return the number of matched messages (messages targeted for the subscription but not
+     * yet able to be dispatched due to the prefetch buffer being full).
+     */
+    public int matched() {
+        synchronized(matchedListMutex) {
+            return matched.size();
+        }
+    }
+
+
     /**
      * Sets the maximum number of pending messages that can be matched against this consumer before old messages are
      * discarded.
@@ -145,6 +170,22 @@ public class TopicSubscription extends AbstractSubscription{
         this.maximumPendingMessages=maximumPendingMessages;
     }
 
+    public MessageEvictionStrategy getMessageEvictionStrategy() {
+        return messageEvictionStrategy;
+    }
+
+    /**
+     * Sets the eviction strategy used to decide which message to evict when the slow consumer
+     * needs to discard messages
+     */
+    public void setMessageEvictionStrategy(MessageEvictionStrategy messageEvictionStrategy) {
+        this.messageEvictionStrategy = messageEvictionStrategy;
+    }
+
+    
+    // Implementation methods
+    // -------------------------------------------------------------------------
+
     private boolean isFull(){
         return dispatched.get()-delivered.get()>=info.getPrefetchSize();
     }
@@ -182,6 +223,6 @@ public class TopicSubscription extends AbstractSubscription{
 
     public String toString(){
         return "TopicSubscription:"+" consumer="+info.getConsumerId()+", destinations="+destinations.size()
-                        +", dispatched="+dispatched+", delivered="+this.delivered+", matched="+this.matched.size();
+                        +", dispatched="+dispatched()+", delivered="+delivered()+", matched="+matched()+", discarded="+discarded();
     }
 }

activemq-core/src/main/java/org/apache/activemq/broker/region/policy/MessageEvictionStrategy.java

@@ -0,0 +1,40 @@
+/**
+ *
+ * Copyright 2005-2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.broker.region.policy;
+
+import org.apache.activemq.broker.region.MessageReference;
+
+import java.io.IOException;
+import java.util.LinkedList;
+
+/**
+ * A strategy for evicting messages from slow consumers.
+ * 
+ * @version $Revision$
+ */
+public interface MessageEvictionStrategy {
+
+    /**
+     * Find the message reference in the given list with oldest messages at the front and newer messages at the end
+     * 
+     * @return the message that has been evicted.
+     * @throws IOException if an exception occurs such as reading a message content (but should not ever happen
+     * as usually all the messages will be in RAM when this method is called).
+     */
+    MessageReference evictMessage(LinkedList messages) throws IOException;
+
+}

activemq-core/src/main/java/org/apache/activemq/broker/region/policy/OldestMessageEvictionStrategy.java

@@ -0,0 +1,36 @@
+/**
+ *
+ * Copyright 2005-2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.broker.region.policy;
+
+import org.apache.activemq.broker.region.MessageReference;
+
+import java.util.LinkedList;
+
+/**
+ * An eviction strategy which evicts the oldest message first (which is the
+ * default).
+ * 
+ * @org.apache.xbean.XBean
+ * 
+ * @version $Revision$
+ */
+public class OldestMessageEvictionStrategy implements MessageEvictionStrategy {
+
+    public MessageReference evictMessage(LinkedList messages) {
+        return (MessageReference) messages.removeFirst();
+    }
+}

activemq-core/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java

@@ -41,6 +41,7 @@ public class PolicyEntry extends DestinationMapEntry {
     private DeadLetterStrategy deadLetterStrategy;
     private int messageGroupHashBucketCount = 1024;
     private PendingMessageLimitStrategy pendingMessageLimitStrategy;
+    private MessageEvictionStrategy messageEvictionStrategy;
 
     public void configure(Queue queue) {
         if (dispatchPolicy != null) {
@@ -81,6 +82,9 @@ public class PolicyEntry extends DestinationMapEntry {
                 subscription.setMaximumPendingMessages(value);
             }
         }
+        if (messageEvictionStrategy != null) {
+            subscription.setMessageEvictionStrategy(messageEvictionStrategy);
+        }
     }
 
     // Properties
@@ -156,4 +160,17 @@ public class PolicyEntry extends DestinationMapEntry {
     public void setPendingMessageLimitStrategy(PendingMessageLimitStrategy pendingMessageLimitStrategy) {
         this.pendingMessageLimitStrategy = pendingMessageLimitStrategy;
     }
+
+    public MessageEvictionStrategy getMessageEvictionStrategy() {
+        return messageEvictionStrategy;
+    }
+
+    /**
+     * Sets the eviction strategy used to decide which message to evict when the
+     * slow consumer needs to discard messages
+     */
+    public void setMessageEvictionStrategy(MessageEvictionStrategy messageEvictionStrategy) {
+        this.messageEvictionStrategy = messageEvictionStrategy;
+    }
+
 }