Author: davsclaus
Date: Sat Apr 16 11:45:55 2011
New Revision: 1093970
URL: http://svn.apache.org/viewvc?rev=1093970&view=rev
Log:
CAMEL-1669: Throttler now uses an expression to set per period, like the other
EIP patterns. Thanks to Ben O'Day for the patch.
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedThrottler.java
camel/trunk/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinition.java
camel/trunk/camel-core/src/main/java/org/apache/camel/model/ThrottleDefinition.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/Throttler.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/ThrottlerTest.java
camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/throttler.xml
camel/trunk/examples/camel-example-management/src/main/resources/META-INF/spring/camel-context.xml
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedThrottler.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedThrottler.java?rev=1093970&r1=1093969&r2=1093970&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedThrottler.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedThrottler.java
Sat Apr 16 11:45:55 2011
@@ -22,6 +22,8 @@ import org.apache.camel.processor.Thrott
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedResource;
+import static org.apache.camel.builder.Builder.constant;
+
/**
* @version
*/
@@ -40,12 +42,12 @@ public class ManagedThrottler extends Ma
@ManagedAttribute(description = "Maximum requires per period")
public long getMaximumRequestsPerPeriod() {
- return getThrottler().getMaximumRequestsPerPeriod();
+ return getThrottler().getCurrentMaximumRequestsPerPeriod();
}
@ManagedAttribute(description = "Maximum requires per period")
public void setMaximumRequestsPerPeriod(long maximumRequestsPerPeriod) {
- getThrottler().setMaximumRequestsPerPeriod(maximumRequestsPerPeriod);
+
getThrottler().setMaximumRequestsPerPeriodExpression(constant(maximumRequestsPerPeriod));
}
@ManagedAttribute(description = "Time period in millis")
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinition.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinition.java?rev=1093970&r1=1093969&r2=1093970&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinition.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/model/ProcessorDefinition.java
Sat Apr 16 11:45:55 2011
@@ -1805,11 +1805,26 @@ public abstract class ProcessorDefinitio
* @return the builder
*/
public ThrottleDefinition throttle(long maximumRequestCount) {
+ return
throttle(ExpressionBuilder.constantExpression(maximumRequestCount));
+ }
+
+ /**
+ * <a href="http://camel.apache.org/throttler.html";>Throttler EIP:</a>
+ * Creates a throttler allowing you to ensure that a specific endpoint
does not get overloaded,
+ * or that we don't exceed an agreed SLA with some external service.
+ * <p/>
+ * Will default use a time period of 1 second, so setting the
maximumRequestCount to eg 10
+ * will default ensure at most 10 messages per second.
+ *
+ * @param maximumRequestCount an expression to calculate the maximum
request count
+ * @return the builder
+ */
+ public ThrottleDefinition throttle(Expression maximumRequestCount) {
ThrottleDefinition answer = new
ThrottleDefinition(maximumRequestCount);
addOutput(answer);
return answer;
}
-
+
/**
* <a href="http://camel.apache.org/loop.html";>Loop EIP:</a>
* Creates a loop allowing to process the a message a number of times and
possibly process them
@@ -2994,5 +3009,5 @@ public abstract class ProcessorDefinitio
public String getLabel() {
return "";
}
-
+
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/model/ThrottleDefinition.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/model/ThrottleDefinition.java?rev=1093970&r1=1093969&r2=1093970&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/model/ThrottleDefinition.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/model/ThrottleDefinition.java
Sat Apr 16 11:45:55 2011
@@ -18,15 +18,20 @@ package org.apache.camel.model;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
+
import javax.xml.bind.annotation.XmlAccessType;
import javax.xml.bind.annotation.XmlAccessorType;
import javax.xml.bind.annotation.XmlAttribute;
import javax.xml.bind.annotation.XmlRootElement;
import javax.xml.bind.annotation.XmlTransient;
+import org.apache.camel.Expression;
import org.apache.camel.Processor;
+import org.apache.camel.builder.ExpressionBuilder;
+import org.apache.camel.model.language.ExpressionDefinition;
import org.apache.camel.processor.Throttler;
import org.apache.camel.spi.RouteContext;
+import org.apache.camel.util.ObjectHelper;
import org.apache.camel.util.concurrent.ExecutorServiceHelper;
/**
@@ -36,33 +41,30 @@ import org.apache.camel.util.concurrent.
*/
@XmlRootElement(name = "throttle")
@XmlAccessorType(XmlAccessType.FIELD)
-public class ThrottleDefinition extends OutputDefinition<ThrottleDefinition>
implements ExecutorServiceAwareDefinition<ThrottleDefinition> {
-
- // TODO: Camel 3.0 Should extend NoOutputDefinition
+public class ThrottleDefinition extends ExpressionNode implements
ExecutorServiceAwareDefinition<ThrottleDefinition> {
+ // TODO: Camel 3.0 Should not support outputs
@XmlTransient
private ExecutorService executorService;
@XmlAttribute
private String executorServiceRef;
- @XmlAttribute(required = true)
- private Long maximumRequestsPerPeriod;
@XmlAttribute
private Long timePeriodMillis;
@XmlAttribute
private Boolean asyncDelayed;
@XmlAttribute
private Boolean callerRunsWhenRejected;
-
+
public ThrottleDefinition() {
}
- public ThrottleDefinition(long maximumRequestsPerPeriod) {
- this.maximumRequestsPerPeriod = maximumRequestsPerPeriod;
+ public ThrottleDefinition(Expression maximumRequestsPerPeriod) {
+ super(maximumRequestsPerPeriod);
}
@Override
public String toString() {
- return "Throttle[" + getMaximumRequestsPerPeriod() + " request per " +
getTimePeriodMillis()
+ return "Throttle[" + getExpression() + " request per " +
getTimePeriodMillis()
+ " millis -> " + getOutputs() + "]";
}
@@ -73,7 +75,7 @@ public class ThrottleDefinition extends
@Override
public String getLabel() {
- return "" + getMaximumRequestsPerPeriod() + " per " +
getTimePeriodMillis() + " (ms)";
+ return "" + getExpression() + " per " + getTimePeriodMillis() + "
(ms)";
}
@Override
@@ -90,10 +92,14 @@ public class ThrottleDefinition extends
// should be default 1000 millis
long period = getTimePeriodMillis() != null ? getTimePeriodMillis() :
1000L;
- Throttler answer = new Throttler(childProcessor,
getMaximumRequestsPerPeriod(), period, scheduled);
+ Expression maxRequestsExpression =
createMaxRequestsPerPeriodExpression(routeContext);
+
+ Throttler answer = new Throttler(childProcessor,
maxRequestsExpression, period, scheduled);
+
if (getAsyncDelayed() != null) {
answer.setAsyncDelayed(getAsyncDelayed());
}
+
if (getCallerRunsWhenRejected() == null) {
// should be true by default
answer.setCallerRunsWhenRejected(true);
@@ -103,9 +109,17 @@ public class ThrottleDefinition extends
return answer;
}
+ private Expression createMaxRequestsPerPeriodExpression(RouteContext
routeContext) {
+ if (getExpression() != null) {
+ if (ObjectHelper.isNotEmpty(getExpression().getExpression()) ||
getExpression().getExpressionValue() != null) {
+ return getExpression().createExpression(routeContext);
+ }
+ }
+ return null;
+ }
+
// Fluent API
//
-------------------------------------------------------------------------
-
/**
* Sets the time period during which the maximum request count is valid for
*
@@ -124,7 +138,7 @@ public class ThrottleDefinition extends
* @return the builder
*/
public ThrottleDefinition maximumRequestsPerPeriod(Long
maximumRequestsPerPeriod) {
- setMaximumRequestsPerPeriod(maximumRequestsPerPeriod);
+ setExpression(new
ExpressionDefinition(ExpressionBuilder.constantExpression(maximumRequestsPerPeriod)));
return this;
}
@@ -164,14 +178,6 @@ public class ThrottleDefinition extends
// Properties
//
-------------------------------------------------------------------------
- public Long getMaximumRequestsPerPeriod() {
- return maximumRequestsPerPeriod;
- }
-
- public void setMaximumRequestsPerPeriod(Long maximumRequestsPerPeriod) {
- this.maximumRequestsPerPeriod = maximumRequestsPerPeriod;
- }
-
public Long getTimePeriodMillis() {
return timePeriodMillis;
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/Throttler.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/Throttler.java?rev=1093970&r1=1093969&r2=1093970&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/Throttler.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/Throttler.java
Sat Apr 16 11:45:55 2011
@@ -19,7 +19,9 @@ package org.apache.camel.processor;
import java.util.concurrent.ScheduledExecutorService;
import org.apache.camel.Exchange;
+import org.apache.camel.Expression;
import org.apache.camel.Processor;
+import org.apache.camel.util.ObjectHelper;
/**
* A <a href="http://camel.apache.org/throttler.html";>Throttler</a>
@@ -34,50 +36,58 @@ import org.apache.camel.Processor;
*/
public class Throttler extends DelayProcessorSupport implements Traceable {
private long maximumRequestsPerPeriod;
- private long timePeriodMillis;
+ private Expression maxRequestsPerPeriodExpression;
+ private long timePeriodMillis = 1000;
private volatile TimeSlot slot;
- public Throttler(Processor processor, long maximumRequestsPerPeriod) {
- this(processor, maximumRequestsPerPeriod, 1000, null);
- }
-
- public Throttler(Processor processor, long maximumRequestsPerPeriod, long
timePeriodMillis, ScheduledExecutorService executorService) {
+ public Throttler(Processor processor, Expression
maxRequestsPerPeriodExpression, long timePeriodMillis, ScheduledExecutorService
executorService) {
super(processor, executorService);
- this.maximumRequestsPerPeriod = maximumRequestsPerPeriod;
- if (maximumRequestsPerPeriod <= 0) {
- throw new IllegalArgumentException("MaximumRequestsPerPeriod
should be a positive number, was: " + maximumRequestsPerPeriod);
+
+ ObjectHelper.notNull(maxRequestsPerPeriodExpression,
"maxRequestsPerPeriodExpression");
+ this.maxRequestsPerPeriodExpression = maxRequestsPerPeriodExpression;
+
+ if (timePeriodMillis <= 0) {
+ throw new IllegalArgumentException("TimePeriodMillis should be a
positive number, was: " + timePeriodMillis);
}
this.timePeriodMillis = timePeriodMillis;
}
@Override
public String toString() {
- return "Throttler[requests: " + maximumRequestsPerPeriod + " per: " +
timePeriodMillis + " (ms) to: "
+ return "Throttler[requests: " + maxRequestsPerPeriodExpression + "
per: " + timePeriodMillis + " (ms) to: "
+ getProcessor() + "]";
}
public String getTraceLabel() {
- return "throttle[" + maximumRequestsPerPeriod + " per: " +
timePeriodMillis + "]";
+ return "throttle[" + maxRequestsPerPeriodExpression + " per: " +
timePeriodMillis + "]";
}
// Properties
// -----------------------------------------------------------------------
- public long getMaximumRequestsPerPeriod() {
- return maximumRequestsPerPeriod;
- }
/**
- * Sets the maximum number of requests per time period
+ * Sets the maximum number of requests per time period expression
*/
- public void setMaximumRequestsPerPeriod(long maximumRequestsPerPeriod) {
- this.maximumRequestsPerPeriod = maximumRequestsPerPeriod;
+ public void setMaximumRequestsPerPeriodExpression(Expression
maxRequestsPerPeriodExpression) {
+ this.maxRequestsPerPeriodExpression = maxRequestsPerPeriodExpression;
}
+ public Expression getMaximumRequestsPerPeriodExpression() {
+ return maxRequestsPerPeriodExpression;
+ }
+
public long getTimePeriodMillis() {
return timePeriodMillis;
}
/**
+ * Gets the current maximum request per period value.
+ */
+ public long getCurrentMaximumRequestsPerPeriod() {
+ return maximumRequestsPerPeriod;
+ }
+
+ /**
* Sets the time period during which the maximum number of requests apply
*/
public void setTimePeriodMillis(long timePeriodMillis) {
@@ -88,6 +98,19 @@ public class Throttler extends DelayProc
// -----------------------------------------------------------------------
protected long calculateDelay(Exchange exchange) {
+ Long longValue = maxRequestsPerPeriodExpression.evaluate(exchange,
Long.class);
+ if (longValue != null) {
+ // log if we changed max period after initial setting
+ if (maximumRequestsPerPeriod > 0 && longValue.longValue() !=
maximumRequestsPerPeriod) {
+ log.debug("Throttler changed maximum requests per period from
{} to {}", maximumRequestsPerPeriod, longValue);
+ }
+ maximumRequestsPerPeriod = longValue;
+ }
+
+ if (maximumRequestsPerPeriod <= 0) {
+ throw new IllegalStateException("The maximumRequestsPerPeriod must
be a positive number, was: " + maximumRequestsPerPeriod);
+ }
+
TimeSlot slot = nextSlot();
if (!slot.isActive()) {
long delay = slot.startTime - currentSystemTime();
@@ -127,7 +150,7 @@ public class Throttler extends DelayProc
protected TimeSlot(long startTime) {
this.startTime = startTime;
}
-
+
protected void assign() {
capacity--;
}
Modified:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/ThrottlerTest.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/ThrottlerTest.java?rev=1093970&r1=1093969&r2=1093970&view=diff
==============================================================================
---
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/ThrottlerTest.java
(original)
+++
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/ThrottlerTest.java
Sat Apr 16 11:45:55 2011
@@ -22,8 +22,11 @@ import java.util.concurrent.Executors;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.DefaultExchange;
import org.apache.camel.processor.Throttler.TimeSlot;
+import static org.apache.camel.builder.Builder.constant;
+
/**
* @version
*/
@@ -71,11 +74,13 @@ public class ThrottlerTest extends Conte
}
public void testTimeSlotCalculus() throws Exception {
- Throttler throttler = new Throttler(null, 2, 1000, null);
+ Throttler throttler = new Throttler(null, constant(3), 1000, null);
+ // calculate will assign a new slot
+ throttler.calculateDelay(new DefaultExchange(context));
TimeSlot slot = throttler.nextSlot();
// start a new time slot
assertNotNull(slot);
- // make sure the same slot is used (2 exchanges per slot)
+ // make sure the same slot is used (3 exchanges per slot)
assertSame(slot, throttler.nextSlot());
assertTrue(slot.isFull());
@@ -85,6 +90,56 @@ public class ThrottlerTest extends Conte
assertFalse(next.isActive());
}
+ public void testConfigurationWithConstantExpression() throws Exception {
+ MockEndpoint resultEndpoint = resolveMandatoryEndpoint("mock:result",
MockEndpoint.class);
+ resultEndpoint.expectedMessageCount(messageCount);
+
+ ExecutorService executor = Executors.newFixedThreadPool(messageCount);
+
+ long start = System.currentTimeMillis();
+ for (int i = 0; i < messageCount; i++) {
+ executor.execute(new Runnable() {
+ public void run() {
+ template.sendBody("direct:expressionConstant",
"<message>payload</message>");
+ }
+ });
+ }
+
+ // let's wait for the exchanges to arrive
+ resultEndpoint.assertIsSatisfied();
+
+ // now assert that they have actually been throttled
+ long minimumTime = (messageCount - 1) * INTERVAL;
+ // add a little slack
+ long delta = System.currentTimeMillis() - start + 200;
+ assertTrue("Should take at least " + minimumTime + "ms, was: " +
delta, delta >= minimumTime);
+ }
+
+ public void testConfigurationWithHeaderExpression() throws Exception {
+ MockEndpoint resultEndpoint = resolveMandatoryEndpoint("mock:result",
MockEndpoint.class);
+ resultEndpoint.expectedMessageCount(messageCount);
+
+ ExecutorService executor = Executors.newFixedThreadPool(messageCount);
+
+ long start = System.currentTimeMillis();
+ for (int i = 0; i < messageCount; i++) {
+ executor.execute(new Runnable() {
+ public void run() {
+ template.sendBodyAndHeader("direct:expressionHeader",
"<message>payload</message>", "throttleValue", 1);
+ }
+ });
+ }
+
+ // let's wait for the exchanges to arrive
+ resultEndpoint.assertIsSatisfied();
+
+ // now assert that they have actually been throttled
+ long minimumTime = (messageCount - 1) * INTERVAL;
+ // add a little slack
+ long delta = System.currentTimeMillis() - start + 200;
+ assertTrue("Should take at least " + minimumTime + "ms, was: " +
delta, delta >= minimumTime);
+ }
+
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
public void configure() {
@@ -93,6 +148,10 @@ public class ThrottlerTest extends Conte
// END SNIPPET: ex
from("direct:a").throttle(1).timePeriodMillis(INTERVAL).to("log:result",
"mock:result");
+
+
from("direct:expressionConstant").throttle(constant(1)).timePeriodMillis(INTERVAL).to("log:result",
"mock:result");
+
+
from("direct:expressionHeader").throttle(header("throttleValue")).timePeriodMillis(INTERVAL).to("log:result",
"mock:result");
}
};
}
Modified:
camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/throttler.xml
URL:
http://svn.apache.org/viewvc/camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/throttler.xml?rev=1093970&r1=1093969&r2=1093970&view=diff
==============================================================================
---
camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/throttler.xml
(original)
+++
camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/throttler.xml
Sat Apr 16 11:45:55 2011
@@ -22,24 +22,44 @@
http://camel.apache.org/schema/spring
http://camel.apache.org/schema/spring/camel-spring.xsd
">
- <camelContext
- xmlns="http://camel.apache.org/schema/spring";>
- <!-- START SNIPPET: example -->
- <route>
- <from uri="seda:a" />
- <throttle maximumRequestsPerPeriod="3" timePeriodMillis="10000">
- <to uri="mock:result" />
- </throttle>
- </route>
- <!-- END SNIPPET: example -->
-
- <route>
- <from uri="direct:a" />
- <throttle maximumRequestsPerPeriod="1" timePeriodMillis="500">
- <to uri="mock:result" />
- </throttle>
- </route>
+ <camelContext xmlns="http://camel.apache.org/schema/spring";>
- </camelContext>
+ <!-- START SNIPPET: example -->
+ <route>
+ <from uri="seda:a"/>
+ <!-- throttle 3 messages per 10 sec -->
+ <throttle timePeriodMillis="10000">
+ <constant>3</constant>
+ <to uri="mock:result"/>
+ </throttle>
+ </route>
+ <!-- END SNIPPET: example -->
+
+ <route>
+ <from uri="direct:a"/>
+ <!-- throttle 1 messages per 0.5 sec -->
+ <throttle timePeriodMillis="500">
+ <constant>1</constant>
+ <to uri="mock:result"/>
+ </throttle>
+ </route>
+
+ <route>
+ <from uri="direct:expressionConstant"/>
+ <throttle timePeriodMillis="500">
+ <constant>1</constant>
+ <to uri="mock:result"/>
+ </throttle>
+ </route>
+
+ <route>
+ <from uri="direct:expressionHeader"/>
+ <throttle timePeriodMillis="500">
+ <header>throttleValue</header>
+ <to uri="mock:result"/>
+ </throttle>
+ </route>
+
+ </camelContext>
</beans>
Modified:
camel/trunk/examples/camel-example-management/src/main/resources/META-INF/spring/camel-context.xml
URL:
http://svn.apache.org/viewvc/camel/trunk/examples/camel-example-management/src/main/resources/META-INF/spring/camel-context.xml?rev=1093970&r1=1093969&r2=1093970&view=diff
==============================================================================
---
camel/trunk/examples/camel-example-management/src/main/resources/META-INF/spring/camel-context.xml
(original)
+++
camel/trunk/examples/camel-example-management/src/main/resources/META-INF/spring/camel-context.xml
Sat Apr 16 11:45:55 2011
@@ -52,7 +52,9 @@
<from uri="file://target/inbox"/>
<split>
<xpath>/stocks/stock</xpath>
- <throttle timePeriodMillis="1000"
maximumRequestsPerPeriod="10">
+ <!-- throttle 10 msg/sec being send the the ActiveMQ queue -->
+ <throttle>
+ <constant>10</constant>
<to uri="activemq:queue:stock"/>
</throttle>
</split>
