Repository: camel Updated Branches: refs/heads/master 9c37b5648 -> 3cd99d0a0
CAMEL-11446: Use awaitility 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/3cd99d0a Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/3cd99d0a Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/3cd99d0a Branch: refs/heads/master Commit: 3cd99d0a099700cb8ab610d6d1356efb021015ed Parents: 9c37b56 Author: Claus Ibsen <[email protected]> Authored: Sun Jul 16 10:22:38 2017 +0200 Committer: Claus Ibsen <[email protected]> Committed: Sun Jul 16 10:23:46 2017 +0200 ---------------------------------------------------------------------- .../timer/TimerWithTimeOptionTest.java | 48 ++++++++++---------- .../management/ManagedSuspendedServiceTest.java | 6 +-- .../MultiCastParallelAndStreamCachingTest.java | 2 +- .../processor/ThrottlerMethodCallTest.java | 22 ++++----- ...ptionRoutePolicyHalfOpenHandlerSedaTest.java | 32 +++++++------ ...ExceptionRoutePolicyHalfOpenHandlerTest.java | 28 +++++++----- ...ottlingExceptionRoutePolicyHalfOpenTest.java | 32 +++++++------ .../ThrottlingExceptionRoutePolicyTest.java | 28 +++++++----- .../processor/ThrottlerMethodCallTest.xml | 6 +-- 9 files changed, 110 insertions(+), 94 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/camel/blob/3cd99d0a/camel-core/src/test/java/org/apache/camel/component/timer/TimerWithTimeOptionTest.java ---------------------------------------------------------------------- diff --git a/camel-core/src/test/java/org/apache/camel/component/timer/TimerWithTimeOptionTest.java b/camel-core/src/test/java/org/apache/camel/component/timer/TimerWithTimeOptionTest.java index eddad81..e84cf57 100644 --- a/camel-core/src/test/java/org/apache/camel/component/timer/TimerWithTimeOptionTest.java +++ b/camel-core/src/test/java/org/apache/camel/component/timer/TimerWithTimeOptionTest.java @@ -39,17 +39,17 @@ public class TimerWithTimeOptionTest extends ContextTestSupport { context.addRoutes(new RouteBuilder() { @Override public void configure() throws Exception { - Date future = new Date(new Date().getTime() + 1000); + Date future = new Date(new Date().getTime() + 10); - SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss"); + SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS"); String time = sdf.format(future); - fromF("timer://foo?time=%s", time).to("mock:result"); + fromF("timer://foo?delay=0&period=10&time=%s", time).to("mock:result"); } }); MockEndpoint mock = getMockEndpoint("mock:result"); - // period is default 1000 so we can get more messages + // period is 10 so we can get more messages mock.expectedMinimumMessageCount(1); context.start(); @@ -61,9 +61,9 @@ public class TimerWithTimeOptionTest extends ContextTestSupport { context.addRoutes(new RouteBuilder() { @Override public void configure() throws Exception { - Date future = new Date(new Date().getTime() + 1000); + Date future = new Date(new Date().getTime() + 10); - SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss"); + SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS"); String time = sdf.format(future); fromF("timer://foo?period=0&time=%s", time).to("mock:result"); @@ -82,17 +82,17 @@ public class TimerWithTimeOptionTest extends ContextTestSupport { context.addRoutes(new RouteBuilder() { @Override public void configure() throws Exception { - Date future = new Date(new Date().getTime() + 1000); + Date future = new Date(new Date().getTime() + 10); - SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss"); + SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS"); String time = sdf.format(future); - fromF("timer://foo?fixedRate=true&time=%s", time).to("mock:result"); + fromF("timer://foo?delay=0&period=10&fixedRate=true&time=%s", time).to("mock:result"); } }); MockEndpoint mock = getMockEndpoint("mock:result"); - // period is default 1000 so we can get more messages + // period is 10 so we can get more messages mock.expectedMinimumMessageCount(1); context.start(); @@ -104,17 +104,17 @@ public class TimerWithTimeOptionTest extends ContextTestSupport { context.addRoutes(new RouteBuilder() { @Override public void configure() throws Exception { - Date future = new Date(new Date().getTime() + 1000); + Date future = new Date(new Date().getTime() + 10); - SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); + SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS"); String time = sdf.format(future); - fromF("timer://foo?time=%s", time).to("mock:result"); + fromF("timer://foo?delay=0&period=10&time=%s", time).to("mock:result"); } }); MockEndpoint mock = getMockEndpoint("mock:result"); - // period is default 1000 so we can get more messages + // period is 10 so we can get more messages mock.expectedMinimumMessageCount(1); context.start(); @@ -126,12 +126,12 @@ public class TimerWithTimeOptionTest extends ContextTestSupport { context.addRoutes(new RouteBuilder() { @Override public void configure() throws Exception { - Date future = new Date(new Date().getTime() + 1000); + Date future = new Date(new Date().getTime() + 10); - SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); + SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS"); String time = sdf.format(future); - fromF("timer://foo?period=0&time=%s", time).to("mock:result"); + fromF("timer://foo?delay=0&period=0&time=%s", time).to("mock:result"); } }); @@ -147,17 +147,17 @@ public class TimerWithTimeOptionTest extends ContextTestSupport { context.addRoutes(new RouteBuilder() { @Override public void configure() throws Exception { - Date future = new Date(new Date().getTime() + 1000); + Date future = new Date(new Date().getTime() + 10); - SimpleDateFormat sdf = new SimpleDateFormat("dd-MM-yyyy HH:mm:ss"); + SimpleDateFormat sdf = new SimpleDateFormat("dd-MM-yyyy HH:mm:ss.SSS"); String time = sdf.format(future); - fromF("timer://foo?time=%s&pattern=dd-MM-yyyy HH:mm:ss", time).to("mock:result"); + fromF("timer://foo?delay=0&period=10&time=%s&pattern=dd-MM-yyyy HH:mm:ss.SSS", time).to("mock:result"); } }); MockEndpoint mock = getMockEndpoint("mock:result"); - // period is default 1000 so we can get more messages + // period is 10 so we can get more messages mock.expectedMinimumMessageCount(1); context.start(); @@ -169,12 +169,12 @@ public class TimerWithTimeOptionTest extends ContextTestSupport { context.addRoutes(new RouteBuilder() { @Override public void configure() throws Exception { - Date future = new Date(new Date().getTime() + 1000); + Date future = new Date(new Date().getTime() + 10); - SimpleDateFormat sdf = new SimpleDateFormat("dd-MM-yyyy HH:mm:ss"); + SimpleDateFormat sdf = new SimpleDateFormat("dd-MM-yyyy HH:mm:ss.SSS"); String time = sdf.format(future); - fromF("timer://foo?period=0&time=%s&pattern=dd-MM-yyyy HH:mm:ss", time).to("mock:result"); + fromF("timer://foo?delay=0&period=0&time=%s&pattern=dd-MM-yyyy HH:mm:ss.SSS", time).to("mock:result"); } }); http://git-wip-us.apache.org/repos/asf/camel/blob/3cd99d0a/camel-core/src/test/java/org/apache/camel/management/ManagedSuspendedServiceTest.java ---------------------------------------------------------------------- diff --git a/camel-core/src/test/java/org/apache/camel/management/ManagedSuspendedServiceTest.java b/camel-core/src/test/java/org/apache/camel/management/ManagedSuspendedServiceTest.java index 21f41da..3237732 100644 --- a/camel-core/src/test/java/org/apache/camel/management/ManagedSuspendedServiceTest.java +++ b/camel-core/src/test/java/org/apache/camel/management/ManagedSuspendedServiceTest.java @@ -70,9 +70,9 @@ public class ManagedSuspendedServiceTest extends ManagementTestSupport { assertMockEndpointsSatisfied(); await().atMost(2, TimeUnit.SECONDS).untilAsserted(() -> { - // now its suspended by the policy - Boolean bool = (Boolean) mbeanServer.getAttribute(on, "Suspended"); - assertEquals(true, bool.booleanValue()); + // now its suspended by the policy + Boolean bool = (Boolean) mbeanServer.getAttribute(on, "Suspended"); + assertEquals(true, bool.booleanValue()); }); // the route is suspended by the policy so we should only receive one http://git-wip-us.apache.org/repos/asf/camel/blob/3cd99d0a/camel-core/src/test/java/org/apache/camel/processor/MultiCastParallelAndStreamCachingTest.java ---------------------------------------------------------------------- diff --git a/camel-core/src/test/java/org/apache/camel/processor/MultiCastParallelAndStreamCachingTest.java b/camel-core/src/test/java/org/apache/camel/processor/MultiCastParallelAndStreamCachingTest.java index 9363987..4f883cb 100644 --- a/camel-core/src/test/java/org/apache/camel/processor/MultiCastParallelAndStreamCachingTest.java +++ b/camel-core/src/test/java/org/apache/camel/processor/MultiCastParallelAndStreamCachingTest.java @@ -76,7 +76,7 @@ public class MultiCastParallelAndStreamCachingTest extends ContextTestSupport { if (withSleepTime) { // simulate some processing in order to get easier concurrency effects - Thread.sleep(900); + Thread.sleep(50); } Object body = exchange.getIn().getBody(); if (body instanceof InputStream) { http://git-wip-us.apache.org/repos/asf/camel/blob/3cd99d0a/camel-core/src/test/java/org/apache/camel/processor/ThrottlerMethodCallTest.java ---------------------------------------------------------------------- diff --git a/camel-core/src/test/java/org/apache/camel/processor/ThrottlerMethodCallTest.java b/camel-core/src/test/java/org/apache/camel/processor/ThrottlerMethodCallTest.java index d4b8a09..a815cff 100644 --- a/camel-core/src/test/java/org/apache/camel/processor/ThrottlerMethodCallTest.java +++ b/camel-core/src/test/java/org/apache/camel/processor/ThrottlerMethodCallTest.java @@ -23,14 +23,14 @@ import org.apache.camel.ContextTestSupport; import org.apache.camel.builder.RouteBuilder; import org.apache.camel.component.mock.MockEndpoint; import org.apache.camel.impl.JndiRegistry; - +import org.apache.camel.util.StopWatch; /** * @version */ public class ThrottlerMethodCallTest extends ContextTestSupport { - private static final int INTERVAL = 500; - protected int messageCount = 9; + private static final int INTERVAL = 100; + protected int messageCount = 10; protected boolean canTest() { // skip test on windows as it does not run well there @@ -44,8 +44,8 @@ public class ThrottlerMethodCallTest extends ContextTestSupport { return jndi; } - public long getMessagesPerSecond() { - return 1; + public long getMessagesPerInterval() { + return 3; } public void testConfigurationWithMethodCallExpression() throws Exception { @@ -58,7 +58,7 @@ public class ThrottlerMethodCallTest extends ContextTestSupport { ExecutorService executor = Executors.newFixedThreadPool(messageCount); - long start = System.currentTimeMillis(); + StopWatch watch = new StopWatch(); for (int i = 0; i < messageCount; i++) { executor.execute(new Runnable() { public void run() { @@ -70,11 +70,9 @@ public class ThrottlerMethodCallTest extends ContextTestSupport { // 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); + // should take a little time + assertTrue(watch.taken() > 100); + executor.shutdownNow(); } @@ -82,7 +80,7 @@ public class ThrottlerMethodCallTest extends ContextTestSupport { return new RouteBuilder() { public void configure() { from("direct:expressionMethod") - .throttle(method("myBean", "getMessagesPerSecond")).timePeriodMillis(INTERVAL) + .throttle(method("myBean", "getMessagesPerInterval")).timePeriodMillis(INTERVAL) .to("log:result", "mock:result"); } }; http://git-wip-us.apache.org/repos/asf/camel/blob/3cd99d0a/camel-core/src/test/java/org/apache/camel/processor/ThrottlingExceptionRoutePolicyHalfOpenHandlerSedaTest.java ---------------------------------------------------------------------- diff --git a/camel-core/src/test/java/org/apache/camel/processor/ThrottlingExceptionRoutePolicyHalfOpenHandlerSedaTest.java b/camel-core/src/test/java/org/apache/camel/processor/ThrottlingExceptionRoutePolicyHalfOpenHandlerSedaTest.java index 625989a..44b89c5 100644 --- a/camel-core/src/test/java/org/apache/camel/processor/ThrottlingExceptionRoutePolicyHalfOpenHandlerSedaTest.java +++ b/camel-core/src/test/java/org/apache/camel/processor/ThrottlingExceptionRoutePolicyHalfOpenHandlerSedaTest.java @@ -18,6 +18,7 @@ package org.apache.camel.processor; import java.util.Arrays; import java.util.List; +import java.util.concurrent.TimeUnit; import org.apache.camel.ContextTestSupport; import org.apache.camel.Exchange; @@ -26,15 +27,18 @@ import org.apache.camel.builder.RouteBuilder; import org.apache.camel.component.mock.MockEndpoint; import org.apache.camel.impl.ThrottlingExceptionHalfOpenHandler; import org.apache.camel.impl.ThrottlingExceptionRoutePolicy; +import org.apache.camel.support.ServiceSupport; import org.junit.Before; import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import static org.awaitility.Awaitility.await; + public class ThrottlingExceptionRoutePolicyHalfOpenHandlerSedaTest extends ContextTestSupport { private static Logger log = LoggerFactory.getLogger(ThrottlingExceptionRoutePolicyHalfOpenHandlerSedaTest.class); - private String url = "seda:foo?concurrentConsumers=20"; + private String url = "seda:foo?concurrentConsumers=2"; private MockEndpoint result; @Before @@ -49,7 +53,7 @@ public class ThrottlingExceptionRoutePolicyHalfOpenHandlerSedaTest extends Conte @Test public void testHalfOpenCircuit() throws Exception { result.expectedMessageCount(2); - List<String> bodies = Arrays.asList(new String[]{"Message One", "Message Two"}); + List<String> bodies = Arrays.asList("Message One", "Message Two"); result.expectedBodiesReceivedInAnyOrder(bodies); result.whenAnyExchangeReceived(new Processor() { @@ -63,11 +67,12 @@ public class ThrottlingExceptionRoutePolicyHalfOpenHandlerSedaTest extends Conte // send two messages which will fail sendMessage("Message One"); sendMessage("Message Two"); - - // wait long enough to - // have the route shutdown - Thread.sleep(3000); - + + final ServiceSupport consumer = (ServiceSupport) context.getRoute("foo").getConsumer(); + + // wait long enough to have the consumer suspended + await().atMost(2, TimeUnit.SECONDS).until(consumer::isSuspended); + // send more messages // but should get there (yet) // due to open circuit @@ -79,12 +84,11 @@ public class ThrottlingExceptionRoutePolicyHalfOpenHandlerSedaTest extends Conte result.reset(); result.expectedMessageCount(2); - bodies = Arrays.asList(new String[]{"Message Three", "Message Four"}); + bodies = Arrays.asList("Message Three", "Message Four"); result.expectedBodiesReceivedInAnyOrder(bodies); - - // wait long enough for - // half open attempt - Thread.sleep(4000); + + // wait long enough to have the consumer resumed + await().atMost(2, TimeUnit.SECONDS).until(consumer::isStarted); // send message // should get through @@ -101,11 +105,11 @@ public class ThrottlingExceptionRoutePolicyHalfOpenHandlerSedaTest extends Conte public void configure() throws Exception { int threshold = 2; long failureWindow = 30; - long halfOpenAfter = 5000; + long halfOpenAfter = 250; ThrottlingExceptionRoutePolicy policy = new ThrottlingExceptionRoutePolicy(threshold, failureWindow, halfOpenAfter, null); policy.setHalfOpenHandler(new AlwaysCloseHandler()); - from(url) + from(url).routeId("foo") .routePolicy(policy) .log("${body}") .to("log:foo?groupSize=10") http://git-wip-us.apache.org/repos/asf/camel/blob/3cd99d0a/camel-core/src/test/java/org/apache/camel/processor/ThrottlingExceptionRoutePolicyHalfOpenHandlerTest.java ---------------------------------------------------------------------- diff --git a/camel-core/src/test/java/org/apache/camel/processor/ThrottlingExceptionRoutePolicyHalfOpenHandlerTest.java b/camel-core/src/test/java/org/apache/camel/processor/ThrottlingExceptionRoutePolicyHalfOpenHandlerTest.java index 6684b6e..1b5b0f3 100644 --- a/camel-core/src/test/java/org/apache/camel/processor/ThrottlingExceptionRoutePolicyHalfOpenHandlerTest.java +++ b/camel-core/src/test/java/org/apache/camel/processor/ThrottlingExceptionRoutePolicyHalfOpenHandlerTest.java @@ -18,6 +18,7 @@ package org.apache.camel.processor; import java.util.Arrays; import java.util.List; +import java.util.concurrent.TimeUnit; import org.apache.camel.ContextTestSupport; import org.apache.camel.Exchange; @@ -26,11 +27,14 @@ import org.apache.camel.builder.RouteBuilder; import org.apache.camel.component.mock.MockEndpoint; import org.apache.camel.impl.ThrottlingExceptionHalfOpenHandler; import org.apache.camel.impl.ThrottlingExceptionRoutePolicy; +import org.apache.camel.support.ServiceSupport; import org.junit.Before; import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import static org.awaitility.Awaitility.await; + public class ThrottlingExceptionRoutePolicyHalfOpenHandlerTest extends ContextTestSupport { private static Logger log = LoggerFactory.getLogger(ThrottlingExceptionRoutePolicyHalfOpenHandlerTest.class); @@ -49,7 +53,7 @@ public class ThrottlingExceptionRoutePolicyHalfOpenHandlerTest extends ContextTe @Test public void testHalfOpenCircuit() throws Exception { result.expectedMessageCount(2); - List<String> bodies = Arrays.asList(new String[]{"Message One", "Message Two"}); + List<String> bodies = Arrays.asList("Message One", "Message Two"); result.expectedBodiesReceivedInAnyOrder(bodies); result.whenAnyExchangeReceived(new Processor() { @@ -63,10 +67,11 @@ public class ThrottlingExceptionRoutePolicyHalfOpenHandlerTest extends ContextTe // send two messages which will fail sendMessage("Message One"); sendMessage("Message Two"); - - // wait long enough to - // have the route shutdown - Thread.sleep(3000); + + final ServiceSupport consumer = (ServiceSupport) context.getRoute("foo").getConsumer(); + + // wait long enough to have the consumer suspended + await().atMost(2, TimeUnit.SECONDS).until(consumer::isSuspended); // send more messages // but never should get there @@ -78,12 +83,11 @@ public class ThrottlingExceptionRoutePolicyHalfOpenHandlerTest extends ContextTe result.reset(); result.expectedMessageCount(1); - bodies = Arrays.asList(new String[]{"Message Four"}); + bodies = Arrays.asList("Message Four"); result.expectedBodiesReceivedInAnyOrder(bodies); - - // wait long enough for - // half open attempt - Thread.sleep(4000); + + // wait long enough to have the consumer resumed + await().atMost(2, TimeUnit.SECONDS).until(consumer::isStarted); // send message // should get through @@ -100,11 +104,11 @@ public class ThrottlingExceptionRoutePolicyHalfOpenHandlerTest extends ContextTe public void configure() throws Exception { int threshold = 2; long failureWindow = 30; - long halfOpenAfter = 5000; + long halfOpenAfter = 250; ThrottlingExceptionRoutePolicy policy = new ThrottlingExceptionRoutePolicy(threshold, failureWindow, halfOpenAfter, null); policy.setHalfOpenHandler(new AlwaysCloseHandler()); - from(url) + from(url).routeId("foo") .routePolicy(policy) .log("${body}") .to("log:foo?groupSize=10") http://git-wip-us.apache.org/repos/asf/camel/blob/3cd99d0a/camel-core/src/test/java/org/apache/camel/processor/ThrottlingExceptionRoutePolicyHalfOpenTest.java ---------------------------------------------------------------------- diff --git a/camel-core/src/test/java/org/apache/camel/processor/ThrottlingExceptionRoutePolicyHalfOpenTest.java b/camel-core/src/test/java/org/apache/camel/processor/ThrottlingExceptionRoutePolicyHalfOpenTest.java index 82c4f5e..8bae664 100644 --- a/camel-core/src/test/java/org/apache/camel/processor/ThrottlingExceptionRoutePolicyHalfOpenTest.java +++ b/camel-core/src/test/java/org/apache/camel/processor/ThrottlingExceptionRoutePolicyHalfOpenTest.java @@ -18,6 +18,7 @@ package org.apache.camel.processor; import java.util.Arrays; import java.util.List; +import java.util.concurrent.TimeUnit; import org.apache.camel.ContextTestSupport; import org.apache.camel.Exchange; @@ -25,11 +26,14 @@ import org.apache.camel.Processor; import org.apache.camel.builder.RouteBuilder; import org.apache.camel.component.mock.MockEndpoint; import org.apache.camel.impl.ThrottlingExceptionRoutePolicy; +import org.apache.camel.support.ServiceSupport; import org.junit.Before; import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import static org.awaitility.Awaitility.await; + public class ThrottlingExceptionRoutePolicyHalfOpenTest extends ContextTestSupport { private static Logger log = LoggerFactory.getLogger(ThrottlingExceptionRoutePolicyHalfOpenTest.class); @@ -49,7 +53,7 @@ public class ThrottlingExceptionRoutePolicyHalfOpenTest extends ContextTestSuppo public void testHalfOpenCircuit() throws Exception { result.reset(); result.expectedMessageCount(2); - List<String> bodies = Arrays.asList(new String[]{"Message One", "Message Two"}); + List<String> bodies = Arrays.asList("Message One", "Message Two"); result.expectedBodiesReceivedInAnyOrder(bodies); result.whenAnyExchangeReceived(new Processor() { @@ -63,11 +67,12 @@ public class ThrottlingExceptionRoutePolicyHalfOpenTest extends ContextTestSuppo // send two messages which will fail sendMessage("Message One"); sendMessage("Message Two"); + + final ServiceSupport consumer = (ServiceSupport) context.getRoute("foo").getConsumer(); - // wait long enough to - // have the route shutdown - Thread.sleep(3000); - + // wait long enough to have the consumer suspended + await().atMost(2, TimeUnit.SECONDS).until(consumer::isSuspended); + // send more messages // but never should get there // due to open circuit @@ -78,13 +83,12 @@ public class ThrottlingExceptionRoutePolicyHalfOpenTest extends ContextTestSuppo result.reset(); result.expectedMessageCount(1); - bodies = Arrays.asList(new String[]{"Message Four"}); + bodies = Arrays.asList("Message Four"); result.expectedBodiesReceivedInAnyOrder(bodies); - - // wait long enough for - // half open attempt - Thread.sleep(4000); - + + // wait long enough to have the consumer resumed + await().atMost(2, TimeUnit.SECONDS).until(consumer::isStarted); + // send message // should get through log.debug("sending message four"); @@ -100,10 +104,10 @@ public class ThrottlingExceptionRoutePolicyHalfOpenTest extends ContextTestSuppo public void configure() throws Exception { int threshold = 2; long failureWindow = 30; - long halfOpenAfter = 5000; + long halfOpenAfter = 250; ThrottlingExceptionRoutePolicy policy = new ThrottlingExceptionRoutePolicy(threshold, failureWindow, halfOpenAfter, null); - - from(url) + + from(url).routeId("foo") .routePolicy(policy) .to("log:foo?groupSize=10") .to("mock:result"); http://git-wip-us.apache.org/repos/asf/camel/blob/3cd99d0a/camel-core/src/test/java/org/apache/camel/processor/ThrottlingExceptionRoutePolicyTest.java ---------------------------------------------------------------------- diff --git a/camel-core/src/test/java/org/apache/camel/processor/ThrottlingExceptionRoutePolicyTest.java b/camel-core/src/test/java/org/apache/camel/processor/ThrottlingExceptionRoutePolicyTest.java index a2e982a..c944576 100644 --- a/camel-core/src/test/java/org/apache/camel/processor/ThrottlingExceptionRoutePolicyTest.java +++ b/camel-core/src/test/java/org/apache/camel/processor/ThrottlingExceptionRoutePolicyTest.java @@ -18,6 +18,7 @@ package org.apache.camel.processor; import java.util.Arrays; import java.util.List; +import java.util.concurrent.TimeUnit; import org.apache.camel.ContextTestSupport; import org.apache.camel.Exchange; @@ -26,15 +27,18 @@ import org.apache.camel.builder.RouteBuilder; import org.apache.camel.component.mock.MockEndpoint; import org.apache.camel.impl.ThrottlingExceptionHalfOpenHandler; import org.apache.camel.impl.ThrottlingExceptionRoutePolicy; +import org.apache.camel.support.ServiceSupport; import org.junit.Before; import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import static org.awaitility.Awaitility.await; + public class ThrottlingExceptionRoutePolicyTest extends ContextTestSupport { private static Logger log = LoggerFactory.getLogger(ThrottlingExceptionRoutePolicyTest.class); - private String url = "seda:foo?concurrentConsumers=20"; + private String url = "seda:foo?concurrentConsumers=2"; private MockEndpoint result; private int size = 100; @@ -63,7 +67,7 @@ public class ThrottlingExceptionRoutePolicyTest extends ContextTestSupport { public void testOpenCircuitToPreventMessageThree() throws Exception { result.reset(); result.expectedMessageCount(2); - List<String> bodies = Arrays.asList(new String[]{"Message One", "Message Two"}); + List<String> bodies = Arrays.asList("Message One", "Message Two"); result.expectedBodiesReceivedInAnyOrder(bodies); result.whenAnyExchangeReceived(new Processor() { @@ -77,19 +81,21 @@ public class ThrottlingExceptionRoutePolicyTest extends ContextTestSupport { // send two messages which will fail template.sendBody(url, "Message One"); template.sendBody(url, "Message Two"); - - // wait long enough to - // have the route shutdown - Thread.sleep(3000); + + final ServiceSupport consumer = (ServiceSupport) context.getRoute("foo").getConsumer(); + + // wait long enough to have the consumer suspended + await().atMost(2, TimeUnit.SECONDS).until(consumer::isSuspended); // send more messages // but never should get there // due to open circuit log.debug("sending message three"); template.sendBody(url, "Message Three"); - - Thread.sleep(2000); - + + // wait a little bit + result.setResultMinimumWaitTime(500); + assertMockEndpointsSatisfied(); } @@ -100,11 +106,11 @@ public class ThrottlingExceptionRoutePolicyTest extends ContextTestSupport { public void configure() throws Exception { int threshold = 2; long failureWindow = 30; - long halfOpenAfter = 5000; + long halfOpenAfter = 1000; ThrottlingExceptionRoutePolicy policy = new ThrottlingExceptionRoutePolicy(threshold, failureWindow, halfOpenAfter, null); policy.setHalfOpenHandler(new NeverCloseHandler()); - from(url) + from(url).routeId("foo") .routePolicy(policy) .log("${body}") .to("log:foo?groupSize=10") http://git-wip-us.apache.org/repos/asf/camel/blob/3cd99d0a/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/ThrottlerMethodCallTest.xml ---------------------------------------------------------------------- diff --git a/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/ThrottlerMethodCallTest.xml b/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/ThrottlerMethodCallTest.xml index 087eef4..f28c78c 100644 --- a/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/ThrottlerMethodCallTest.xml +++ b/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/ThrottlerMethodCallTest.xml @@ -30,9 +30,9 @@ <route> <from uri="direct:expressionMethod"/> - <throttle timePeriodMillis="500"> - <!-- use a java bean method call to determine how many messages to throttle per 0.5 sec --> - <method ref="myBean" method="getMessagesPerSecond"/> + <throttle timePeriodMillis="100"> + <!-- use a java bean method call to determine how many messages to throttle per 0.1 sec --> + <method ref="myBean" method="getMessagesPerInterval"/> <to uri="mock:result"/> </throttle> </route>
