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 consumers = new CopyOnWriteArrayList(); @@ -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 @@ - +