You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ra...@apache.org on 2007/10/17 07:05:51 UTC
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/
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>