Return-Path: Delivered-To: apmail-activemq-commits-archive@www.apache.org Received: (qmail 2391 invoked from network); 17 Oct 2007 05:06:14 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (140.211.11.2) by minotaur.apache.org with SMTP; 17 Oct 2007 05:06:14 -0000 Received: (qmail 76922 invoked by uid 500); 17 Oct 2007 05:06:02 -0000 Delivered-To: apmail-activemq-commits-archive@activemq.apache.org Received: (qmail 76893 invoked by uid 500); 17 Oct 2007 05:06:01 -0000 Mailing-List: contact commits-help@activemq.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: dev@activemq.apache.org Delivered-To: mailing list commits@activemq.apache.org Received: (qmail 76881 invoked by uid 99); 17 Oct 2007 05:06:01 -0000 Received: from athena.apache.org (HELO athena.apache.org) (140.211.11.136) by apache.org (qpsmtpd/0.29) with ESMTP; Tue, 16 Oct 2007 22:06:01 -0700 X-ASF-Spam-Status: No, hits=-100.0 required=10.0 tests=ALL_TRUSTED X-Spam-Check-By: apache.org Received: from [140.211.11.3] (HELO eris.apache.org) (140.211.11.3) by apache.org (qpsmtpd/0.29) with ESMTP; Wed, 17 Oct 2007 05:06:12 +0000 Received: by eris.apache.org (Postfix, from userid 65534) id B9D981A9838; Tue, 16 Oct 2007 22:05:52 -0700 (PDT) Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit 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 -0000 To: commits@activemq.apache.org From: rajdavies@apache.org X-Mailer: svnmailer-1.0.8 Message-Id: <20071017050552.B9D981A9838@eris.apache.org> X-Virus-Checked: Checked by ClamAV on apache.org 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 @@ - +