001/** 002 * Licensed to the Apache Software Foundation (ASF) under one or more 003 * contributor license agreements. See the NOTICE file distributed with 004 * this work for additional information regarding copyright ownership. 005 * The ASF licenses this file to You under the Apache License, Version 2.0 006 * (the "License"); you may not use this file except in compliance with 007 * the License. You may obtain a copy of the License at 008 * 009 * http://www.apache.org/licenses/LICENSE-2.0 010 * 011 * Unless required by applicable law or agreed to in writing, software 012 * distributed under the License is distributed on an "AS IS" BASIS, 013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 014 * See the License for the specific language governing permissions and 015 * limitations under the License. 016 */ 017package org.apache.camel.component.mock; 018 019import java.io.File; 020import java.util.ArrayList; 021import java.util.Arrays; 022import java.util.Collection; 023import java.util.Date; 024import java.util.HashMap; 025import java.util.List; 026import java.util.Map; 027import java.util.Set; 028import java.util.concurrent.CopyOnWriteArrayList; 029import java.util.concurrent.CopyOnWriteArraySet; 030import java.util.concurrent.CountDownLatch; 031import java.util.concurrent.TimeUnit; 032 033import org.apache.camel.AsyncCallback; 034import org.apache.camel.CamelContext; 035import org.apache.camel.Component; 036import org.apache.camel.Consumer; 037import org.apache.camel.Endpoint; 038import org.apache.camel.Exchange; 039import org.apache.camel.ExchangePattern; 040import org.apache.camel.Expression; 041import org.apache.camel.Handler; 042import org.apache.camel.Message; 043import org.apache.camel.Predicate; 044import org.apache.camel.Processor; 045import org.apache.camel.Producer; 046import org.apache.camel.builder.ProcessorBuilder; 047import org.apache.camel.impl.DefaultAsyncProducer; 048import org.apache.camel.impl.DefaultEndpoint; 049import org.apache.camel.impl.InterceptSendToEndpoint; 050import org.apache.camel.spi.BrowsableEndpoint; 051import org.apache.camel.spi.Metadata; 052import org.apache.camel.spi.UriEndpoint; 053import org.apache.camel.spi.UriParam; 054import org.apache.camel.spi.UriPath; 055import org.apache.camel.util.CamelContextHelper; 056import org.apache.camel.util.ExchangeHelper; 057import org.apache.camel.util.ExpressionComparator; 058import org.apache.camel.util.FileUtil; 059import org.apache.camel.util.ObjectHelper; 060import org.apache.camel.util.StopWatch; 061import org.slf4j.Logger; 062import org.slf4j.LoggerFactory; 063 064/** 065 * The mock component is used for testing routes and mediation rules using mocks. 066 * <p/> 067 * A Mock endpoint which provides a literate, fluent API for testing routes 068 * using a <a href="http://jmock.org/">JMock style</a> API. 069 * <p/> 070 * The mock endpoint have two set of methods 071 * <ul> 072 * <li>expectedXXX or expectsXXX - To set pre conditions, before the test is executed</li> 073 * <li>assertXXX - To assert assertions, after the test has been executed</li> 074 * </ul> 075 * Its <b>important</b> to know the difference between the two set. The former is used to 076 * set expectations before the test is being started (eg before the mock receives messages). 077 * The latter is used after the test has been executed, to verify the expectations; or 078 * other assertions which you can perform after the test has been completed. 079 * <p/> 080 * <b>Beware:</b> If you want to expect a mock does not receive any messages, by calling 081 * {@link #setExpectedMessageCount(int)} with <tt>0</tt>, then take extra care, 082 * as <tt>0</tt> matches when the tests starts, so you need to set a assert period time 083 * to let the test run for a while to make sure there are still no messages arrived; for 084 * that use {@link #setAssertPeriod(long)}. 085 * An alternative is to use <a href="http://camel.apache.org/notifybuilder.html">NotifyBuilder</a>, and use the notifier 086 * to know when Camel is done routing some messages, before you call the {@link #assertIsSatisfied()} method on the mocks. 087 * This allows you to not use a fixed assert period, to speedup testing times. 088 * <p/> 089 * <b>Important:</b> If using {@link #expectedMessageCount(int)} and also {@link #expectedBodiesReceived(java.util.List)} or 090 * {@link #expectedHeaderReceived(String, Object)} then the latter overrides the number of expected message based on the 091 * number of values provided in the bodies/headers. 092 * 093 * @version 094 */ 095@UriEndpoint(firstVersion = "1.0.0", scheme = "mock", title = "Mock", syntax = "mock:name", producerOnly = true, label = "core,testing", lenientProperties = true) 096public class MockEndpoint extends DefaultEndpoint implements BrowsableEndpoint { 097 private static final Logger LOG = LoggerFactory.getLogger(MockEndpoint.class); 098 // must be volatile so changes is visible between the thread which performs the assertions 099 // and the threads which process the exchanges when routing messages in Camel 100 protected volatile Processor reporter; 101 102 private volatile Processor defaultProcessor; 103 private volatile Map<Integer, Processor> processors; 104 private volatile List<Exchange> receivedExchanges; 105 private volatile List<Throwable> failures; 106 private volatile List<Runnable> tests; 107 private volatile CountDownLatch latch; 108 private volatile int expectedMinimumCount; 109 private volatile List<?> expectedBodyValues; 110 private volatile List<Object> actualBodyValues; 111 private volatile Map<String, Object> expectedHeaderValues; 112 private volatile Map<String, Object> actualHeaderValues; 113 private volatile Map<String, Object> expectedPropertyValues; 114 private volatile Map<String, Object> actualPropertyValues; 115 116 private volatile int counter; 117 118 @UriPath(description = "Name of mock endpoint") @Metadata(required = "true") 119 private String name; 120 @UriParam(label = "producer", defaultValue = "-1") 121 private int expectedCount; 122 @UriParam(label = "producer", defaultValue = "0") 123 private long sleepForEmptyTest; 124 @UriParam(label = "producer", defaultValue = "0") 125 private long resultWaitTime; 126 @UriParam(label = "producer", defaultValue = "0") 127 private long resultMinimumWaitTime; 128 @UriParam(label = "producer", defaultValue = "0") 129 private long assertPeriod; 130 @UriParam(label = "producer", defaultValue = "-1") 131 private int retainFirst; 132 @UriParam(label = "producer", defaultValue = "-1") 133 private int retainLast; 134 @UriParam(label = "producer") 135 private int reportGroup; 136 @UriParam(label = "producer,advanced", defaultValue = "true") 137 private boolean copyOnExchange = true; 138 139 public MockEndpoint(String endpointUri, Component component) { 140 super(endpointUri, component); 141 init(); 142 } 143 144 @Deprecated 145 public MockEndpoint(String endpointUri) { 146 super(endpointUri); 147 init(); 148 } 149 150 public MockEndpoint() { 151 this(null); 152 } 153 154 /** 155 * A helper method to resolve the mock endpoint of the given URI on the given context 156 * 157 * @param context the camel context to try resolve the mock endpoint from 158 * @param uri the uri of the endpoint to resolve 159 * @return the endpoint 160 */ 161 public static MockEndpoint resolve(CamelContext context, String uri) { 162 return CamelContextHelper.getMandatoryEndpoint(context, uri, MockEndpoint.class); 163 } 164 165 public static void assertWait(long timeout, TimeUnit unit, MockEndpoint... endpoints) throws InterruptedException { 166 long start = System.currentTimeMillis(); 167 long left = unit.toMillis(timeout); 168 long end = start + left; 169 for (MockEndpoint endpoint : endpoints) { 170 if (!endpoint.await(left, TimeUnit.MILLISECONDS)) { 171 throw new AssertionError("Timeout waiting for endpoints to receive enough messages. " + endpoint.getEndpointUri() + " timed out."); 172 } 173 left = end - System.currentTimeMillis(); 174 if (left <= 0) { 175 left = 0; 176 } 177 } 178 } 179 180 public static void assertIsSatisfied(long timeout, TimeUnit unit, MockEndpoint... endpoints) throws InterruptedException { 181 assertWait(timeout, unit, endpoints); 182 for (MockEndpoint endpoint : endpoints) { 183 endpoint.assertIsSatisfied(); 184 } 185 } 186 187 public static void assertIsSatisfied(MockEndpoint... endpoints) throws InterruptedException { 188 for (MockEndpoint endpoint : endpoints) { 189 endpoint.assertIsSatisfied(); 190 } 191 } 192 193 194 /** 195 * Asserts that all the expectations on any {@link MockEndpoint} instances registered 196 * in the given context are valid 197 * 198 * @param context the camel context used to find all the available endpoints to be asserted 199 */ 200 public static void assertIsSatisfied(CamelContext context) throws InterruptedException { 201 ObjectHelper.notNull(context, "camelContext"); 202 Collection<Endpoint> endpoints = context.getEndpoints(); 203 for (Endpoint endpoint : endpoints) { 204 // if the endpoint was intercepted we should get the delegate 205 if (endpoint instanceof InterceptSendToEndpoint) { 206 endpoint = ((InterceptSendToEndpoint) endpoint).getDelegate(); 207 } 208 if (endpoint instanceof MockEndpoint) { 209 MockEndpoint mockEndpoint = (MockEndpoint) endpoint; 210 mockEndpoint.assertIsSatisfied(); 211 } 212 } 213 } 214 215 /** 216 * Asserts that all the expectations on any {@link MockEndpoint} instances registered 217 * in the given context are valid 218 * 219 * @param context the camel context used to find all the available endpoints to be asserted 220 * @param timeout timeout 221 * @param unit time unit 222 */ 223 public static void assertIsSatisfied(CamelContext context, long timeout, TimeUnit unit) throws InterruptedException { 224 ObjectHelper.notNull(context, "camelContext"); 225 ObjectHelper.notNull(unit, "unit"); 226 Collection<Endpoint> endpoints = context.getEndpoints(); 227 long millis = unit.toMillis(timeout); 228 for (Endpoint endpoint : endpoints) { 229 // if the endpoint was intercepted we should get the delegate 230 if (endpoint instanceof InterceptSendToEndpoint) { 231 endpoint = ((InterceptSendToEndpoint) endpoint).getDelegate(); 232 } 233 if (endpoint instanceof MockEndpoint) { 234 MockEndpoint mockEndpoint = (MockEndpoint) endpoint; 235 mockEndpoint.setResultWaitTime(millis); 236 mockEndpoint.assertIsSatisfied(); 237 } 238 } 239 } 240 241 /** 242 * Sets the assert period on all the expectations on any {@link MockEndpoint} instances registered 243 * in the given context. 244 * 245 * @param context the camel context used to find all the available endpoints 246 * @param period the period in millis 247 */ 248 public static void setAssertPeriod(CamelContext context, long period) { 249 ObjectHelper.notNull(context, "camelContext"); 250 Collection<Endpoint> endpoints = context.getEndpoints(); 251 for (Endpoint endpoint : endpoints) { 252 // if the endpoint was intercepted we should get the delegate 253 if (endpoint instanceof InterceptSendToEndpoint) { 254 endpoint = ((InterceptSendToEndpoint) endpoint).getDelegate(); 255 } 256 if (endpoint instanceof MockEndpoint) { 257 MockEndpoint mockEndpoint = (MockEndpoint) endpoint; 258 mockEndpoint.setAssertPeriod(period); 259 } 260 } 261 } 262 263 /** 264 * Reset all mock endpoints 265 * 266 * @param context the camel context used to find all the available endpoints to reset 267 */ 268 public static void resetMocks(CamelContext context) { 269 ObjectHelper.notNull(context, "camelContext"); 270 Collection<Endpoint> endpoints = context.getEndpoints(); 271 for (Endpoint endpoint : endpoints) { 272 // if the endpoint was intercepted we should get the delegate 273 if (endpoint instanceof InterceptSendToEndpoint) { 274 endpoint = ((InterceptSendToEndpoint) endpoint).getDelegate(); 275 } 276 if (endpoint instanceof MockEndpoint) { 277 MockEndpoint mockEndpoint = (MockEndpoint) endpoint; 278 mockEndpoint.reset(); 279 } 280 } 281 } 282 283 public static void expectsMessageCount(int count, MockEndpoint... endpoints) throws InterruptedException { 284 for (MockEndpoint endpoint : endpoints) { 285 endpoint.setExpectedMessageCount(count); 286 } 287 } 288 289 public List<Exchange> getExchanges() { 290 return getReceivedExchanges(); 291 } 292 293 public Consumer createConsumer(Processor processor) throws Exception { 294 throw new UnsupportedOperationException("You cannot consume from this endpoint"); 295 } 296 297 public Producer createProducer() throws Exception { 298 return new DefaultAsyncProducer(this) { 299 public boolean process(Exchange exchange, AsyncCallback callback) { 300 onExchange(exchange); 301 callback.done(true); 302 return true; 303 } 304 }; 305 } 306 307 public void reset() { 308 init(); 309 } 310 311 312 // Testing API 313 // ------------------------------------------------------------------------- 314 315 /** 316 * Handles the incoming exchange. 317 * <p/> 318 * This method turns this mock endpoint into a bean which you can use 319 * in the Camel routes, which allows you to inject MockEndpoint as beans 320 * in your routes and use the features of the mock to control the bean. 321 * 322 * @param exchange the exchange 323 * @throws Exception can be thrown 324 */ 325 @Handler 326 public void handle(Exchange exchange) throws Exception { 327 onExchange(exchange); 328 } 329 330 /** 331 * Set the processor that will be invoked when the index 332 * message is received. 333 */ 334 public void whenExchangeReceived(int index, Processor processor) { 335 this.processors.put(index, processor); 336 } 337 338 /** 339 * Set the processor that will be invoked when the some message 340 * is received. 341 * 342 * This processor could be overwritten by 343 * {@link #whenExchangeReceived(int, Processor)} method. 344 */ 345 public void whenAnyExchangeReceived(Processor processor) { 346 this.defaultProcessor = processor; 347 } 348 349 /** 350 * Set the expression which value will be set to the message body 351 * @param expression which is use to set the message body 352 */ 353 public void returnReplyBody(Expression expression) { 354 this.defaultProcessor = ProcessorBuilder.setBody(expression); 355 } 356 357 /** 358 * Set the expression which value will be set to the message header 359 * @param headerName that will be set value 360 * @param expression which is use to set the message header 361 */ 362 public void returnReplyHeader(String headerName, Expression expression) { 363 this.defaultProcessor = ProcessorBuilder.setHeader(headerName, expression); 364 } 365 366 367 /** 368 * Validates that all the available expectations on this endpoint are 369 * satisfied; or throw an exception 370 */ 371 public void assertIsSatisfied() throws InterruptedException { 372 assertIsSatisfied(sleepForEmptyTest); 373 } 374 375 /** 376 * Validates that all the available expectations on this endpoint are 377 * satisfied; or throw an exception 378 * 379 * @param timeoutForEmptyEndpoints the timeout in milliseconds that we 380 * should wait for the test to be true 381 */ 382 public void assertIsSatisfied(long timeoutForEmptyEndpoints) throws InterruptedException { 383 LOG.info("Asserting: {} is satisfied", this); 384 doAssertIsSatisfied(timeoutForEmptyEndpoints); 385 if (assertPeriod > 0) { 386 // if an assert period was set then re-assert again to ensure the assertion is still valid 387 Thread.sleep(assertPeriod); 388 LOG.info("Re-asserting: {} is satisfied after {} millis", this, assertPeriod); 389 // do not use timeout when we re-assert 390 doAssertIsSatisfied(0); 391 } 392 } 393 394 protected void doAssertIsSatisfied(long timeoutForEmptyEndpoints) throws InterruptedException { 395 if (expectedCount == 0) { 396 if (timeoutForEmptyEndpoints > 0) { 397 LOG.debug("Sleeping for: {} millis to check there really are no messages received", timeoutForEmptyEndpoints); 398 Thread.sleep(timeoutForEmptyEndpoints); 399 } 400 assertEquals("Received message count", expectedCount, getReceivedCounter()); 401 } else if (expectedCount > 0) { 402 if (expectedCount != getReceivedCounter()) { 403 waitForCompleteLatch(); 404 } 405 assertEquals("Received message count", expectedCount, getReceivedCounter()); 406 } else if (expectedMinimumCount > 0 && getReceivedCounter() < expectedMinimumCount) { 407 waitForCompleteLatch(); 408 } 409 410 if (expectedMinimumCount >= 0) { 411 int receivedCounter = getReceivedCounter(); 412 assertTrue("Received message count " + receivedCounter + ", expected at least " + expectedMinimumCount, expectedMinimumCount <= receivedCounter); 413 } 414 415 for (Runnable test : tests) { 416 test.run(); 417 } 418 419 for (Throwable failure : failures) { 420 if (failure != null) { 421 LOG.error("Caught on " + getEndpointUri() + " Exception: " + failure, failure); 422 fail("Failed due to caught exception: " + failure); 423 } 424 } 425 } 426 427 /** 428 * Validates that the assertions fail on this endpoint 429 */ 430 public void assertIsNotSatisfied() throws InterruptedException { 431 boolean failed = false; 432 try { 433 assertIsSatisfied(); 434 // did not throw expected error... fail! 435 failed = true; 436 } catch (AssertionError e) { 437 LOG.info("Caught expected failure: " + e); 438 } 439 if (failed) { 440 // fail() throws the AssertionError to indicate the test failed. 441 fail("Expected assertion failure but test succeeded!"); 442 } 443 } 444 445 /** 446 * Validates that the assertions fail on this endpoint 447 448 * @param timeoutForEmptyEndpoints the timeout in milliseconds that we 449 * should wait for the test to be true 450 */ 451 public void assertIsNotSatisfied(long timeoutForEmptyEndpoints) throws InterruptedException { 452 boolean failed = false; 453 try { 454 assertIsSatisfied(timeoutForEmptyEndpoints); 455 // did not throw expected error... fail! 456 failed = true; 457 } catch (AssertionError e) { 458 LOG.info("Caught expected failure: " + e); 459 } 460 if (failed) { 461 // fail() throws the AssertionError to indicate the test failed. 462 fail("Expected assertion failure but test succeeded!"); 463 } 464 } 465 466 /** 467 * Specifies the expected number of message exchanges that should be 468 * received by this endpoint 469 * 470 * If you want to assert that <b>exactly</b> n messages arrives to this mock 471 * endpoint, then see also the {@link #setAssertPeriod(long)} method for further details. 472 * 473 * @param expectedCount the number of message exchanges that should be 474 * expected by this endpoint 475 * @see #setAssertPeriod(long) 476 */ 477 public void expectedMessageCount(int expectedCount) { 478 setExpectedMessageCount(expectedCount); 479 } 480 481 /** 482 * Sets a grace period after which the mock endpoint will re-assert 483 * to ensure the preliminary assertion is still valid. 484 * <p/> 485 * This is used for example to assert that <b>exactly</b> a number of messages 486 * arrives. For example if {@link #expectedMessageCount(int)} was set to 5, then 487 * the assertion is satisfied when 5 or more message arrives. To ensure that 488 * exactly 5 messages arrives, then you would need to wait a little period 489 * to ensure no further message arrives. This is what you can use this 490 * {@link #setAssertPeriod(long)} method for. 491 * <p/> 492 * By default this period is disabled. 493 * 494 * @param period grace period in millis 495 */ 496 public void setAssertPeriod(long period) { 497 this.assertPeriod = period; 498 } 499 500 /** 501 * Specifies the minimum number of expected message exchanges that should be 502 * received by this endpoint 503 * 504 * @param expectedCount the number of message exchanges that should be 505 * expected by this endpoint 506 */ 507 public void expectedMinimumMessageCount(int expectedCount) { 508 setMinimumExpectedMessageCount(expectedCount); 509 } 510 511 /** 512 * Sets an expectation that the given header name & value are received by this endpoint 513 * <p/> 514 * You can set multiple expectations for different header names. 515 * If you set a value of <tt>null</tt> that means we accept either the header is absent, or its value is <tt>null</tt> 516 * <p/> 517 * <b>Important:</b> This overrides any previous set value using {@link #expectedMessageCount(int)} 518 */ 519 public void expectedHeaderReceived(final String name, final Object value) { 520 if (expectedCount == -1) { 521 expectedMessageCount(1); 522 } 523 if (expectedHeaderValues == null) { 524 expectedHeaderValues = getCamelContext().getHeadersMapFactory().newMap(); 525 // we just wants to expects to be called once 526 expects(new Runnable() { 527 public void run() { 528 for (int i = 0; i < getReceivedExchanges().size(); i++) { 529 Exchange exchange = getReceivedExchange(i); 530 for (Map.Entry<String, Object> entry : expectedHeaderValues.entrySet()) { 531 String key = entry.getKey(); 532 Object expectedValue = entry.getValue(); 533 534 // we accept that an expectedValue of null also means that the header may be absent 535 if (expectedValue != null) { 536 assertTrue("Exchange " + i + " has no headers", exchange.getIn().hasHeaders()); 537 boolean hasKey = exchange.getIn().getHeaders().containsKey(key); 538 assertTrue("No header with name " + key + " found for message: " + i, hasKey); 539 } 540 541 Object actualValue = exchange.getIn().getHeader(key); 542 actualValue = extractActualValue(exchange, actualValue, expectedValue); 543 544 assertEquals("Header with name " + key + " for message: " + i, expectedValue, actualValue); 545 } 546 } 547 } 548 }); 549 } 550 expectedHeaderValues.put(name, value); 551 } 552 553 /** 554 * Adds an expectation that the given header values are received by this 555 * endpoint in any order. 556 * <p/> 557 * <b>Important:</b> The number of values must match the expected number of messages, so if you expect 3 messages, then 558 * there must be 3 values. 559 * <p/> 560 * <b>Important:</b> This overrides any previous set value using {@link #expectedMessageCount(int)} 561 */ 562 public void expectedHeaderValuesReceivedInAnyOrder(final String name, final List<?> values) { 563 expectedMessageCount(values.size()); 564 565 expects(new Runnable() { 566 public void run() { 567 // these are the expected values to find 568 final Set<Object> actualHeaderValues = new CopyOnWriteArraySet<Object>(values); 569 570 for (int i = 0; i < getReceivedExchanges().size(); i++) { 571 Exchange exchange = getReceivedExchange(i); 572 573 Object actualValue = exchange.getIn().getHeader(name); 574 for (Object expectedValue : actualHeaderValues) { 575 actualValue = extractActualValue(exchange, actualValue, expectedValue); 576 // remove any found values 577 actualHeaderValues.remove(actualValue); 578 } 579 } 580 581 // should be empty, as we should find all the values 582 assertTrue("Expected " + values.size() + " headers with key[" + name + "], received " + (values.size() - actualHeaderValues.size()) 583 + " headers. Expected header values: " + actualHeaderValues, actualHeaderValues.isEmpty()); 584 } 585 }); 586 } 587 588 /** 589 * Adds an expectation that the given header values are received by this 590 * endpoint in any order 591 * <p/> 592 * <b>Important:</b> The number of values must match the expected number of messages, so if you expect 3 messages, then 593 * there must be 3 values. 594 * <p/> 595 * <b>Important:</b> This overrides any previous set value using {@link #expectedMessageCount(int)} 596 */ 597 public void expectedHeaderValuesReceivedInAnyOrder(String name, Object... values) { 598 List<Object> valueList = new ArrayList<Object>(); 599 valueList.addAll(Arrays.asList(values)); 600 expectedHeaderValuesReceivedInAnyOrder(name, valueList); 601 } 602 603 /** 604 * Sets an expectation that the given property name & value are received by this endpoint 605 * <p/> 606 * You can set multiple expectations for different property names. 607 * If you set a value of <tt>null</tt> that means we accept either the property is absent, or its value is <tt>null</tt> 608 */ 609 public void expectedPropertyReceived(final String name, final Object value) { 610 if (expectedPropertyValues == null) { 611 expectedPropertyValues = new HashMap<String, Object>(); 612 } 613 expectedPropertyValues.put(name, value); 614 615 expects(new Runnable() { 616 public void run() { 617 for (int i = 0; i < getReceivedExchanges().size(); i++) { 618 Exchange exchange = getReceivedExchange(i); 619 for (Map.Entry<String, Object> entry : expectedPropertyValues.entrySet()) { 620 String key = entry.getKey(); 621 Object expectedValue = entry.getValue(); 622 623 // we accept that an expectedValue of null also means that the property may be absent 624 if (expectedValue != null) { 625 assertTrue("Exchange " + i + " has no properties", !exchange.getProperties().isEmpty()); 626 boolean hasKey = exchange.getProperties().containsKey(key); 627 assertTrue("No property with name " + key + " found for message: " + i, hasKey); 628 } 629 630 Object actualValue = exchange.getProperty(key); 631 actualValue = extractActualValue(exchange, actualValue, expectedValue); 632 633 assertEquals("Property with name " + key + " for message: " + i, expectedValue, actualValue); 634 } 635 } 636 } 637 }); 638 } 639 640 /** 641 * Adds an expectation that the given property values are received by this 642 * endpoint in any order. 643 * <p/> 644 * <b>Important:</b> The number of values must match the expected number of messages, so if you expect 3 messages, then 645 * there must be 3 values. 646 * <p/> 647 * <b>Important:</b> This overrides any previous set value using {@link #expectedMessageCount(int)} 648 */ 649 public void expectedPropertyValuesReceivedInAnyOrder(final String name, final List<?> values) { 650 expectedMessageCount(values.size()); 651 652 expects(new Runnable() { 653 public void run() { 654 // these are the expected values to find 655 final Set<Object> actualPropertyValues = new CopyOnWriteArraySet<Object>(values); 656 657 for (int i = 0; i < getReceivedExchanges().size(); i++) { 658 Exchange exchange = getReceivedExchange(i); 659 660 Object actualValue = exchange.getProperty(name); 661 for (Object expectedValue : actualPropertyValues) { 662 actualValue = extractActualValue(exchange, actualValue, expectedValue); 663 // remove any found values 664 actualPropertyValues.remove(actualValue); 665 } 666 } 667 668 // should be empty, as we should find all the values 669 assertTrue("Expected " + values.size() + " properties with key[" + name + "], received " + (values.size() - actualPropertyValues.size()) 670 + " properties. Expected property values: " + actualPropertyValues, actualPropertyValues.isEmpty()); 671 } 672 }); 673 } 674 675 /** 676 * Adds an expectation that the given property values are received by this 677 * endpoint in any order 678 * <p/> 679 * <b>Important:</b> The number of values must match the expected number of messages, so if you expect 3 messages, then 680 * there must be 3 values. 681 * <p/> 682 * <b>Important:</b> This overrides any previous set value using {@link #expectedMessageCount(int)} 683 */ 684 public void expectedPropertyValuesReceivedInAnyOrder(String name, Object... values) { 685 List<Object> valueList = new ArrayList<Object>(); 686 valueList.addAll(Arrays.asList(values)); 687 expectedPropertyValuesReceivedInAnyOrder(name, valueList); 688 } 689 690 /** 691 * Adds an expectation that the given body values are received by this 692 * endpoint in the specified order 693 * <p/> 694 * <b>Important:</b> The number of values must match the expected number of messages, so if you expect 3 messages, then 695 * there must be 3 values. 696 * <p/> 697 * <b>Important:</b> This overrides any previous set value using {@link #expectedMessageCount(int)} 698 */ 699 public void expectedBodiesReceived(final List<?> bodies) { 700 expectedMessageCount(bodies.size()); 701 this.expectedBodyValues = bodies; 702 this.actualBodyValues = new ArrayList<Object>(); 703 704 expects(new Runnable() { 705 public void run() { 706 for (int i = 0; i < expectedBodyValues.size(); i++) { 707 Exchange exchange = getReceivedExchange(i); 708 assertTrue("No exchange received for counter: " + i, exchange != null); 709 710 Object expectedBody = expectedBodyValues.get(i); 711 Object actualBody = null; 712 if (i < actualBodyValues.size()) { 713 actualBody = actualBodyValues.get(i); 714 } 715 actualBody = extractActualValue(exchange, actualBody, expectedBody); 716 717 assertEquals("Body of message: " + i, expectedBody, actualBody); 718 } 719 } 720 }); 721 } 722 723 private Object extractActualValue(Exchange exchange, Object actualValue, Object expectedValue) { 724 if (actualValue == null) { 725 return null; 726 } 727 728 if (actualValue instanceof Expression) { 729 Class clazz = Object.class; 730 if (expectedValue != null) { 731 clazz = expectedValue.getClass(); 732 } 733 actualValue = ((Expression)actualValue).evaluate(exchange, clazz); 734 } else if (actualValue instanceof Predicate) { 735 actualValue = ((Predicate)actualValue).matches(exchange); 736 } else if (expectedValue != null) { 737 String from = actualValue.getClass().getName(); 738 String to = expectedValue.getClass().getName(); 739 actualValue = getCamelContext().getTypeConverter().convertTo(expectedValue.getClass(), exchange, actualValue); 740 assertTrue("There is no type conversion possible from " + from + " to " + to, actualValue != null); 741 } 742 return actualValue; 743 } 744 745 /** 746 * Sets an expectation that the given predicates matches the received messages by this endpoint 747 */ 748 public void expectedMessagesMatches(Predicate... predicates) { 749 for (int i = 0; i < predicates.length; i++) { 750 final int messageIndex = i; 751 final Predicate predicate = predicates[i]; 752 final AssertionClause clause = new AssertionClause(this) { 753 public void run() { 754 addPredicate(predicate); 755 applyAssertionOn(MockEndpoint.this, messageIndex, assertExchangeReceived(messageIndex)); 756 } 757 }; 758 expects(clause); 759 } 760 } 761 762 /** 763 * Sets an expectation that the given body values are received by this endpoint 764 * <p/> 765 * <b>Important:</b> The number of bodies must match the expected number of messages, so if you expect 3 messages, then 766 * there must be 3 bodies. 767 * <p/> 768 * <b>Important:</b> This overrides any previous set value using {@link #expectedMessageCount(int)} 769 */ 770 public void expectedBodiesReceived(Object... bodies) { 771 List<Object> bodyList = new ArrayList<Object>(); 772 bodyList.addAll(Arrays.asList(bodies)); 773 expectedBodiesReceived(bodyList); 774 } 775 776 /** 777 * Adds an expectation that the given body value are received by this endpoint 778 */ 779 public AssertionClause expectedBodyReceived() { 780 expectedMessageCount(1); 781 final AssertionClause clause = new AssertionClause(this) { 782 public void run() { 783 Exchange exchange = getReceivedExchange(0); 784 assertTrue("No exchange received for counter: " + 0, exchange != null); 785 786 Object actualBody = exchange.getIn().getBody(); 787 Expression exp = createExpression(getCamelContext()); 788 Object expectedBody = exp.evaluate(exchange, Object.class); 789 790 assertEquals("Body of message: " + 0, expectedBody, actualBody); 791 } 792 }; 793 expects(clause); 794 return clause; 795 } 796 797 /** 798 * Adds an expectation that the given body values are received by this 799 * endpoint in any order 800 * <p/> 801 * <b>Important:</b> The number of bodies must match the expected number of messages, so if you expect 3 messages, then 802 * there must be 3 bodies. 803 * <p/> 804 * <b>Important:</b> This overrides any previous set value using {@link #expectedMessageCount(int)} 805 */ 806 public void expectedBodiesReceivedInAnyOrder(final List<?> bodies) { 807 expectedMessageCount(bodies.size()); 808 this.expectedBodyValues = bodies; 809 this.actualBodyValues = new ArrayList<Object>(); 810 811 expects(new Runnable() { 812 public void run() { 813 List<Object> actualBodyValuesSet = new ArrayList<Object>(actualBodyValues); 814 for (int i = 0; i < expectedBodyValues.size(); i++) { 815 Exchange exchange = getReceivedExchange(i); 816 assertTrue("No exchange received for counter: " + i, exchange != null); 817 818 Object expectedBody = expectedBodyValues.get(i); 819 assertTrue("Message with body " + expectedBody + " was expected but not found in " + actualBodyValuesSet, actualBodyValuesSet.remove(expectedBody)); 820 } 821 } 822 }); 823 } 824 825 /** 826 * Adds an expectation that the given body values are received by this 827 * endpoint in any order 828 * <p/> 829 * <b>Important:</b> The number of bodies must match the expected number of messages, so if you expect 3 messages, then 830 * there must be 3 bodies. 831 * <p/> 832 * <b>Important:</b> This overrides any previous set value using {@link #expectedMessageCount(int)} 833 */ 834 public void expectedBodiesReceivedInAnyOrder(Object... bodies) { 835 List<Object> bodyList = new ArrayList<Object>(); 836 bodyList.addAll(Arrays.asList(bodies)); 837 expectedBodiesReceivedInAnyOrder(bodyList); 838 } 839 840 /** 841 * Adds an expectation that a file exists with the given name 842 * 843 * @param name name of file, will cater for / and \ on different OS platforms 844 */ 845 public void expectedFileExists(final String name) { 846 expectedFileExists(name, null); 847 } 848 849 /** 850 * Adds an expectation that a file exists with the given name 851 * <p/> 852 * Will wait at most 5 seconds while checking for the existence of the file. 853 * 854 * @param name name of file, will cater for / and \ on different OS platforms 855 * @param content content of file to compare, can be <tt>null</tt> to not compare content 856 */ 857 public void expectedFileExists(final String name, final String content) { 858 final File file = new File(FileUtil.normalizePath(name)); 859 860 expects(new Runnable() { 861 public void run() { 862 // wait at most 5 seconds for the file to exists 863 final long timeout = System.currentTimeMillis() + 5000; 864 865 boolean stop = false; 866 while (!stop && !file.exists()) { 867 try { 868 Thread.sleep(50); 869 } catch (InterruptedException e) { 870 // ignore 871 } 872 stop = System.currentTimeMillis() > timeout; 873 } 874 875 assertTrue("The file should exists: " + name, file.exists()); 876 877 if (content != null) { 878 String body = getCamelContext().getTypeConverter().convertTo(String.class, file); 879 assertEquals("Content of file: " + name, content, body); 880 } 881 } 882 }); 883 } 884 885 /** 886 * Adds an expectation that messages received should have the given exchange pattern 887 */ 888 public void expectedExchangePattern(final ExchangePattern exchangePattern) { 889 expectedMessagesMatches(new Predicate() { 890 public boolean matches(Exchange exchange) { 891 return exchange.getPattern().equals(exchangePattern); 892 } 893 }); 894 } 895 896 /** 897 * Adds an expectation that messages received should have ascending values 898 * of the given expression such as a user generated counter value 899 */ 900 public void expectsAscending(final Expression expression) { 901 expects(new Runnable() { 902 public void run() { 903 assertMessagesAscending(expression); 904 } 905 }); 906 } 907 908 /** 909 * Adds an expectation that messages received should have ascending values 910 * of the given expression such as a user generated counter value 911 */ 912 public AssertionClause expectsAscending() { 913 final AssertionClause clause = new AssertionClause(this) { 914 public void run() { 915 assertMessagesAscending(createExpression(getCamelContext())); 916 } 917 }; 918 expects(clause); 919 return clause; 920 } 921 922 /** 923 * Adds an expectation that messages received should have descending values 924 * of the given expression such as a user generated counter value 925 */ 926 public void expectsDescending(final Expression expression) { 927 expects(new Runnable() { 928 public void run() { 929 assertMessagesDescending(expression); 930 } 931 }); 932 } 933 934 /** 935 * Adds an expectation that messages received should have descending values 936 * of the given expression such as a user generated counter value 937 */ 938 public AssertionClause expectsDescending() { 939 final AssertionClause clause = new AssertionClause(this) { 940 public void run() { 941 assertMessagesDescending(createExpression(getCamelContext())); 942 } 943 }; 944 expects(clause); 945 return clause; 946 } 947 948 /** 949 * Adds an expectation that no duplicate messages should be received using 950 * the expression to determine the message ID 951 * 952 * @param expression the expression used to create a unique message ID for 953 * message comparison (which could just be the message 954 * payload if the payload can be tested for uniqueness using 955 * {@link Object#equals(Object)} and 956 * {@link Object#hashCode()} 957 */ 958 public void expectsNoDuplicates(final Expression expression) { 959 expects(new Runnable() { 960 public void run() { 961 assertNoDuplicates(expression); 962 } 963 }); 964 } 965 966 /** 967 * Adds an expectation that no duplicate messages should be received using 968 * the expression to determine the message ID 969 */ 970 public AssertionClause expectsNoDuplicates() { 971 final AssertionClause clause = new AssertionClause(this) { 972 public void run() { 973 assertNoDuplicates(createExpression(getCamelContext())); 974 } 975 }; 976 expects(clause); 977 return clause; 978 } 979 980 /** 981 * Asserts that the messages have ascending values of the given expression 982 */ 983 public void assertMessagesAscending(Expression expression) { 984 assertMessagesSorted(expression, true); 985 } 986 987 /** 988 * Asserts that the messages have descending values of the given expression 989 */ 990 public void assertMessagesDescending(Expression expression) { 991 assertMessagesSorted(expression, false); 992 } 993 994 protected void assertMessagesSorted(Expression expression, boolean ascending) { 995 String type = ascending ? "ascending" : "descending"; 996 ExpressionComparator comparator = new ExpressionComparator(expression); 997 List<Exchange> list = getReceivedExchanges(); 998 for (int i = 1; i < list.size(); i++) { 999 int j = i - 1; 1000 Exchange e1 = list.get(j); 1001 Exchange e2 = list.get(i); 1002 int result = comparator.compare(e1, e2); 1003 if (result == 0) { 1004 fail("Messages not " + type + ". Messages" + j + " and " + i + " are equal with value: " 1005 + expression.evaluate(e1, Object.class) + " for expression: " + expression + ". Exchanges: " + e1 + " and " + e2); 1006 } else { 1007 if (!ascending) { 1008 result = result * -1; 1009 } 1010 if (result > 0) { 1011 fail("Messages not " + type + ". Message " + j + " has value: " + expression.evaluate(e1, Object.class) 1012 + " and message " + i + " has value: " + expression.evaluate(e2, Object.class) + " for expression: " 1013 + expression + ". Exchanges: " + e1 + " and " + e2); 1014 } 1015 } 1016 } 1017 } 1018 1019 public void assertNoDuplicates(Expression expression) { 1020 Map<Object, Exchange> map = new HashMap<Object, Exchange>(); 1021 List<Exchange> list = getReceivedExchanges(); 1022 for (int i = 0; i < list.size(); i++) { 1023 Exchange e2 = list.get(i); 1024 Object key = expression.evaluate(e2, Object.class); 1025 Exchange e1 = map.get(key); 1026 if (e1 != null) { 1027 fail("Duplicate message found on message " + i + " has value: " + key + " for expression: " + expression + ". Exchanges: " + e1 + " and " + e2); 1028 } else { 1029 map.put(key, e2); 1030 } 1031 } 1032 } 1033 1034 /** 1035 * Adds the expectation which will be invoked when enough messages are received 1036 */ 1037 public void expects(Runnable runnable) { 1038 tests.add(runnable); 1039 } 1040 1041 /** 1042 * Adds an assertion to the given message index 1043 * 1044 * @param messageIndex the number of the message 1045 * @return the assertion clause 1046 */ 1047 public AssertionClause message(final int messageIndex) { 1048 final AssertionClause clause = new AssertionClause(this) { 1049 public void run() { 1050 applyAssertionOn(MockEndpoint.this, messageIndex, assertExchangeReceived(messageIndex)); 1051 } 1052 }; 1053 expects(clause); 1054 return clause; 1055 } 1056 1057 /** 1058 * Adds an assertion to all the received messages 1059 * 1060 * @return the assertion clause 1061 */ 1062 public AssertionClause allMessages() { 1063 final AssertionClause clause = new AssertionClause(this) { 1064 public void run() { 1065 List<Exchange> list = getReceivedExchanges(); 1066 int index = 0; 1067 for (Exchange exchange : list) { 1068 applyAssertionOn(MockEndpoint.this, index++, exchange); 1069 } 1070 } 1071 }; 1072 expects(clause); 1073 return clause; 1074 } 1075 1076 /** 1077 * Asserts that the given index of message is received (starting at zero) 1078 */ 1079 public Exchange assertExchangeReceived(int index) { 1080 int count = getReceivedCounter(); 1081 assertTrue("Not enough messages received. Was: " + count, count > index); 1082 return getReceivedExchange(index); 1083 } 1084 1085 // Properties 1086 // ------------------------------------------------------------------------- 1087 1088 public String getName() { 1089 return name; 1090 } 1091 1092 public void setName(String name) { 1093 this.name = name; 1094 } 1095 1096 public List<Throwable> getFailures() { 1097 return failures; 1098 } 1099 1100 public int getReceivedCounter() { 1101 return counter; 1102 } 1103 1104 public List<Exchange> getReceivedExchanges() { 1105 return receivedExchanges; 1106 } 1107 1108 public int getExpectedCount() { 1109 return expectedCount; 1110 } 1111 1112 public long getSleepForEmptyTest() { 1113 return sleepForEmptyTest; 1114 } 1115 1116 /** 1117 * Allows a sleep to be specified to wait to check that this endpoint really 1118 * is empty when {@link #expectedMessageCount(int)} is called with zero 1119 * 1120 * @param sleepForEmptyTest the milliseconds to sleep for to determine that 1121 * this endpoint really is empty 1122 */ 1123 public void setSleepForEmptyTest(long sleepForEmptyTest) { 1124 this.sleepForEmptyTest = sleepForEmptyTest; 1125 } 1126 1127 public long getResultWaitTime() { 1128 return resultWaitTime; 1129 } 1130 1131 /** 1132 * Sets the maximum amount of time (in millis) the {@link #assertIsSatisfied()} will 1133 * wait on a latch until it is satisfied 1134 */ 1135 public void setResultWaitTime(long resultWaitTime) { 1136 this.resultWaitTime = resultWaitTime; 1137 } 1138 1139 /** 1140 * Sets the minimum expected amount of time (in millis) the {@link #assertIsSatisfied()} will 1141 * wait on a latch until it is satisfied 1142 */ 1143 public void setResultMinimumWaitTime(long resultMinimumWaitTime) { 1144 this.resultMinimumWaitTime = resultMinimumWaitTime; 1145 } 1146 1147 /** 1148 * @deprecated use {@link #setResultMinimumWaitTime(long)} 1149 */ 1150 @Deprecated 1151 public void setMinimumResultWaitTime(long resultMinimumWaitTime) { 1152 setResultMinimumWaitTime(resultMinimumWaitTime); 1153 } 1154 1155 /** 1156 * Specifies the expected number of message exchanges that should be 1157 * received by this endpoint. 1158 * <p/> 1159 * <b>Beware:</b> If you want to expect that <tt>0</tt> messages, then take extra care, 1160 * as <tt>0</tt> matches when the tests starts, so you need to set a assert period time 1161 * to let the test run for a while to make sure there are still no messages arrived; for 1162 * that use {@link #setAssertPeriod(long)}. 1163 * An alternative is to use <a href="http://camel.apache.org/notifybuilder.html">NotifyBuilder</a>, and use the notifier 1164 * to know when Camel is done routing some messages, before you call the {@link #assertIsSatisfied()} method on the mocks. 1165 * This allows you to not use a fixed assert period, to speedup testing times. 1166 * <p/> 1167 * If you want to assert that <b>exactly</b> n'th message arrives to this mock 1168 * endpoint, then see also the {@link #setAssertPeriod(long)} method for further details. 1169 * 1170 * @param expectedCount the number of message exchanges that should be 1171 * expected by this endpoint 1172 * @see #setAssertPeriod(long) 1173 */ 1174 public void setExpectedCount(int expectedCount) { 1175 setExpectedMessageCount(expectedCount); 1176 } 1177 1178 /** 1179 * @see #setExpectedCount(int) 1180 */ 1181 public void setExpectedMessageCount(int expectedCount) { 1182 this.expectedCount = expectedCount; 1183 if (expectedCount <= 0) { 1184 latch = null; 1185 } else { 1186 latch = new CountDownLatch(expectedCount); 1187 } 1188 } 1189 1190 /** 1191 * Specifies the minimum number of expected message exchanges that should be 1192 * received by this endpoint 1193 * 1194 * @param expectedCount the number of message exchanges that should be 1195 * expected by this endpoint 1196 */ 1197 public void setMinimumExpectedMessageCount(int expectedCount) { 1198 this.expectedMinimumCount = expectedCount; 1199 if (expectedCount <= 0) { 1200 latch = null; 1201 } else { 1202 latch = new CountDownLatch(expectedMinimumCount); 1203 } 1204 } 1205 1206 public Processor getReporter() { 1207 return reporter; 1208 } 1209 1210 /** 1211 * Allows a processor to added to the endpoint to report on progress of the test 1212 */ 1213 public void setReporter(Processor reporter) { 1214 this.reporter = reporter; 1215 } 1216 1217 /** 1218 * Specifies to only retain the first n'th number of received {@link Exchange}s. 1219 * <p/> 1220 * This is used when testing with big data, to reduce memory consumption by not storing 1221 * copies of every {@link Exchange} this mock endpoint receives. 1222 * <p/> 1223 * <b>Important:</b> When using this limitation, then the {@link #getReceivedCounter()} 1224 * will still return the actual number of received {@link Exchange}s. For example 1225 * if we have received 5000 {@link Exchange}s, and have configured to only retain the first 1226 * 10 {@link Exchange}s, then the {@link #getReceivedCounter()} will still return <tt>5000</tt> 1227 * but there is only the first 10 {@link Exchange}s in the {@link #getExchanges()} and 1228 * {@link #getReceivedExchanges()} methods. 1229 * <p/> 1230 * When using this method, then some of the other expectation methods is not supported, 1231 * for example the {@link #expectedBodiesReceived(Object...)} sets a expectation on the first 1232 * number of bodies received. 1233 * <p/> 1234 * You can configure both {@link #setRetainFirst(int)} and {@link #setRetainLast(int)} methods, 1235 * to limit both the first and last received. 1236 * 1237 * @param retainFirst to limit and only keep the first n'th received {@link Exchange}s, use 1238 * <tt>0</tt> to not retain any messages, or <tt>-1</tt> to retain all. 1239 * @see #setRetainLast(int) 1240 */ 1241 public void setRetainFirst(int retainFirst) { 1242 this.retainFirst = retainFirst; 1243 } 1244 1245 /** 1246 * Specifies to only retain the last n'th number of received {@link Exchange}s. 1247 * <p/> 1248 * This is used when testing with big data, to reduce memory consumption by not storing 1249 * copies of every {@link Exchange} this mock endpoint receives. 1250 * <p/> 1251 * <b>Important:</b> When using this limitation, then the {@link #getReceivedCounter()} 1252 * will still return the actual number of received {@link Exchange}s. For example 1253 * if we have received 5000 {@link Exchange}s, and have configured to only retain the last 1254 * 20 {@link Exchange}s, then the {@link #getReceivedCounter()} will still return <tt>5000</tt> 1255 * but there is only the last 20 {@link Exchange}s in the {@link #getExchanges()} and 1256 * {@link #getReceivedExchanges()} methods. 1257 * <p/> 1258 * When using this method, then some of the other expectation methods is not supported, 1259 * for example the {@link #expectedBodiesReceived(Object...)} sets a expectation on the first 1260 * number of bodies received. 1261 * <p/> 1262 * You can configure both {@link #setRetainFirst(int)} and {@link #setRetainLast(int)} methods, 1263 * to limit both the first and last received. 1264 * 1265 * @param retainLast to limit and only keep the last n'th received {@link Exchange}s, use 1266 * <tt>0</tt> to not retain any messages, or <tt>-1</tt> to retain all. 1267 * @see #setRetainFirst(int) 1268 */ 1269 public void setRetainLast(int retainLast) { 1270 this.retainLast = retainLast; 1271 } 1272 1273 public int isReportGroup() { 1274 return reportGroup; 1275 } 1276 1277 /** 1278 * A number that is used to turn on throughput logging based on groups of the size. 1279 */ 1280 public void setReportGroup(int reportGroup) { 1281 this.reportGroup = reportGroup; 1282 } 1283 1284 public boolean isCopyOnExchange() { 1285 return copyOnExchange; 1286 } 1287 1288 /** 1289 * Sets whether to make a deep copy of the incoming {@link Exchange} when received at this mock endpoint. 1290 * <p/> 1291 * Is by default <tt>true</tt>. 1292 */ 1293 public void setCopyOnExchange(boolean copyOnExchange) { 1294 this.copyOnExchange = copyOnExchange; 1295 } 1296 1297 // Implementation methods 1298 // ------------------------------------------------------------------------- 1299 private void init() { 1300 expectedCount = -1; 1301 counter = 0; 1302 defaultProcessor = null; 1303 processors = new HashMap<Integer, Processor>(); 1304 receivedExchanges = new CopyOnWriteArrayList<Exchange>(); 1305 failures = new CopyOnWriteArrayList<Throwable>(); 1306 tests = new CopyOnWriteArrayList<Runnable>(); 1307 latch = null; 1308 sleepForEmptyTest = 0; 1309 resultWaitTime = 0; 1310 resultMinimumWaitTime = 0L; 1311 assertPeriod = 0L; 1312 expectedMinimumCount = -1; 1313 expectedBodyValues = null; 1314 actualBodyValues = new ArrayList<Object>(); 1315 expectedHeaderValues = null; 1316 actualHeaderValues = null; 1317 expectedPropertyValues = null; 1318 actualPropertyValues = null; 1319 retainFirst = -1; 1320 retainLast = -1; 1321 } 1322 1323 protected synchronized void onExchange(Exchange exchange) { 1324 try { 1325 if (reporter != null) { 1326 reporter.process(exchange); 1327 } 1328 Exchange copy = exchange; 1329 if (copyOnExchange) { 1330 // copy the exchange so the mock stores the copy and not the actual exchange 1331 copy = ExchangeHelper.createCopy(exchange, true); 1332 } 1333 performAssertions(exchange, copy); 1334 } catch (Throwable e) { 1335 // must catch java.lang.Throwable as AssertionError extends java.lang.Error 1336 failures.add(e); 1337 } finally { 1338 // make sure latch is counted down to avoid test hanging forever 1339 if (latch != null) { 1340 latch.countDown(); 1341 } 1342 } 1343 } 1344 1345 /** 1346 * Performs the assertions on the incoming exchange. 1347 * 1348 * @param exchange the actual exchange 1349 * @param copy a copy of the exchange (only store this) 1350 * @throws Exception can be thrown if something went wrong 1351 */ 1352 protected void performAssertions(Exchange exchange, Exchange copy) throws Exception { 1353 Message in = copy.getIn(); 1354 Object actualBody = in.getBody(); 1355 1356 if (expectedHeaderValues != null) { 1357 if (actualHeaderValues == null) { 1358 actualHeaderValues = getCamelContext().getHeadersMapFactory().newMap(); 1359 } 1360 if (in.hasHeaders()) { 1361 actualHeaderValues.putAll(in.getHeaders()); 1362 } 1363 } 1364 1365 if (expectedPropertyValues != null) { 1366 if (actualPropertyValues == null) { 1367 actualPropertyValues = getCamelContext().getHeadersMapFactory().newMap(); 1368 } 1369 actualPropertyValues.putAll(copy.getProperties()); 1370 } 1371 1372 if (expectedBodyValues != null) { 1373 int index = actualBodyValues.size(); 1374 if (expectedBodyValues.size() > index) { 1375 Object expectedBody = expectedBodyValues.get(index); 1376 if (expectedBody != null) { 1377 // prefer to convert body early, for example when using files 1378 // we need to read the content at this time 1379 Object body = in.getBody(expectedBody.getClass()); 1380 if (body != null) { 1381 actualBody = body; 1382 } 1383 } 1384 actualBodyValues.add(actualBody); 1385 } 1386 } 1387 1388 // let counter be 0 index-based in the logs 1389 if (LOG.isDebugEnabled()) { 1390 String msg = getEndpointUri() + " >>>> " + counter + " : " + copy + " with body: " + actualBody; 1391 if (copy.getIn().hasHeaders()) { 1392 msg += " and headers:" + copy.getIn().getHeaders(); 1393 } 1394 LOG.debug(msg); 1395 } 1396 1397 // record timestamp when exchange was received 1398 copy.setProperty(Exchange.RECEIVED_TIMESTAMP, new Date()); 1399 1400 // add a copy of the received exchange 1401 addReceivedExchange(copy); 1402 // and then increment counter after adding received exchange 1403 ++counter; 1404 1405 Processor processor = processors.get(getReceivedCounter()) != null 1406 ? processors.get(getReceivedCounter()) : defaultProcessor; 1407 1408 if (processor != null) { 1409 try { 1410 // must process the incoming exchange and NOT the copy as the idea 1411 // is the end user can manipulate the exchange 1412 processor.process(exchange); 1413 } catch (Exception e) { 1414 // set exceptions on exchange so we can throw exceptions to simulate errors 1415 exchange.setException(e); 1416 } 1417 } 1418 } 1419 1420 /** 1421 * Adds the received exchange. 1422 * 1423 * @param copy a copy of the received exchange 1424 */ 1425 protected void addReceivedExchange(Exchange copy) { 1426 if (retainFirst == 0 && retainLast == 0) { 1427 // do not retain any messages at all 1428 } else if (retainFirst < 0 && retainLast < 0) { 1429 // no limitation so keep them all 1430 receivedExchanges.add(copy); 1431 } else { 1432 // okay there is some sort of limitations, so figure out what to retain 1433 if (retainFirst > 0 && counter < retainFirst) { 1434 // store a copy as its within the retain first limitation 1435 receivedExchanges.add(copy); 1436 } else if (retainLast > 0) { 1437 // remove the oldest from the last retained boundary, 1438 int index = receivedExchanges.size() - retainLast; 1439 if (index >= 0) { 1440 // but must be outside the first range as well 1441 // otherwise we should not remove the oldest 1442 if (retainFirst <= 0 || retainFirst <= index) { 1443 receivedExchanges.remove(index); 1444 } 1445 } 1446 // store a copy of the last n'th received 1447 receivedExchanges.add(copy); 1448 } 1449 } 1450 } 1451 1452 protected void waitForCompleteLatch() throws InterruptedException { 1453 if (latch == null) { 1454 fail("Should have a latch!"); 1455 } 1456 1457 StopWatch watch = new StopWatch(); 1458 waitForCompleteLatch(resultWaitTime); 1459 long delta = watch.taken(); 1460 LOG.debug("Took {} millis to complete latch", delta); 1461 1462 if (resultMinimumWaitTime > 0 && delta < resultMinimumWaitTime) { 1463 fail("Expected minimum " + resultMinimumWaitTime 1464 + " millis waiting on the result, but was faster with " + delta + " millis."); 1465 } 1466 } 1467 1468 protected void waitForCompleteLatch(long timeout) throws InterruptedException { 1469 // Wait for a default 10 seconds if resultWaitTime is not set 1470 long waitTime = timeout == 0 ? 10000L : timeout; 1471 1472 // now let's wait for the results 1473 LOG.debug("Waiting on the latch for: {} millis", timeout); 1474 latch.await(waitTime, TimeUnit.MILLISECONDS); 1475 } 1476 1477 protected void assertEquals(String message, Object expectedValue, Object actualValue) { 1478 if (!ObjectHelper.equal(expectedValue, actualValue)) { 1479 fail(message + ". Expected: <" + expectedValue + "> but was: <" + actualValue + ">"); 1480 } 1481 } 1482 1483 protected void assertTrue(String message, boolean predicate) { 1484 if (!predicate) { 1485 fail(message); 1486 } 1487 } 1488 1489 protected void fail(Object message) { 1490 if (LOG.isDebugEnabled()) { 1491 List<Exchange> list = getReceivedExchanges(); 1492 int index = 0; 1493 for (Exchange exchange : list) { 1494 LOG.debug("{} failed and received[{}]: {}", new Object[]{getEndpointUri(), ++index, exchange}); 1495 } 1496 } 1497 throw new AssertionError(getEndpointUri() + " " + message); 1498 } 1499 1500 public int getExpectedMinimumCount() { 1501 return expectedMinimumCount; 1502 } 1503 1504 public void await() throws InterruptedException { 1505 if (latch != null) { 1506 latch.await(); 1507 } 1508 } 1509 1510 public boolean await(long timeout, TimeUnit unit) throws InterruptedException { 1511 if (latch != null) { 1512 return latch.await(timeout, unit); 1513 } 1514 return true; 1515 } 1516 1517 public boolean isSingleton() { 1518 return true; 1519 } 1520 1521 public boolean isLenientProperties() { 1522 return true; 1523 } 1524 1525 private Exchange getReceivedExchange(int index) { 1526 if (index <= receivedExchanges.size() - 1) { 1527 return receivedExchanges.get(index); 1528 } else { 1529 return null; 1530 } 1531 } 1532 1533}