You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2019/08/02 14:46:29 UTC
[camel] 01/08: CAMEL-12003: Add failFast option to mock endpoint
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit d8627e2a9ad60131cc71753ef25ae77e5ae3f51a
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Fri Aug 2 07:24:31 2019 +0200
CAMEL-12003: Add failFast option to mock endpoint
---
.../camel-mock/src/main/docs/mock-component.adoc | 3 +-
.../apache/camel/component/mock/AssertionTask.java | 31 ++++
.../apache/camel/component/mock/MockEndpoint.java | 166 ++++++++++++++-------
3 files changed, 148 insertions(+), 52 deletions(-)
diff --git a/components/camel-mock/src/main/docs/mock-component.adoc b/components/camel-mock/src/main/docs/mock-component.adoc
index 9a11ced..409a9e2 100644
--- a/components/camel-mock/src/main/docs/mock-component.adoc
+++ b/components/camel-mock/src/main/docs/mock-component.adoc
@@ -112,7 +112,7 @@ with the following path and query parameters:
|===
-=== Query Parameters (12 parameters):
+=== Query Parameters (13 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
@@ -120,6 +120,7 @@ with the following path and query parameters:
| Name | Description | Default | Type
| *assertPeriod* (producer) | Sets a grace period after which the mock endpoint will re-assert to ensure the preliminary assertion is still valid. This is used for example to assert that exactly a number of messages arrives. For example if expectedMessageCount(int) was set to 5, then the assertion is satisfied when 5 or more message arrives. To ensure that exactly 5 messages arrives, then you would need to wait a little period to ensure no further message arrives. This is what you can us [...]
| *expectedCount* (producer) | Specifies the expected number of message exchanges that should be received by this endpoint. Beware: If you want to expect that 0 messages, then take extra care, as 0 matches when the tests starts, so you need to set a assert period time to let the test run for a while to make sure there are still no messages arrived; for that use setAssertPeriod(long). An alternative is to use NotifyBuilder, and use the notifier to know when Camel is done routing some mess [...]
+| *failFast* (producer) | Sets whether assertIsSatisfied() should fail fast at the first detected failed expectation while it may otherwise wait for all expected messages to arrive before performing expectations verifications. Is by default true. Set to false to use behavior as in Camel 2.x. | false | boolean
| *lazyStartProducer* (producer) | Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during starting and cause the route to fail being started. By deferring this startup to be lazy then the startup failure can be handled during routing messages via Camel's routing error handlers. Beware that when the first message is processed then creating and [...]
| *reportGroup* (producer) | A number that is used to turn on throughput logging based on groups of the size. | | int
| *resultMinimumWaitTime* (producer) | Sets the minimum expected amount of time (in millis) the assertIsSatisfied() will wait on a latch until it is satisfied | 0 | long
diff --git a/components/camel-mock/src/main/java/org/apache/camel/component/mock/AssertionTask.java b/components/camel-mock/src/main/java/org/apache/camel/component/mock/AssertionTask.java
new file mode 100644
index 0000000..16daf49
--- /dev/null
+++ b/components/camel-mock/src/main/java/org/apache/camel/component/mock/AssertionTask.java
@@ -0,0 +1,31 @@
+/*
+ * 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.camel.component.mock;
+
+/**
+ * Assertion task that supports fail fast mode by running the assertion asap on the n'th received message.
+ */
+public abstract class AssertionTask implements Runnable {
+
+ /**
+ * Asserts on the n'th received message
+ *
+ * @param index the n'th received message
+ */
+ abstract void assertOnIndex(int index);
+
+}
diff --git a/components/camel-mock/src/main/java/org/apache/camel/component/mock/MockEndpoint.java b/components/camel-mock/src/main/java/org/apache/camel/component/mock/MockEndpoint.java
index 8a9826f..cdb42de 100644
--- a/components/camel-mock/src/main/java/org/apache/camel/component/mock/MockEndpoint.java
+++ b/components/camel-mock/src/main/java/org/apache/camel/component/mock/MockEndpoint.java
@@ -102,6 +102,7 @@ public class MockEndpoint extends DefaultEndpoint implements BrowsableEndpoint,
private volatile List<Throwable> failures;
private volatile List<Runnable> tests;
private volatile CountDownLatch latch;
+ private volatile AssertionError failFastAssertionError;
private volatile int expectedMinimumCount;
private volatile List<?> expectedBodyValues;
private volatile List<Object> actualBodyValues;
@@ -130,6 +131,8 @@ public class MockEndpoint extends DefaultEndpoint implements BrowsableEndpoint,
private int retainLast;
@UriParam(label = "producer")
private int reportGroup;
+ @UriParam(label = "producer")
+ private boolean failFast = true;
@UriParam(label = "producer,advanced", defaultValue = "true")
private boolean copyOnExchange = true;
@@ -181,7 +184,6 @@ public class MockEndpoint extends DefaultEndpoint implements BrowsableEndpoint,
}
}
-
/**
* Asserts that all the expectations on any {@link MockEndpoint} instances registered
* in the given context are valid
@@ -299,7 +301,6 @@ public class MockEndpoint extends DefaultEndpoint implements BrowsableEndpoint,
doInit();
}
-
// Testing API
// -------------------------------------------------------------------------
@@ -399,18 +400,28 @@ public class MockEndpoint extends DefaultEndpoint implements BrowsableEndpoint,
if (expectedCount != getReceivedCounter()) {
waitForCompleteLatch();
}
- assertEquals("Received message count", expectedCount, getReceivedCounter());
+ if (failFastAssertionError == null) {
+ assertEquals("Received message count", expectedCount, getReceivedCounter());
+ }
} else if (expectedMinimumCount > 0 && getReceivedCounter() < expectedMinimumCount) {
waitForCompleteLatch();
}
+ if (failFastAssertionError != null) {
+ throw failFastAssertionError;
+ }
+
if (expectedMinimumCount >= 0) {
int receivedCounter = getReceivedCounter();
assertTrue("Received message count " + receivedCounter + ", expected at least " + expectedMinimumCount, expectedMinimumCount <= receivedCounter);
}
for (Runnable test : tests) {
- test.run();
+ // skip tasks which we have already been running in fail fast mode
+ boolean skip = failFast && test instanceof AssertionTask;
+ if (!skip) {
+ test.run();
+ }
}
for (Throwable failure : failures) {
@@ -528,26 +539,31 @@ public class MockEndpoint extends DefaultEndpoint implements BrowsableEndpoint,
if (expectedHeaderValues == null) {
expectedHeaderValues = getCamelContext().getHeadersMapFactory().newMap();
// we just wants to expects to be called once
- expects(new Runnable() {
- public void run() {
- for (int i = 0; i < getReceivedExchanges().size(); i++) {
- Exchange exchange = getReceivedExchange(i);
- for (Map.Entry<String, Object> entry : expectedHeaderValues.entrySet()) {
- String key = entry.getKey();
- Object expectedValue = entry.getValue();
-
- // we accept that an expectedValue of null also means that the header may be absent
- if (expectedValue != null) {
- assertTrue("Exchange " + i + " has no headers", exchange.getIn().hasHeaders());
- boolean hasKey = exchange.getIn().getHeaders().containsKey(key);
- assertTrue("No header with name " + key + " found for message: " + i, hasKey);
- }
-
- Object actualValue = exchange.getIn().getHeader(key);
- actualValue = extractActualValue(exchange, actualValue, expectedValue);
+ expects(new AssertionTask() {
+ @Override
+ void assertOnIndex(int i) {
+ Exchange exchange = getReceivedExchange(i);
+ for (Map.Entry<String, Object> entry : expectedHeaderValues.entrySet()) {
+ String key = entry.getKey();
+ Object expectedValue = entry.getValue();
- assertEquals("Header with name " + key + " for message: " + i, expectedValue, actualValue);
+ // we accept that an expectedValue of null also means that the header may be absent
+ if (expectedValue != null) {
+ assertTrue("Exchange " + i + " has no headers", exchange.getIn().hasHeaders());
+ boolean hasKey = exchange.getIn().getHeaders().containsKey(key);
+ assertTrue("No header with name " + key + " found for message: " + i, hasKey);
}
+
+ Object actualValue = exchange.getIn().getHeader(key);
+ actualValue = extractActualValue(exchange, actualValue, expectedValue);
+
+ assertEquals("Header with name " + key + " for message: " + i, expectedValue, actualValue);
+ }
+ }
+
+ public void run() {
+ for (int i = 0; i < getReceivedExchanges().size(); i++) {
+ assertOnIndex(i);
}
}
});
@@ -617,26 +633,31 @@ public class MockEndpoint extends DefaultEndpoint implements BrowsableEndpoint,
}
expectedPropertyValues.put(name, value);
- expects(new Runnable() {
- public void run() {
- for (int i = 0; i < getReceivedExchanges().size(); i++) {
- Exchange exchange = getReceivedExchange(i);
- for (Map.Entry<String, Object> entry : expectedPropertyValues.entrySet()) {
- String key = entry.getKey();
- Object expectedValue = entry.getValue();
+ expects(new AssertionTask() {
+ @Override
+ void assertOnIndex(int i) {
+ Exchange exchange = getReceivedExchange(i);
+ for (Map.Entry<String, Object> entry : expectedPropertyValues.entrySet()) {
+ String key = entry.getKey();
+ Object expectedValue = entry.getValue();
+
+ // we accept that an expectedValue of null also means that the property may be absent
+ if (expectedValue != null) {
+ assertTrue("Exchange " + i + " has no properties", !exchange.getProperties().isEmpty());
+ boolean hasKey = exchange.getProperties().containsKey(key);
+ assertTrue("No property with name " + key + " found for message: " + i, hasKey);
+ }
- // we accept that an expectedValue of null also means that the property may be absent
- if (expectedValue != null) {
- assertTrue("Exchange " + i + " has no properties", !exchange.getProperties().isEmpty());
- boolean hasKey = exchange.getProperties().containsKey(key);
- assertTrue("No property with name " + key + " found for message: " + i, hasKey);
- }
+ Object actualValue = exchange.getProperty(key);
+ actualValue = extractActualValue(exchange, actualValue, expectedValue);
- Object actualValue = exchange.getProperty(key);
- actualValue = extractActualValue(exchange, actualValue, expectedValue);
+ assertEquals("Property with name " + key + " for message: " + i, expectedValue, actualValue);
+ }
+ }
- assertEquals("Property with name " + key + " for message: " + i, expectedValue, actualValue);
- }
+ public void run() {
+ for (int i = 0; i < getReceivedExchanges().size(); i++) {
+ assertOnIndex(i);
}
}
});
@@ -706,20 +727,25 @@ public class MockEndpoint extends DefaultEndpoint implements BrowsableEndpoint,
this.expectedBodyValues = bodies;
this.actualBodyValues = new ArrayList<>();
- expects(new Runnable() {
- public void run() {
- for (int i = 0; i < expectedBodyValues.size(); i++) {
- Exchange exchange = getReceivedExchange(i);
- assertTrue("No exchange received for counter: " + i, exchange != null);
+ expects(new AssertionTask() {
+ @Override
+ void assertOnIndex(int i) {
+ Exchange exchange = getReceivedExchange(i);
+ assertTrue("No exchange received for counter: " + i, exchange != null);
- Object expectedBody = expectedBodyValues.get(i);
- Object actualBody = null;
- if (i < actualBodyValues.size()) {
- actualBody = actualBodyValues.get(i);
- }
- actualBody = extractActualValue(exchange, actualBody, expectedBody);
+ Object expectedBody = expectedBodyValues.get(i);
+ Object actualBody = null;
+ if (i < actualBodyValues.size()) {
+ actualBody = actualBodyValues.get(i);
+ }
+ actualBody = extractActualValue(exchange, actualBody, expectedBody);
+
+ assertEquals("Body of message: " + i, expectedBody, actualBody);
+ }
- assertEquals("Body of message: " + i, expectedBody, actualBody);
+ public void run() {
+ for (int i = 0; i < expectedBodyValues.size(); i++) {
+ assertOnIndex(i);
}
}
});
@@ -1314,6 +1340,21 @@ public class MockEndpoint extends DefaultEndpoint implements BrowsableEndpoint,
this.copyOnExchange = copyOnExchange;
}
+ public boolean isFailFast() {
+ return failFast;
+ }
+
+ /**
+ * Sets whether {@link #assertIsSatisfied()} should fail fast
+ * at the first detected failed expectation while it may otherwise wait for all expected
+ * messages to arrive before performing expectations verifications.
+ *
+ * Is by default <tt>true</tt>. Set to <tt>false</tt> to use behavior as in Camel 2.x.
+ */
+ public void setFailFast(boolean failFast) {
+ this.failFast = failFast;
+ }
+
// Implementation methods
// -------------------------------------------------------------------------
protected void doInit() {
@@ -1325,6 +1366,7 @@ public class MockEndpoint extends DefaultEndpoint implements BrowsableEndpoint,
failures = new CopyOnWriteArrayList<>();
tests = new CopyOnWriteArrayList<>();
latch = null;
+ failFastAssertionError = null;
sleepForEmptyTest = 0;
resultWaitTime = 0;
resultMinimumWaitTime = 0L;
@@ -1351,6 +1393,28 @@ public class MockEndpoint extends DefaultEndpoint implements BrowsableEndpoint,
copy = ExchangeHelper.createCopy(exchange, true);
}
performAssertions(exchange, copy);
+
+ if (failFast) {
+ // fail fast mode so check n'th expectations as soon as possible
+ int index = getReceivedCounter() - 1;
+ for (Runnable test : tests) {
+ // only assertion tasks can support fail fast mode
+ if (test instanceof AssertionTask) {
+ AssertionTask task = (AssertionTask) test;
+ try {
+ log.debug("Running assertOnIndex({}) on task: {}", index, task);
+ task.assertOnIndex(index);
+ } catch (AssertionError e) {
+ failFastAssertionError = e;
+ // signal latch we are done as we are failing fast
+ log.debug("Assertion failed fast on " + index + " received exchange due to " + e.getMessage());
+ while (latch != null && latch.getCount() > 0) {
+ latch.countDown();
+ }
+ }
+ }
+ }
+ }
} catch (Throwable e) {
// must catch java.lang.Throwable as AssertionError extends java.lang.Error
failures.add(e);