Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Make producerFlowControl configurable per Destination 

default=true

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@585375 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Robert Davies 2007-10-17 05:05:50 +00:00
parent 4c137cd2bc
commit 0da68b4cb3
8 changed files with 66 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
activemq-core/src/main/java/org/apache/activemq/broker/region/BaseDestination.java Executable file
View File

@ -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;
}
}

4
activemq-core/src/main/java/org/apache/activemq/broker/region/Destination.java
View File

@ -61,4 +61,8 @@ public interface Destination extends Service {
String getName();
MessageStore getMessageStore();
boolean isProducerFlowControl();
void setProducerFlowControl(boolean value);
}

10
activemq-core/src/main/java/org/apache/activemq/broker/region/DestinationFilter.java
View File

@ -121,4 +121,14 @@ public class DestinationFilter implements Destination {
public MessageStore getMessageStore() {
return next.getMessageStore();
}
public boolean isProducerFlowControl() {
return next.isProducerFlowControl();
}
public void setProducerFlowControl(boolean value){
next.setProducerFlowControl(value);
}
}

4
activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java
View File

@ -72,7 +72,7 @@ import org.apache.commons.logging.LogFactory;
*
* @version $Revision: 1.28 $
*/
public class Queue implements Destination, Task {
public class Queue extends BaseDestination implements Task {
final Broker broker;
@ -361,7 +361,7 @@ public class Queue implements Destination, Task {
}
return;
}
if (context.isProducerFlowControl() && memoryUsage.isFull()) {
if (isProducerFlowControl() && context.isProducerFlowControl() && memoryUsage.isFull()) {
if (systemUsage.isSendFailIfNoSpace()) {
throw new javax.jms.ResourceAllocationException("SystemUsage memory limit reached");
}

4
activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java
View File

@ -64,7 +64,7 @@ import org.apache.commons.logging.LogFactory;
*
* @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 @@ public class Topic implements Destination {
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");
}

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

@ -51,6 +51,7 @@ public class PolicyEntry extends DestinationMapEntry {
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 @@ public class PolicyEntry extends DestinationMapEntry {
PendingMessageCursor messages = pendingQueuePolicy.getQueuePendingMessageCursor(queue, tmpStore);
queue.setMessages(messages);
}
queue.setProducerFlowControl(isProducerFlowControl());
}
public void configure(Topic topic) {
@ -83,6 +85,7 @@ public class PolicyEntry extends DestinationMapEntry {
if (memoryLimit > 0) {
topic.getBrokerMemoryUsage().setLimit(memoryLimit);
}
topic.setProducerFlowControl(isProducerFlowControl());
}
public void configure(Broker broker, SystemUsage memoryManager, TopicSubscription subscription) {
@ -263,4 +266,12 @@ public class PolicyEntry extends DestinationMapEntry {
this.pendingSubscriberPolicy = pendingSubscriberPolicy;
}
public boolean isProducerFlowControl() {
return producerFlowControl;
}
public void setProducerFlowControl(boolean producerFlowControl) {
this.producerFlowControl = producerFlowControl;
}
}

2
activemq-core/src/test/java/org/apache/activemq/broker/policy/DestinationCursorConfigTest.java
View File

@ -67,6 +67,8 @@ public class DestinationCursorConfigTest extends TestSupport {
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);
}
}

2
activemq-core/src/test/resources/org/apache/activemq/broker/policy/cursor.xml
View File

@ -26,7 +26,7 @@
<destinationPolicy>
<policyMap>
<policyEntries>
<policyEntry topic="org.apache.>">
<policyEntry topic="org.apache.>" producerFlowControl="false" memoryLimit="1mb">
<dispatchPolicy>
<strictOrderDispatchPolicy />
</dispatchPolicy>