Repository: camel Updated Branches: refs/heads/master 4f872f075 -> ef217c3e3
CAMEL-11446: Use awaitility in camel-core for testing where we otherwise use thred sleep which can be speeded up. Project: http://git-wip-us.apache.org/repos/asf/camel/repo Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/ef217c3e Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/ef217c3e Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/ef217c3e Branch: refs/heads/master Commit: ef217c3e338b9d493e8dd7837423df2b51353e37 Parents: 4f872f0 Author: Claus Ibsen <davscl...@apache.org> Authored: Sun Jul 9 13:17:34 2017 +0200 Committer: Claus Ibsen <davscl...@apache.org> Committed: Sun Jul 9 13:39:41 2017 +0200 ---------------------------------------------------------------------- .../camel/impl/StartAndStopRoutesTest.java | 18 +- .../camel/management/BacklogDebuggerTest.java | 199 ++++++++++--------- .../camel/management/ManagedThrottlerTest.java | 9 +- .../apache/camel/processor/ThrottlerTest.java | 29 ++- .../spring/processor/SpringThrottlerTest.java | 1 + .../apache/camel/spring/processor/throttler.xml | 16 +- 6 files changed, 136 insertions(+), 136 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/camel/blob/ef217c3e/camel-core/src/test/java/org/apache/camel/impl/StartAndStopRoutesTest.java ---------------------------------------------------------------------- diff --git a/camel-core/src/test/java/org/apache/camel/impl/StartAndStopRoutesTest.java b/camel-core/src/test/java/org/apache/camel/impl/StartAndStopRoutesTest.java index 8a9fa49..912ff8e 100644 --- a/camel-core/src/test/java/org/apache/camel/impl/StartAndStopRoutesTest.java +++ b/camel-core/src/test/java/org/apache/camel/impl/StartAndStopRoutesTest.java @@ -19,9 +19,9 @@ package org.apache.camel.impl; import java.util.List; import org.apache.camel.ContextTestSupport; +import org.apache.camel.Endpoint; import org.apache.camel.builder.RouteBuilder; import org.apache.camel.component.mock.MockEndpoint; -import org.apache.camel.component.seda.SedaEndpoint; import org.apache.camel.model.FromDefinition; import org.apache.camel.model.RouteDefinition; @@ -31,9 +31,9 @@ import org.apache.camel.model.RouteDefinition; * @version */ public class StartAndStopRoutesTest extends ContextTestSupport { - protected SedaEndpoint endpointA; - protected SedaEndpoint endpointB; - protected SedaEndpoint endpointC; + protected Endpoint endpointA; + protected Endpoint endpointB; + protected Endpoint endpointC; protected Object expectedBody = "<hello>world!</hello>"; public void testStartRouteThenStopMutateAndStartRouteAgain() throws Exception { @@ -41,9 +41,9 @@ public class StartAndStopRoutesTest extends ContextTestSupport { assertCollectionSize("Route", routes, 1); RouteDefinition route = routes.get(0); - endpointA = getMandatoryEndpoint("seda:test.a", SedaEndpoint.class); - endpointB = getMandatoryEndpoint("seda:test.b", SedaEndpoint.class); - endpointC = getMandatoryEndpoint("seda:test.C", SedaEndpoint.class); + endpointA = getMandatoryEndpoint("direct:test.a"); + endpointB = getMandatoryEndpoint("seda:test.b"); + endpointC = getMandatoryEndpoint("direct:test.C"); // send from A over B to results MockEndpoint results = getMockEndpoint("mock:results"); @@ -58,7 +58,7 @@ public class StartAndStopRoutesTest extends ContextTestSupport { // lets mutate the route... FromDefinition fromType = assertOneElement(route.getInputs()); - fromType.setUri("seda:test.C"); + fromType.setUri("direct:test.C"); context.startRoute(route); // now lets check it works @@ -77,7 +77,7 @@ public class StartAndStopRoutesTest extends ContextTestSupport { return new RouteBuilder() { @Override public void configure() throws Exception { - from("seda:test.a"). + from("direct:test.a"). to("seda:test.b"). to("mock:results"); } http://git-wip-us.apache.org/repos/asf/camel/blob/ef217c3e/camel-core/src/test/java/org/apache/camel/management/BacklogDebuggerTest.java ---------------------------------------------------------------------- diff --git a/camel-core/src/test/java/org/apache/camel/management/BacklogDebuggerTest.java b/camel-core/src/test/java/org/apache/camel/management/BacklogDebuggerTest.java index 135a260..84e0b3c 100644 --- a/camel-core/src/test/java/org/apache/camel/management/BacklogDebuggerTest.java +++ b/camel-core/src/test/java/org/apache/camel/management/BacklogDebuggerTest.java @@ -17,12 +17,15 @@ package org.apache.camel.management; import java.util.Set; +import java.util.concurrent.TimeUnit; import javax.management.MBeanServer; import javax.management.ObjectName; import org.apache.camel.builder.RouteBuilder; import org.apache.camel.component.mock.MockEndpoint; +import static org.awaitility.Awaitility.await; + public class BacklogDebuggerTest extends ManagementTestSupport { @SuppressWarnings("unchecked") @@ -51,17 +54,19 @@ public class BacklogDebuggerTest extends ManagementTestSupport { MockEndpoint mock = getMockEndpoint("mock:result"); mock.expectedMessageCount(0); - mock.setSleepForEmptyTest(1000); + mock.setSleepForEmptyTest(100); template.sendBody("seda:start", "Hello World"); assertMockEndpointsSatisfied(); - // add breakpoint at bar - Set<String> nodes = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); - assertNotNull(nodes); - assertEquals(1, nodes.size()); - assertEquals("bar", nodes.iterator().next()); + // wait for breakpoint at bar + await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> { + Set<String> suspended = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); + assertNotNull(suspended); + assertEquals(1, suspended.size()); + assertEquals("bar", suspended.iterator().next()); + }); // the message should be ours String xml = (String) mbeanServer.invoke(on, "dumpTracedMessagesAsXml", new Object[]{"bar"}, new String[]{"java.lang.String"}); @@ -80,7 +85,7 @@ public class BacklogDebuggerTest extends ManagementTestSupport { assertMockEndpointsSatisfied(); // and no suspended anymore - nodes = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); + Set<String> nodes = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); assertNotNull(nodes); assertEquals(0, nodes.size()); } @@ -112,7 +117,7 @@ public class BacklogDebuggerTest extends ManagementTestSupport { MockEndpoint mock = getMockEndpoint("mock:result"); mock.expectedMessageCount(0); - mock.setSleepForEmptyTest(1000); + mock.setSleepForEmptyTest(100); template.sendBody("seda:start", "Hello World"); @@ -131,13 +136,13 @@ public class BacklogDebuggerTest extends ManagementTestSupport { // resume breakpoint mbeanServer.invoke(on, "resumeBreakpoint", new Object[]{"foo"}, new String[]{"java.lang.String"}); - Thread.sleep(1000); - - // add breakpoint at bar - nodes = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); - assertNotNull(nodes); - assertEquals(1, nodes.size()); - assertEquals("bar", nodes.iterator().next()); + // wait for breakpoint at bar + await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> { + Set<String> suspended = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); + assertNotNull(suspended); + assertEquals(1, suspended.size()); + assertEquals("bar", suspended.iterator().next()); + }); // the message should be ours String xml = (String) mbeanServer.invoke(on, "dumpTracedMessagesAsXml", new Object[]{"bar"}, new String[]{"java.lang.String"}); @@ -189,7 +194,7 @@ public class BacklogDebuggerTest extends ManagementTestSupport { MockEndpoint mock = getMockEndpoint("mock:result"); mock.expectedMessageCount(0); - mock.setSleepForEmptyTest(1000); + mock.setSleepForEmptyTest(100); template.sendBody("seda:start", "Hello World"); @@ -210,13 +215,13 @@ public class BacklogDebuggerTest extends ManagementTestSupport { // resume breakpoint mbeanServer.invoke(on, "resumeBreakpoint", new Object[]{"foo"}, new String[]{"java.lang.String"}); - Thread.sleep(1000); - - // add breakpoint at bar - nodes = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); - assertNotNull(nodes); - assertEquals(1, nodes.size()); - assertEquals("bar", nodes.iterator().next()); + // wait for breakpoint at bar + await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> { + Set<String> suspended = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); + assertNotNull(suspended); + assertEquals(1, suspended.size()); + assertEquals("bar", suspended.iterator().next()); + }); // the message should be ours String xml = (String) mbeanServer.invoke(on, "dumpTracedMessagesAsXml", new Object[]{"bar"}, new String[]{"java.lang.String"}); @@ -268,7 +273,7 @@ public class BacklogDebuggerTest extends ManagementTestSupport { MockEndpoint mock = getMockEndpoint("mock:result"); mock.expectedMessageCount(0); - mock.setSleepForEmptyTest(1000); + mock.setSleepForEmptyTest(100); template.sendBody("seda:start", "Hello World"); @@ -287,13 +292,13 @@ public class BacklogDebuggerTest extends ManagementTestSupport { // resume breakpoint mbeanServer.invoke(on, "resumeBreakpoint", new Object[]{"foo"}, new String[]{"java.lang.String"}); - Thread.sleep(1000); - - // add breakpoint at bar - nodes = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); - assertNotNull(nodes); - assertEquals(1, nodes.size()); - assertEquals("bar", nodes.iterator().next()); + // wait for breakpoint at bar + await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> { + Set<String> suspended = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); + assertNotNull(suspended); + assertEquals(1, suspended.size()); + assertEquals("bar", suspended.iterator().next()); + }); // the message should be ours String xml = (String) mbeanServer.invoke(on, "dumpTracedMessagesAsXml", new Object[]{"bar"}, new String[]{"java.lang.String"}); @@ -431,7 +436,7 @@ public class BacklogDebuggerTest extends ManagementTestSupport { resetMocks(); mock.expectedMessageCount(0); - mock.setSleepForEmptyTest(1000); + mock.setSleepForEmptyTest(100); template.sendBody("seda:start", "Hello Camel"); @@ -490,7 +495,7 @@ public class BacklogDebuggerTest extends ManagementTestSupport { MockEndpoint mock = getMockEndpoint("mock:result"); mock.expectedMessageCount(0); - mock.setSleepForEmptyTest(1000); + mock.setSleepForEmptyTest(100); template.sendBody("seda:start", "Hello World"); @@ -509,58 +514,57 @@ public class BacklogDebuggerTest extends ManagementTestSupport { mbeanServer.invoke(on, "stepBreakpoint", new Object[]{"foo"}, new String[]{"java.lang.String"}); // then at bar now - Thread.sleep(1000); - nodes = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); - assertNotNull(nodes); - assertEquals(1, nodes.size()); - assertEquals("bar", nodes.iterator().next()); - stepMode = (Boolean) mbeanServer.getAttribute(on, "SingleStepMode"); - assertEquals("Should be in step mode", Boolean.TRUE, stepMode); + await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> { + Set<String> suspended = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); + assertNotNull(suspended); + assertEquals(1, suspended.size()); + assertEquals("bar", suspended.iterator().next()); + }); // step mbeanServer.invoke(on, "step", null, null); // then at transform now - Thread.sleep(1000); - nodes = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); - assertNotNull(nodes); - assertEquals(1, nodes.size()); - assertEquals("transform", nodes.iterator().next()); - stepMode = (Boolean) mbeanServer.getAttribute(on, "SingleStepMode"); - assertEquals("Should be in step mode", Boolean.TRUE, stepMode); + await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> { + Set<String> suspended = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); + assertNotNull(suspended); + assertEquals(1, suspended.size()); + assertEquals("transform", suspended.iterator().next()); + }); // step mbeanServer.invoke(on, "step", null, null); // then at cheese now - Thread.sleep(1000); - nodes = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); - assertNotNull(nodes); - assertEquals(1, nodes.size()); - assertEquals("cheese", nodes.iterator().next()); - stepMode = (Boolean) mbeanServer.getAttribute(on, "SingleStepMode"); - assertEquals("Should be in step mode", Boolean.TRUE, stepMode); + await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> { + Set<String> suspended = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); + assertNotNull(suspended); + assertEquals(1, suspended.size()); + assertEquals("cheese", suspended.iterator().next()); + }); // step mbeanServer.invoke(on, "step", null, null); // then at result now - Thread.sleep(1000); - nodes = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); - assertNotNull(nodes); - assertEquals(1, nodes.size()); - assertEquals("result", nodes.iterator().next()); - stepMode = (Boolean) mbeanServer.getAttribute(on, "SingleStepMode"); - assertEquals("Should be in step mode", Boolean.TRUE, stepMode); + await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> { + Set<String> suspended = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); + assertNotNull(suspended); + assertEquals(1, suspended.size()); + assertEquals("result", suspended.iterator().next()); + }); // step mbeanServer.invoke(on, "step", null, null); // then the exchange is completed - Thread.sleep(1000); - nodes = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); - assertNotNull(nodes); - assertEquals(0, nodes.size()); + await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> { + Set<String> suspended = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); + assertNotNull(suspended); + assertEquals(0, suspended.size()); + }); + + // should no longer be in step mode stepMode = (Boolean) mbeanServer.getAttribute(on, "SingleStepMode"); assertEquals("Should not be in step mode", Boolean.FALSE, stepMode); } @@ -591,7 +595,7 @@ public class BacklogDebuggerTest extends ManagementTestSupport { MockEndpoint mock = getMockEndpoint("mock:result"); mock.expectedMessageCount(0); - mock.setSleepForEmptyTest(1000); + mock.setSleepForEmptyTest(100); template.sendBody("seda:start", "Hello World"); @@ -610,58 +614,57 @@ public class BacklogDebuggerTest extends ManagementTestSupport { mbeanServer.invoke(on, "stepBreakpoint", new Object[]{"foo"}, new String[]{"java.lang.String"}); // then at bar now - Thread.sleep(1000); - nodes = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); - assertNotNull(nodes); - assertEquals(1, nodes.size()); - assertEquals("bar", nodes.iterator().next()); - stepMode = (Boolean) mbeanServer.getAttribute(on, "SingleStepMode"); - assertEquals("Should be in step mode", Boolean.TRUE, stepMode); + await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> { + Set<String> suspended = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); + assertNotNull(suspended); + assertEquals(1, suspended.size()); + assertEquals("bar", suspended.iterator().next()); + }); // step mbeanServer.invoke(on, "stepBreakpoint", new Object[]{"bar"}, new String[]{"java.lang.String"}); // then at transform now - Thread.sleep(1000); - nodes = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); - assertNotNull(nodes); - assertEquals(1, nodes.size()); - assertEquals("transform", nodes.iterator().next()); - stepMode = (Boolean) mbeanServer.getAttribute(on, "SingleStepMode"); - assertEquals("Should be in step mode", Boolean.TRUE, stepMode); + await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> { + Set<String> suspended = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); + assertNotNull(suspended); + assertEquals(1, suspended.size()); + assertEquals("transform", suspended.iterator().next()); + }); // step mbeanServer.invoke(on, "stepBreakpoint", new Object[]{"transform"}, new String[]{"java.lang.String"}); // then at cheese now - Thread.sleep(1000); - nodes = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); - assertNotNull(nodes); - assertEquals(1, nodes.size()); - assertEquals("cheese", nodes.iterator().next()); - stepMode = (Boolean) mbeanServer.getAttribute(on, "SingleStepMode"); - assertEquals("Should be in step mode", Boolean.TRUE, stepMode); + await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> { + Set<String> suspended = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); + assertNotNull(suspended); + assertEquals(1, suspended.size()); + assertEquals("cheese", suspended.iterator().next()); + }); // step mbeanServer.invoke(on, "stepBreakpoint", new Object[]{"cheese"}, new String[]{"java.lang.String"}); // then at result now - Thread.sleep(1000); - nodes = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); - assertNotNull(nodes); - assertEquals(1, nodes.size()); - assertEquals("result", nodes.iterator().next()); - stepMode = (Boolean) mbeanServer.getAttribute(on, "SingleStepMode"); - assertEquals("Should be in step mode", Boolean.TRUE, stepMode); + await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> { + Set<String> suspended = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); + assertNotNull(suspended); + assertEquals(1, suspended.size()); + assertEquals("result", suspended.iterator().next()); + }); // step mbeanServer.invoke(on, "stepBreakpoint", new Object[]{"result"}, new String[]{"java.lang.String"}); // then the exchange is completed - Thread.sleep(1000); - nodes = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); - assertNotNull(nodes); - assertEquals(0, nodes.size()); + await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> { + Set<String> suspended = (Set<String>) mbeanServer.invoke(on, "getSuspendedBreakpointNodeIds", null, null); + assertNotNull(suspended); + assertEquals(0, suspended.size()); + }); + + // should no longer be in step mode stepMode = (Boolean) mbeanServer.getAttribute(on, "SingleStepMode"); assertEquals("Should not be in step mode", Boolean.FALSE, stepMode); } http://git-wip-us.apache.org/repos/asf/camel/blob/ef217c3e/camel-core/src/test/java/org/apache/camel/management/ManagedThrottlerTest.java ---------------------------------------------------------------------- diff --git a/camel-core/src/test/java/org/apache/camel/management/ManagedThrottlerTest.java b/camel-core/src/test/java/org/apache/camel/management/ManagedThrottlerTest.java index 1a06721..90f9477 100644 --- a/camel-core/src/test/java/org/apache/camel/management/ManagedThrottlerTest.java +++ b/camel-core/src/test/java/org/apache/camel/management/ManagedThrottlerTest.java @@ -76,11 +76,11 @@ public class ManagedThrottlerTest extends ManagementTestSupport { assertEquals(10, completed.longValue()); Long timePeriod = (Long) mbeanServer.getAttribute(throttlerName, "TimePeriodMillis"); - assertEquals(1000, timePeriod.longValue()); + assertEquals(250, timePeriod.longValue()); Long total = (Long) mbeanServer.getAttribute(routeName, "TotalProcessingTime"); - assertTrue("Should take at most 2.0 sec: was " + total, total < 2000); + assertTrue("Should take at most 1.0 sec: was " + total, total < 1000); // change the throttler using JMX mbeanServer.setAttribute(throttlerName, new Attribute("MaximumRequestsPerPeriod", (long) 2)); @@ -101,7 +101,7 @@ public class ManagedThrottlerTest extends ManagementTestSupport { assertEquals(10, completed.longValue()); total = (Long) mbeanServer.getAttribute(routeName, "TotalProcessingTime"); - assertTrue("Should be around 5 sec now: was " + total, total > 3500); + assertTrue("Should be around 1 sec now: was " + total, total > 1000); } public void testThrottleVisableViaJmx() throws Exception { @@ -303,7 +303,7 @@ public class ManagedThrottlerTest extends ManagementTestSupport { public void configure() throws Exception { from("direct:start") .to("log:foo") - .throttle(10).id("mythrottler") + .throttle(10).timePeriodMillis(250).id("mythrottler") .to("mock:result"); from("seda:throttleCount") @@ -318,7 +318,6 @@ public class ManagedThrottlerTest extends ManagementTestSupport { .throttle(1).asyncDelayed().timePeriodMillis(250).id("mythrottler4") .to("mock:endAsyncException") .process(new Processor() { - @Override public void process(Exchange exchange) throws Exception { throw new RuntimeException("Fail me"); http://git-wip-us.apache.org/repos/asf/camel/blob/ef217c3e/camel-core/src/test/java/org/apache/camel/processor/ThrottlerTest.java ---------------------------------------------------------------------- diff --git a/camel-core/src/test/java/org/apache/camel/processor/ThrottlerTest.java b/camel-core/src/test/java/org/apache/camel/processor/ThrottlerTest.java index e4dd502..5edc730 100644 --- a/camel-core/src/test/java/org/apache/camel/processor/ThrottlerTest.java +++ b/camel-core/src/test/java/org/apache/camel/processor/ThrottlerTest.java @@ -59,11 +59,9 @@ public class ThrottlerTest extends ContextTestSupport { MockEndpoint resultEndpoint = resolveMandatoryEndpoint("mock:result", MockEndpoint.class); resultEndpoint.expectedMessageCount(2); - resultEndpoint.setResultWaitTime(2000); - + MockEndpoint errorEndpoint = resolveMandatoryEndpoint("mock:error", MockEndpoint.class); errorEndpoint.expectedMessageCount(4); - errorEndpoint.setResultWaitTime(2000); for (int i = 0; i < 6; i++) { template.sendBody("direct:start", "<message>" + i + "</message>"); @@ -71,8 +69,7 @@ public class ThrottlerTest extends ContextTestSupport { // lets pause to give the requests time to be processed // to check that the throttle really does kick in - resultEndpoint.assertIsSatisfied(); - errorEndpoint.assertIsSatisfied(); + assertMockEndpointsSatisfied(); } public void testSendLotsOfMessagesSimultaneouslyButOnly3GetThrough() throws Exception { @@ -82,7 +79,7 @@ public class ThrottlerTest extends ContextTestSupport { MockEndpoint resultEndpoint = resolveMandatoryEndpoint("mock:result", MockEndpoint.class); long elapsed = sendMessagesAndAwaitDelivery(MESSAGE_COUNT, "direct:a", MESSAGE_COUNT, resultEndpoint); - assertThrottlerTiming(elapsed, 1, INTERVAL, MESSAGE_COUNT); + assertThrottlerTiming(elapsed, 5, INTERVAL, MESSAGE_COUNT); } public void testConfigurationWithConstantExpression() throws Exception { @@ -92,7 +89,7 @@ public class ThrottlerTest extends ContextTestSupport { MockEndpoint resultEndpoint = resolveMandatoryEndpoint("mock:result", MockEndpoint.class); long elapsed = sendMessagesAndAwaitDelivery(MESSAGE_COUNT, "direct:expressionConstant", MESSAGE_COUNT, resultEndpoint); - assertThrottlerTiming(elapsed, 1, INTERVAL, MESSAGE_COUNT); + assertThrottlerTiming(elapsed, 5, INTERVAL, MESSAGE_COUNT); } public void testConfigurationWithHeaderExpression() throws Exception { @@ -105,7 +102,7 @@ public class ThrottlerTest extends ContextTestSupport { ExecutorService executor = Executors.newFixedThreadPool(MESSAGE_COUNT); try { - sendMessagesWithHeaderExpression(executor, resultEndpoint, 1, INTERVAL, MESSAGE_COUNT); + sendMessagesWithHeaderExpression(executor, resultEndpoint, 5, INTERVAL, MESSAGE_COUNT); } finally { executor.shutdownNow(); } @@ -119,7 +116,7 @@ public class ThrottlerTest extends ContextTestSupport { ExecutorService executor = Executors.newFixedThreadPool(5); try { MockEndpoint resultEndpoint = resolveMandatoryEndpoint("mock:result", MockEndpoint.class); - sendMessagesWithHeaderExpression(executor, resultEndpoint, 1, INTERVAL, MESSAGE_COUNT); + sendMessagesWithHeaderExpression(executor, resultEndpoint, 5, INTERVAL, MESSAGE_COUNT); Thread.sleep(INTERVAL + TOLERANCE); // sleep here to ensure the first throttle rate does not influence the next one. resultEndpoint.reset(); @@ -127,7 +124,7 @@ public class ThrottlerTest extends ContextTestSupport { Thread.sleep(INTERVAL + TOLERANCE); // sleep here to ensure the first throttle rate does not influence the next one. resultEndpoint.reset(); - sendMessagesWithHeaderExpression(executor, resultEndpoint, 1, INTERVAL, MESSAGE_COUNT); + sendMessagesWithHeaderExpression(executor, resultEndpoint, 5, INTERVAL, MESSAGE_COUNT); Thread.sleep(INTERVAL + TOLERANCE); // sleep here to ensure the first throttle rate does not influence the next one. resultEndpoint.reset(); @@ -145,7 +142,7 @@ public class ThrottlerTest extends ContextTestSupport { int messageCount = 20000; MockEndpoint resultEndpoint = resolveMandatoryEndpoint("mock:result", MockEndpoint.class); long elapsed = sendMessagesAndAwaitDelivery(messageCount, "direct:highThrottleRate", 5, resultEndpoint); - assertThrottlerTiming(elapsed, 1000, INTERVAL, messageCount); + assertThrottlerTiming(elapsed, 10000, INTERVAL, messageCount); } private void assertThrottlerTiming(final long elapsedTimeMs, final int throttle, final int intervalMs, final int messageCount) { @@ -224,18 +221,18 @@ public class ThrottlerTest extends ContextTestSupport { .to("mock:error"); // START SNIPPET: ex - from("seda:a").throttle(3).timePeriodMillis(10000).to("log:result", "mock:result"); + from("seda:a").throttle(3).timePeriodMillis(1000).to("log:result", "mock:result"); // END SNIPPET: ex - from("direct:a").throttle(1).timePeriodMillis(INTERVAL).to("log:result", "mock:result"); + from("direct:a").throttle(5).timePeriodMillis(INTERVAL).to("log:result", "mock:result"); - from("direct:expressionConstant").throttle(constant(1)).timePeriodMillis(INTERVAL).to("log:result", "mock:result"); + from("direct:expressionConstant").throttle(constant(5)).timePeriodMillis(INTERVAL).to("log:result", "mock:result"); from("direct:expressionHeader").throttle(header("throttleValue")).timePeriodMillis(INTERVAL).to("log:result", "mock:result"); - from("direct:start").throttle(2).timePeriodMillis(10000).rejectExecution(true).to("log:result", "mock:result"); + from("direct:start").throttle(2).timePeriodMillis(1000).rejectExecution(true).to("log:result", "mock:result"); - from("direct:highThrottleRate").throttle(1000).timePeriodMillis(INTERVAL).to("mock:result"); + from("direct:highThrottleRate").throttle(10000).timePeriodMillis(INTERVAL).to("mock:result"); } }; } http://git-wip-us.apache.org/repos/asf/camel/blob/ef217c3e/components/camel-spring/src/test/java/org/apache/camel/spring/processor/SpringThrottlerTest.java ---------------------------------------------------------------------- diff --git a/components/camel-spring/src/test/java/org/apache/camel/spring/processor/SpringThrottlerTest.java b/components/camel-spring/src/test/java/org/apache/camel/spring/processor/SpringThrottlerTest.java index fbfe79d..c0eabf2 100644 --- a/components/camel-spring/src/test/java/org/apache/camel/spring/processor/SpringThrottlerTest.java +++ b/components/camel-spring/src/test/java/org/apache/camel/spring/processor/SpringThrottlerTest.java @@ -21,6 +21,7 @@ import org.apache.camel.processor.ThrottlerTest; import static org.apache.camel.spring.processor.SpringTestHelper.createSpringCamelContext; public class SpringThrottlerTest extends ThrottlerTest { + protected CamelContext createCamelContext() throws Exception { return createSpringCamelContext(this, "org/apache/camel/spring/processor/throttler.xml"); http://git-wip-us.apache.org/repos/asf/camel/blob/ef217c3e/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/throttler.xml ---------------------------------------------------------------------- diff --git a/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/throttler.xml b/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/throttler.xml index 0906278..c572a07 100644 --- a/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/throttler.xml +++ b/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/throttler.xml @@ -36,8 +36,8 @@ <!-- START SNIPPET: example --> <route> <from uri="seda:a"/> - <!-- throttle 3 messages per 10 sec --> - <throttle timePeriodMillis="10000"> + <!-- throttle 3 messages per 1 sec --> + <throttle timePeriodMillis="1000"> <constant>3</constant> <to uri="log:result"/> <to uri="mock:result"/> @@ -47,9 +47,9 @@ <route> <from uri="direct:a"/> - <!-- throttle 1 messages per 0.5 sec --> + <!-- throttle 5 messages per 0.5 sec --> <throttle timePeriodMillis="500"> - <constant>1</constant> + <constant>5</constant> <to uri="log:result"/> <to uri="mock:result"/> </throttle> @@ -58,7 +58,7 @@ <route> <from uri="direct:expressionConstant"/> <throttle timePeriodMillis="500"> - <constant>1</constant> + <constant>5</constant> <to uri="mock:result"/> </throttle> </route> @@ -77,8 +77,8 @@ <route> <from uri="direct:start"/> - <!-- throttle 2 messages per 10 sec --> - <throttle timePeriodMillis="10000" rejectExecution="true"> + <!-- throttle 2 messages per 1 sec --> + <throttle timePeriodMillis="1000" rejectExecution="true"> <constant>2</constant> <to uri="log:result"/> <to uri="mock:result"/> @@ -88,7 +88,7 @@ <route> <from uri="direct:highThrottleRate"/> <throttle timePeriodMillis="500"> - <constant>1000</constant> + <constant>10000</constant> <to uri="mock:result"/> </throttle> </route>
