activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rajdav...@apache.org
Subject svn commit: r585375 - in /activemq/trunk/activemq-core/src: main/java/org/apache/activemq/broker/region/ main/java/org/apache/activemq/broker/region/policy/ test/java/org/apache/activemq/broker/policy/ test/resources/org/apache/activemq/broker/policy/
Date Wed, 17 Oct 2007 05:05:51 GMT
Author: rajdavies
Date: Tue Oct 16 22:05:50 2007
New Revision: 585375

URL: http://svn.apache.org/viewvc?rev=585375&view=rev
Log:
Make producerFlowControl configurable per Destination
default=true

Added:
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/BaseDestination.java
  (with props)
Modified:
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Destination.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/DestinationFilter.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java
    activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/policy/DestinationCursorConfigTest.java
    activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/broker/policy/cursor.xml

Added: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/BaseDestination.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/BaseDestination.java?rev=585375&view=auto
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/BaseDestination.java
(added)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/BaseDestination.java
Tue Oct 16 22:05:50 2007
@@ -0,0 +1,34 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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;
+
+
+/**
+ * @version $Revision: 1.12 $
+ */
+public abstract class BaseDestination implements Destination {
+
+    private boolean producerFlowControl = true;
+
+    public boolean isProducerFlowControl() {
+        return this.producerFlowControl;
+    }
+
+    public void setProducerFlowControl(boolean value) {
+        this.producerFlowControl = value;
+    }
+}

Propchange: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/BaseDestination.java
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/BaseDestination.java
------------------------------------------------------------------------------
    svn:executable = *

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Destination.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Destination.java?rev=585375&r1=585374&r2=585375&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Destination.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Destination.java
Tue Oct 16 22:05:50 2007
@@ -61,4 +61,8 @@
     String getName();
 
     MessageStore getMessageStore();
+    
+    boolean isProducerFlowControl();
+    
+    void setProducerFlowControl(boolean value);
 }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/DestinationFilter.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/DestinationFilter.java?rev=585375&r1=585374&r2=585375&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/DestinationFilter.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/DestinationFilter.java
Tue Oct 16 22:05:50 2007
@@ -121,4 +121,14 @@
     public MessageStore getMessageStore() {
         return next.getMessageStore();
     }
+
+    public boolean isProducerFlowControl() {
+        return next.isProducerFlowControl();
+    }
+    
+    public void setProducerFlowControl(boolean value){
+        next.setProducerFlowControl(value);
+    }
+    
+    
 }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java?rev=585375&r1=585374&r2=585375&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java
Tue Oct 16 22:05:50 2007
@@ -72,7 +72,7 @@
  * 
  * @version $Revision: 1.28 $
  */
-public class Queue implements Destination, Task {
+public class Queue extends BaseDestination implements Task {
 
     final Broker broker;
 
@@ -361,7 +361,7 @@
             }
             return;
         }
-        if (context.isProducerFlowControl() && memoryUsage.isFull()) {
+        if (isProducerFlowControl() && context.isProducerFlowControl() &&
memoryUsage.isFull()) {
             if (systemUsage.isSendFailIfNoSpace()) {
                 throw new javax.jms.ResourceAllocationException("SystemUsage memory limit
reached");
             }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java?rev=585375&r1=585374&r2=585375&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java
Tue Oct 16 22:05:50 2007
@@ -64,7 +64,7 @@
  * 
  * @version $Revision: 1.21 $
  */
-public class Topic implements Destination {
+public class Topic  extends BaseDestination {
     private static final Log LOG = LogFactory.getLog(Topic.class);
     protected final ActiveMQDestination destination;
     protected final CopyOnWriteArrayList<Subscription> consumers = new CopyOnWriteArrayList<Subscription>();
@@ -296,7 +296,7 @@
             return;
         }
 
-        if (context.isProducerFlowControl() && memoryUsage.isFull()) {
+        if (isProducerFlowControl() && context.isProducerFlowControl() &&
memoryUsage.isFull()) {
             if (systemUsage.isSendFailIfNoSpace()) {
                 throw new javax.jms.ResourceAllocationException("Usage Manager memory limit
reached");
             }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java?rev=585375&r1=585374&r2=585375&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java
Tue Oct 16 22:05:50 2007
@@ -51,6 +51,7 @@
     private PendingQueueMessageStoragePolicy pendingQueuePolicy;
     private PendingDurableSubscriberMessageStoragePolicy pendingDurableSubscriberPolicy;
     private PendingSubscriberMessageStoragePolicy pendingSubscriberPolicy;
+    private boolean producerFlowControl = true;
 
     public void configure(Queue queue, Store tmpStore) {
         if (dispatchPolicy != null) {
@@ -67,6 +68,7 @@
             PendingMessageCursor messages = pendingQueuePolicy.getQueuePendingMessageCursor(queue,
tmpStore);
             queue.setMessages(messages);
         }
+        queue.setProducerFlowControl(isProducerFlowControl());
     }
 
     public void configure(Topic topic) {
@@ -83,6 +85,7 @@
         if (memoryLimit > 0) {
             topic.getBrokerMemoryUsage().setLimit(memoryLimit);
         }
+        topic.setProducerFlowControl(isProducerFlowControl());
     }
 
     public void configure(Broker broker, SystemUsage memoryManager, TopicSubscription subscription)
{
@@ -261,6 +264,14 @@
      */
     public void setPendingSubscriberPolicy(PendingSubscriberMessageStoragePolicy pendingSubscriberPolicy)
{
         this.pendingSubscriberPolicy = pendingSubscriberPolicy;
+    }
+
+    public boolean isProducerFlowControl() {
+        return producerFlowControl;
+    }
+
+    public void setProducerFlowControl(boolean producerFlowControl) {
+        this.producerFlowControl = producerFlowControl;
     }
 
 }

Modified: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/policy/DestinationCursorConfigTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/policy/DestinationCursorConfigTest.java?rev=585375&r1=585374&r2=585375&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/policy/DestinationCursorConfigTest.java
(original)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/policy/DestinationCursorConfigTest.java
Tue Oct 16 22:05:50 2007
@@ -67,6 +67,8 @@
         PolicyEntry entry = broker.getDestinationPolicy().getEntryFor(destination);
         PendingSubscriberMessageStoragePolicy policy = entry.getPendingSubscriberPolicy();
         assertNotNull(policy);
+        assertFalse(entry.isProducerFlowControl());
+        assertTrue(entry.getMemoryLimit()==(1024*1024));
         assertTrue("subscriberPolicy is: " + policy, policy instanceof VMPendingSubscriberMessageStoragePolicy);
     }
 }

Modified: activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/broker/policy/cursor.xml
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/broker/policy/cursor.xml?rev=585375&r1=585374&r2=585375&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/broker/policy/cursor.xml
(original)
+++ activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/broker/policy/cursor.xml
Tue Oct 16 22:05:50 2007
@@ -26,7 +26,7 @@
     <destinationPolicy>
       <policyMap>
         <policyEntries>
-          <policyEntry topic="org.apache.>">
+          <policyEntry topic="org.apache.>" producerFlowControl="false" memoryLimit="1mb">
             <dispatchPolicy>
               <strictOrderDispatchPolicy />
             </dispatchPolicy>



Mime
View raw message