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 if (LOG.isDebugEnabled()) { 438 // log incl stacktrace 439 LOG.debug("Caught expected failure: " + e.getMessage(), e); 440 } else { 441 LOG.info("Caught expected failure: " + e.getMessage()); 442 } 443 } 444 if (failed) { 445 // fail() throws the AssertionError to indicate the test failed. 446 fail("Expected assertion failure but test succeeded!"); 447 } 448 } 449 450 /** 451 * Validates that the assertions fail on this endpoint 452 453 * @param timeoutForEmptyEndpoints the timeout in milliseconds that we 454 * should wait for the test to be true 455 */ 456 public void assertIsNotSatisfied(long timeoutForEmptyEndpoints) throws InterruptedException { 457 boolean failed = false; 458 try { 459 assertIsSatisfied(timeoutForEmptyEndpoints); 460 // did not throw expected error... fail! 461 failed = true; 462 } catch (AssertionError e) { 463 if (LOG.isDebugEnabled()) { 464 // log incl stacktrace 465 LOG.debug("Caught expected failure: " + e.getMessage(), e); 466 } else { 467 LOG.info("Caught expected failure: " + e.getMessage()); 468 } 469 } 470 if (failed) { 471 // fail() throws the AssertionError to indicate the test failed. 472 fail("Expected assertion failure but test succeeded!"); 473 } 474 } 475 476 /** 477 * Specifies the expected number of message exchanges that should be 478 * received by this endpoint 479 * 480 * If you want to assert that <b>exactly</b> n messages arrives to this mock 481 * endpoint, then see also the {@link #setAssertPeriod(long)} method for further details. 482 * 483 * @param expectedCount the number of message exchanges that should be 484 * expected by this endpoint 485 * @see #setAssertPeriod(long) 486 */ 487 public void expectedMessageCount(int expectedCount) { 488 setExpectedMessageCount(expectedCount); 489 } 490 491 /** 492 * Sets a grace period after which the mock endpoint will re-assert 493 * to ensure the preliminary assertion is still valid. 494 * <p/> 495 * This is used for example to assert that <b>exactly</b> a number of messages 496 * arrives. For example if {@link #expectedMessageCount(int)} was set to 5, then 497 * the assertion is satisfied when 5 or more message arrives. To ensure that 498 * exactly 5 messages arrives, then you would need to wait a little period 499 * to ensure no further message arrives. This is what you can use this 500 * {@link #setAssertPeriod(long)} method for. 501 * <p/> 502 * By default this period is disabled. 503 * 504 * @param period grace period in millis 505 */ 506 public void setAssertPeriod(long period) { 507 this.assertPeriod = period; 508 } 509 510 /** 511 * Specifies the minimum number of expected message exchanges that should be 512 * received by this endpoint 513 * 514 * @param expectedCount the number of message exchanges that should be 515 * expected by this endpoint 516 */ 517 public void expectedMinimumMessageCount(int expectedCount) { 518 setMinimumExpectedMessageCount(expectedCount); 519 } 520 521 /** 522 * Sets an expectation that the given header name & value are received by this endpoint 523 * <p/> 524 * You can set multiple expectations for different header names. 525 * 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> 526 * <p/> 527 * <b>Important:</b> This overrides any previous set value using {@link #expectedMessageCount(int)} 528 */ 529 public void expectedHeaderReceived(final String name, final Object value) { 530 if (expectedHeaderValues == null) { 531 expectedHeaderValues = getCamelContext().getHeadersMapFactory().newMap(); 532 // we just wants to expects to be called once 533 expects(new Runnable() { 534 public void run() { 535 for (int i = 0; i < getReceivedExchanges().size(); i++) { 536 Exchange exchange = getReceivedExchange(i); 537 for (Map.Entry<String, Object> entry : expectedHeaderValues.entrySet()) { 538 String key = entry.getKey(); 539 Object expectedValue = entry.getValue(); 540 541 // we accept that an expectedValue of null also means that the header may be absent 542 if (expectedValue != null) { 543 assertTrue("Exchange " + i + " has no headers", exchange.getIn().hasHeaders()); 544 boolean hasKey = exchange.getIn().getHeaders().containsKey(key); 545 assertTrue("No header with name " + key + " found for message: " + i, hasKey); 546 } 547 548 Object actualValue = exchange.getIn().getHeader(key); 549 actualValue = extractActualValue(exchange, actualValue, expectedValue); 550 551 assertEquals("Header with name " + key + " for message: " + i, expectedValue, actualValue); 552 } 553 } 554 } 555 }); 556 } 557 expectedHeaderValues.put(name, value); 558 } 559 560 /** 561 * Adds an expectation that the given header values are received by this 562 * endpoint in any order. 563 * <p/> 564 * <b>Important:</b> The number of values must match the expected number of messages, so if you expect 3 messages, then 565 * there must be 3 values. 566 * <p/> 567 * <b>Important:</b> This overrides any previous set value using {@link #expectedMessageCount(int)} 568 */ 569 public void expectedHeaderValuesReceivedInAnyOrder(final String name, final List<?> values) { 570 expectedMessageCount(values.size()); 571 572 expects(new Runnable() { 573 public void run() { 574 // these are the expected values to find 575 final Set<Object> actualHeaderValues = new CopyOnWriteArraySet<>(values); 576 577 for (int i = 0; i < getReceivedExchanges().size(); i++) { 578 Exchange exchange = getReceivedExchange(i); 579 580 Object actualValue = exchange.getIn().getHeader(name); 581 for (Object expectedValue : actualHeaderValues) { 582 actualValue = extractActualValue(exchange, actualValue, expectedValue); 583 // remove any found values 584 actualHeaderValues.remove(actualValue); 585 } 586 } 587 588 // should be empty, as we should find all the values 589 assertTrue("Expected " + values.size() + " headers with key[" + name + "], received " + (values.size() - actualHeaderValues.size()) 590 + " headers. Expected header values: " + actualHeaderValues, actualHeaderValues.isEmpty()); 591 } 592 }); 593 } 594 595 /** 596 * Adds an expectation that the given header values are received by this 597 * endpoint in any order 598 * <p/> 599 * <b>Important:</b> The number of values must match the expected number of messages, so if you expect 3 messages, then 600 * there must be 3 values. 601 * <p/> 602 * <b>Important:</b> This overrides any previous set value using {@link #expectedMessageCount(int)} 603 */ 604 public void expectedHeaderValuesReceivedInAnyOrder(String name, Object... values) { 605 List<Object> valueList = new ArrayList<>(); 606 valueList.addAll(Arrays.asList(values)); 607 expectedHeaderValuesReceivedInAnyOrder(name, valueList); 608 } 609 610 /** 611 * Sets an expectation that the given property name & value are received by this endpoint 612 * <p/> 613 * You can set multiple expectations for different property names. 614 * 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> 615 */ 616 public void expectedPropertyReceived(final String name, final Object value) { 617 if (expectedPropertyValues == null) { 618 expectedPropertyValues = new HashMap<>(); 619 } 620 expectedPropertyValues.put(name, value); 621 622 expects(new Runnable() { 623 public void run() { 624 for (int i = 0; i < getReceivedExchanges().size(); i++) { 625 Exchange exchange = getReceivedExchange(i); 626 for (Map.Entry<String, Object> entry : expectedPropertyValues.entrySet()) { 627 String key = entry.getKey(); 628 Object expectedValue = entry.getValue(); 629 630 // we accept that an expectedValue of null also means that the property may be absent 631 if (expectedValue != null) { 632 assertTrue("Exchange " + i + " has no properties", !exchange.getProperties().isEmpty()); 633 boolean hasKey = exchange.getProperties().containsKey(key); 634 assertTrue("No property with name " + key + " found for message: " + i, hasKey); 635 } 636 637 Object actualValue = exchange.getProperty(key); 638 actualValue = extractActualValue(exchange, actualValue, expectedValue); 639 640 assertEquals("Property with name " + key + " for message: " + i, expectedValue, actualValue); 641 } 642 } 643 } 644 }); 645 } 646 647 /** 648 * Adds an expectation that the given property values are received by this 649 * endpoint in any order. 650 * <p/> 651 * <b>Important:</b> The number of values must match the expected number of messages, so if you expect 3 messages, then 652 * there must be 3 values. 653 * <p/> 654 * <b>Important:</b> This overrides any previous set value using {@link #expectedMessageCount(int)} 655 */ 656 public void expectedPropertyValuesReceivedInAnyOrder(final String name, final List<?> values) { 657 expectedMessageCount(values.size()); 658 659 expects(new Runnable() { 660 public void run() { 661 // these are the expected values to find 662 final Set<Object> actualPropertyValues = new CopyOnWriteArraySet<>(values); 663 664 for (int i = 0; i < getReceivedExchanges().size(); i++) { 665 Exchange exchange = getReceivedExchange(i); 666 667 Object actualValue = exchange.getProperty(name); 668 for (Object expectedValue : actualPropertyValues) { 669 actualValue = extractActualValue(exchange, actualValue, expectedValue); 670 // remove any found values 671 actualPropertyValues.remove(actualValue); 672 } 673 } 674 675 // should be empty, as we should find all the values 676 assertTrue("Expected " + values.size() + " properties with key[" + name + "], received " + (values.size() - actualPropertyValues.size()) 677 + " properties. Expected property values: " + actualPropertyValues, actualPropertyValues.isEmpty()); 678 } 679 }); 680 } 681 682 /** 683 * Adds an expectation that the given property values are received by this 684 * endpoint in any order 685 * <p/> 686 * <b>Important:</b> The number of values must match the expected number of messages, so if you expect 3 messages, then 687 * there must be 3 values. 688 * <p/> 689 * <b>Important:</b> This overrides any previous set value using {@link #expectedMessageCount(int)} 690 */ 691 public void expectedPropertyValuesReceivedInAnyOrder(String name, Object... values) { 692 List<Object> valueList = new ArrayList<>(); 693 valueList.addAll(Arrays.asList(values)); 694 expectedPropertyValuesReceivedInAnyOrder(name, valueList); 695 } 696 697 /** 698 * Adds an expectation that the given body values are received by this 699 * endpoint in the specified order 700 * <p/> 701 * <b>Important:</b> The number of values must match the expected number of messages, so if you expect 3 messages, then 702 * there must be 3 values. 703 * <p/> 704 * <b>Important:</b> This overrides any previous set value using {@link #expectedMessageCount(int)} 705 */ 706 public void expectedBodiesReceived(final List<?> bodies) { 707 expectedMessageCount(bodies.size()); 708 this.expectedBodyValues = bodies; 709 this.actualBodyValues = new ArrayList<>(); 710 711 expects(new Runnable() { 712 public void run() { 713 for (int i = 0; i < expectedBodyValues.size(); i++) { 714 Exchange exchange = getReceivedExchange(i); 715 assertTrue("No exchange received for counter: " + i, exchange != null); 716 717 Object expectedBody = expectedBodyValues.get(i); 718 Object actualBody = null; 719 if (i < actualBodyValues.size()) { 720 actualBody = actualBodyValues.get(i); 721 } 722 actualBody = extractActualValue(exchange, actualBody, expectedBody); 723 724 assertEquals("Body of message: " + i, expectedBody, actualBody); 725 } 726 } 727 }); 728 } 729 730 private Object extractActualValue(Exchange exchange, Object actualValue, Object expectedValue) { 731 if (actualValue == null) { 732 return null; 733 } 734 735 if (actualValue instanceof Expression) { 736 Class clazz = Object.class; 737 if (expectedValue != null) { 738 clazz = expectedValue.getClass(); 739 } 740 actualValue = ((Expression)actualValue).evaluate(exchange, clazz); 741 } else if (actualValue instanceof Predicate) { 742 actualValue = ((Predicate)actualValue).matches(exchange); 743 } else if (expectedValue != null) { 744 String from = actualValue.getClass().getName(); 745 String to = expectedValue.getClass().getName(); 746 actualValue = getCamelContext().getTypeConverter().convertTo(expectedValue.getClass(), exchange, actualValue); 747 assertTrue("There is no type conversion possible from " + from + " to " + to, actualValue != null); 748 } 749 return actualValue; 750 } 751 752 /** 753 * Sets an expectation that the given predicates matches the received messages by this endpoint 754 */ 755 public void expectedMessagesMatches(Predicate... predicates) { 756 for (int i = 0; i < predicates.length; i++) { 757 final int messageIndex = i; 758 final Predicate predicate = predicates[i]; 759 final AssertionClause clause = new AssertionClause(this) { 760 public void run() { 761 addPredicate(predicate); 762 applyAssertionOn(MockEndpoint.this, messageIndex, assertExchangeReceived(messageIndex)); 763 } 764 }; 765 expects(clause); 766 } 767 } 768 769 /** 770 * Sets an expectation that the given body values are received by this endpoint 771 * <p/> 772 * <b>Important:</b> The number of bodies must match the expected number of messages, so if you expect 3 messages, then 773 * there must be 3 bodies. 774 * <p/> 775 * <b>Important:</b> This overrides any previous set value using {@link #expectedMessageCount(int)} 776 */ 777 public void expectedBodiesReceived(Object... bodies) { 778 List<Object> bodyList = new ArrayList<>(); 779 bodyList.addAll(Arrays.asList(bodies)); 780 expectedBodiesReceived(bodyList); 781 } 782 783 /** 784 * Adds an expectation that the given body value are received by this endpoint 785 */ 786 public AssertionClause expectedBodyReceived() { 787 expectedMessageCount(1); 788 final AssertionClause clause = new AssertionClause(this) { 789 public void run() { 790 Exchange exchange = getReceivedExchange(0); 791 assertTrue("No exchange received for counter: " + 0, exchange != null); 792 793 Object actualBody = exchange.getIn().getBody(); 794 Expression exp = createExpression(getCamelContext()); 795 Object expectedBody = exp.evaluate(exchange, Object.class); 796 797 assertEquals("Body of message: " + 0, expectedBody, actualBody); 798 } 799 }; 800 expects(clause); 801 return clause; 802 } 803 804 /** 805 * Adds an expectation that the given body values are received by this 806 * endpoint in any order 807 * <p/> 808 * <b>Important:</b> The number of bodies must match the expected number of messages, so if you expect 3 messages, then 809 * there must be 3 bodies. 810 * <p/> 811 * <b>Important:</b> This overrides any previous set value using {@link #expectedMessageCount(int)} 812 */ 813 public void expectedBodiesReceivedInAnyOrder(final List<?> bodies) { 814 expectedMessageCount(bodies.size()); 815 this.expectedBodyValues = bodies; 816 this.actualBodyValues = new ArrayList<>(); 817 818 expects(new Runnable() { 819 public void run() { 820 List<Object> actualBodyValuesSet = new ArrayList<>(actualBodyValues); 821 for (int i = 0; i < expectedBodyValues.size(); i++) { 822 Exchange exchange = getReceivedExchange(i); 823 assertTrue("No exchange received for counter: " + i, exchange != null); 824 825 Object expectedBody = expectedBodyValues.get(i); 826 assertTrue("Message with body " + expectedBody + " was expected but not found in " + actualBodyValuesSet, actualBodyValuesSet.remove(expectedBody)); 827 } 828 } 829 }); 830 } 831 832 /** 833 * Adds an expectation that the given body values are received by this 834 * endpoint in any order 835 * <p/> 836 * <b>Important:</b> The number of bodies must match the expected number of messages, so if you expect 3 messages, then 837 * there must be 3 bodies. 838 * <p/> 839 * <b>Important:</b> This overrides any previous set value using {@link #expectedMessageCount(int)} 840 */ 841 public void expectedBodiesReceivedInAnyOrder(Object... bodies) { 842 List<Object> bodyList = new ArrayList<>(); 843 bodyList.addAll(Arrays.asList(bodies)); 844 expectedBodiesReceivedInAnyOrder(bodyList); 845 } 846 847 /** 848 * Adds an expectation that a file exists with the given name 849 * 850 * @param name name of file, will cater for / and \ on different OS platforms 851 */ 852 public void expectedFileExists(final String name) { 853 expectedFileExists(name, null); 854 } 855 856 /** 857 * Adds an expectation that a file exists with the given name 858 * <p/> 859 * Will wait at most 5 seconds while checking for the existence of the file. 860 * 861 * @param name name of file, will cater for / and \ on different OS platforms 862 * @param content content of file to compare, can be <tt>null</tt> to not compare content 863 */ 864 public void expectedFileExists(final String name, final String content) { 865 final File file = new File(FileUtil.normalizePath(name)); 866 867 expects(new Runnable() { 868 public void run() { 869 // wait at most 5 seconds for the file to exists 870 final long timeout = System.currentTimeMillis() + 5000; 871 872 boolean stop = false; 873 while (!stop && !file.exists()) { 874 try { 875 Thread.sleep(50); 876 } catch (InterruptedException e) { 877 // ignore 878 } 879 stop = System.currentTimeMillis() > timeout; 880 } 881 882 assertTrue("The file should exists: " + name, file.exists()); 883 884 if (content != null) { 885 String body = getCamelContext().getTypeConverter().convertTo(String.class, file); 886 assertEquals("Content of file: " + name, content, body); 887 } 888 } 889 }); 890 } 891 892 /** 893 * Adds an expectation that messages received should have the given exchange pattern 894 */ 895 public void expectedExchangePattern(final ExchangePattern exchangePattern) { 896 expectedMessagesMatches(new Predicate() { 897 public boolean matches(Exchange exchange) { 898 return exchange.getPattern().equals(exchangePattern); 899 } 900 }); 901 } 902 903 /** 904 * Adds an expectation that messages received should have ascending values 905 * of the given expression such as a user generated counter value 906 */ 907 public void expectsAscending(final Expression expression) { 908 expects(new Runnable() { 909 public void run() { 910 assertMessagesAscending(expression); 911 } 912 }); 913 } 914 915 /** 916 * Adds an expectation that messages received should have ascending values 917 * of the given expression such as a user generated counter value 918 */ 919 public AssertionClause expectsAscending() { 920 final AssertionClause clause = new AssertionClause(this) { 921 public void run() { 922 assertMessagesAscending(createExpression(getCamelContext())); 923 } 924 }; 925 expects(clause); 926 return clause; 927 } 928 929 /** 930 * Adds an expectation that messages received should have descending values 931 * of the given expression such as a user generated counter value 932 */ 933 public void expectsDescending(final Expression expression) { 934 expects(new Runnable() { 935 public void run() { 936 assertMessagesDescending(expression); 937 } 938 }); 939 } 940 941 /** 942 * Adds an expectation that messages received should have descending values 943 * of the given expression such as a user generated counter value 944 */ 945 public AssertionClause expectsDescending() { 946 final AssertionClause clause = new AssertionClause(this) { 947 public void run() { 948 assertMessagesDescending(createExpression(getCamelContext())); 949 } 950 }; 951 expects(clause); 952 return clause; 953 } 954 955 /** 956 * Adds an expectation that no duplicate messages should be received using 957 * the expression to determine the message ID 958 * 959 * @param expression the expression used to create a unique message ID for 960 * message comparison (which could just be the message 961 * payload if the payload can be tested for uniqueness using 962 * {@link Object#equals(Object)} and 963 * {@link Object#hashCode()} 964 */ 965 public void expectsNoDuplicates(final Expression expression) { 966 expects(new Runnable() { 967 public void run() { 968 assertNoDuplicates(expression); 969 } 970 }); 971 } 972 973 /** 974 * Adds an expectation that no duplicate messages should be received using 975 * the expression to determine the message ID 976 */ 977 public AssertionClause expectsNoDuplicates() { 978 final AssertionClause clause = new AssertionClause(this) { 979 public void run() { 980 assertNoDuplicates(createExpression(getCamelContext())); 981 } 982 }; 983 expects(clause); 984 return clause; 985 } 986 987 /** 988 * Asserts that the messages have ascending values of the given expression 989 */ 990 public void assertMessagesAscending(Expression expression) { 991 assertMessagesSorted(expression, true); 992 } 993 994 /** 995 * Asserts that the messages have descending values of the given expression 996 */ 997 public void assertMessagesDescending(Expression expression) { 998 assertMessagesSorted(expression, false); 999 } 1000 1001 protected void assertMessagesSorted(Expression expression, boolean ascending) { 1002 String type = ascending ? "ascending" : "descending"; 1003 ExpressionComparator comparator = new ExpressionComparator(expression); 1004 List<Exchange> list = getReceivedExchanges(); 1005 for (int i = 1; i < list.size(); i++) { 1006 int j = i - 1; 1007 Exchange e1 = list.get(j); 1008 Exchange e2 = list.get(i); 1009 int result = comparator.compare(e1, e2); 1010 if (result == 0) { 1011 fail("Messages not " + type + ". Messages" + j + " and " + i + " are equal with value: " 1012 + expression.evaluate(e1, Object.class) + " for expression: " + expression + ". Exchanges: " + e1 + " and " + e2); 1013 } else { 1014 if (!ascending) { 1015 result = result * -1; 1016 } 1017 if (result > 0) { 1018 fail("Messages not " + type + ". Message " + j + " has value: " + expression.evaluate(e1, Object.class) 1019 + " and message " + i + " has value: " + expression.evaluate(e2, Object.class) + " for expression: " 1020 + expression + ". Exchanges: " + e1 + " and " + e2); 1021 } 1022 } 1023 } 1024 } 1025 1026 public void assertNoDuplicates(Expression expression) { 1027 Map<Object, Exchange> map = new HashMap<>(); 1028 List<Exchange> list = getReceivedExchanges(); 1029 for (int i = 0; i < list.size(); i++) { 1030 Exchange e2 = list.get(i); 1031 Object key = expression.evaluate(e2, Object.class); 1032 Exchange e1 = map.get(key); 1033 if (e1 != null) { 1034 fail("Duplicate message found on message " + i + " has value: " + key + " for expression: " + expression + ". Exchanges: " + e1 + " and " + e2); 1035 } else { 1036 map.put(key, e2); 1037 } 1038 } 1039 } 1040 1041 /** 1042 * Adds the expectation which will be invoked when enough messages are received 1043 */ 1044 public void expects(Runnable runnable) { 1045 tests.add(runnable); 1046 } 1047 1048 /** 1049 * Adds an assertion to the given message index 1050 * 1051 * @param messageIndex the number of the message 1052 * @return the assertion clause 1053 */ 1054 public AssertionClause message(final int messageIndex) { 1055 final AssertionClause clause = new AssertionClause(this) { 1056 public void run() { 1057 applyAssertionOn(MockEndpoint.this, messageIndex, assertExchangeReceived(messageIndex)); 1058 } 1059 }; 1060 expects(clause); 1061 return clause; 1062 } 1063 1064 /** 1065 * Adds an assertion to all the received messages 1066 * 1067 * @return the assertion clause 1068 */ 1069 public AssertionClause allMessages() { 1070 final AssertionClause clause = new AssertionClause(this) { 1071 public void run() { 1072 List<Exchange> list = getReceivedExchanges(); 1073 int index = 0; 1074 for (Exchange exchange : list) { 1075 applyAssertionOn(MockEndpoint.this, index++, exchange); 1076 } 1077 } 1078 }; 1079 expects(clause); 1080 return clause; 1081 } 1082 1083 /** 1084 * Asserts that the given index of message is received (starting at zero) 1085 */ 1086 public Exchange assertExchangeReceived(int index) { 1087 int count = getReceivedCounter(); 1088 assertTrue("Not enough messages received. Was: " + count, count > index); 1089 return getReceivedExchange(index); 1090 } 1091 1092 // Properties 1093 // ------------------------------------------------------------------------- 1094 1095 public String getName() { 1096 return name; 1097 } 1098 1099 public void setName(String name) { 1100 this.name = name; 1101 } 1102 1103 public List<Throwable> getFailures() { 1104 return failures; 1105 } 1106 1107 public int getReceivedCounter() { 1108 return counter; 1109 } 1110 1111 public List<Exchange> getReceivedExchanges() { 1112 return receivedExchanges; 1113 } 1114 1115 public int getExpectedCount() { 1116 return expectedCount; 1117 } 1118 1119 public long getSleepForEmptyTest() { 1120 return sleepForEmptyTest; 1121 } 1122 1123 /** 1124 * Allows a sleep to be specified to wait to check that this endpoint really 1125 * is empty when {@link #expectedMessageCount(int)} is called with zero 1126 * 1127 * @param sleepForEmptyTest the milliseconds to sleep for to determine that 1128 * this endpoint really is empty 1129 */ 1130 public void setSleepForEmptyTest(long sleepForEmptyTest) { 1131 this.sleepForEmptyTest = sleepForEmptyTest; 1132 } 1133 1134 public long getResultWaitTime() { 1135 return resultWaitTime; 1136 } 1137 1138 /** 1139 * Sets the maximum amount of time (in millis) the {@link #assertIsSatisfied()} will 1140 * wait on a latch until it is satisfied 1141 */ 1142 public void setResultWaitTime(long resultWaitTime) { 1143 this.resultWaitTime = resultWaitTime; 1144 } 1145 1146 /** 1147 * Sets the minimum expected amount of time (in millis) the {@link #assertIsSatisfied()} will 1148 * wait on a latch until it is satisfied 1149 */ 1150 public void setResultMinimumWaitTime(long resultMinimumWaitTime) { 1151 this.resultMinimumWaitTime = resultMinimumWaitTime; 1152 } 1153 1154 /** 1155 * @deprecated use {@link #setResultMinimumWaitTime(long)} 1156 */ 1157 @Deprecated 1158 public void setMinimumResultWaitTime(long resultMinimumWaitTime) { 1159 setResultMinimumWaitTime(resultMinimumWaitTime); 1160 } 1161 1162 /** 1163 * Specifies the expected number of message exchanges that should be 1164 * received by this endpoint. 1165 * <p/> 1166 * <b>Beware:</b> If you want to expect that <tt>0</tt> messages, then take extra care, 1167 * as <tt>0</tt> matches when the tests starts, so you need to set a assert period time 1168 * to let the test run for a while to make sure there are still no messages arrived; for 1169 * that use {@link #setAssertPeriod(long)}. 1170 * An alternative is to use <a href="http://camel.apache.org/notifybuilder.html">NotifyBuilder</a>, and use the notifier 1171 * to know when Camel is done routing some messages, before you call the {@link #assertIsSatisfied()} method on the mocks. 1172 * This allows you to not use a fixed assert period, to speedup testing times. 1173 * <p/> 1174 * If you want to assert that <b>exactly</b> n'th message arrives to this mock 1175 * endpoint, then see also the {@link #setAssertPeriod(long)} method for further details. 1176 * 1177 * @param expectedCount the number of message exchanges that should be 1178 * expected by this endpoint 1179 * @see #setAssertPeriod(long) 1180 */ 1181 public void setExpectedCount(int expectedCount) { 1182 setExpectedMessageCount(expectedCount); 1183 } 1184 1185 /** 1186 * @see #setExpectedCount(int) 1187 */ 1188 public void setExpectedMessageCount(int expectedCount) { 1189 this.expectedCount = expectedCount; 1190 if (expectedCount <= 0) { 1191 latch = null; 1192 } else { 1193 latch = new CountDownLatch(expectedCount); 1194 } 1195 } 1196 1197 /** 1198 * Specifies the minimum number of expected message exchanges that should be 1199 * received by this endpoint 1200 * 1201 * @param expectedCount the number of message exchanges that should be 1202 * expected by this endpoint 1203 */ 1204 public void setMinimumExpectedMessageCount(int expectedCount) { 1205 this.expectedMinimumCount = expectedCount; 1206 if (expectedCount <= 0) { 1207 latch = null; 1208 } else { 1209 latch = new CountDownLatch(expectedMinimumCount); 1210 } 1211 } 1212 1213 public Processor getReporter() { 1214 return reporter; 1215 } 1216 1217 /** 1218 * Allows a processor to added to the endpoint to report on progress of the test 1219 */ 1220 public void setReporter(Processor reporter) { 1221 this.reporter = reporter; 1222 } 1223 1224 /** 1225 * Specifies to only retain the first n'th number of received {@link Exchange}s. 1226 * <p/> 1227 * This is used when testing with big data, to reduce memory consumption by not storing 1228 * copies of every {@link Exchange} this mock endpoint receives. 1229 * <p/> 1230 * <b>Important:</b> When using this limitation, then the {@link #getReceivedCounter()} 1231 * will still return the actual number of received {@link Exchange}s. For example 1232 * if we have received 5000 {@link Exchange}s, and have configured to only retain the first 1233 * 10 {@link Exchange}s, then the {@link #getReceivedCounter()} will still return <tt>5000</tt> 1234 * but there is only the first 10 {@link Exchange}s in the {@link #getExchanges()} and 1235 * {@link #getReceivedExchanges()} methods. 1236 * <p/> 1237 * When using this method, then some of the other expectation methods is not supported, 1238 * for example the {@link #expectedBodiesReceived(Object...)} sets a expectation on the first 1239 * number of bodies received. 1240 * <p/> 1241 * You can configure both {@link #setRetainFirst(int)} and {@link #setRetainLast(int)} methods, 1242 * to limit both the first and last received. 1243 * 1244 * @param retainFirst to limit and only keep the first n'th received {@link Exchange}s, use 1245 * <tt>0</tt> to not retain any messages, or <tt>-1</tt> to retain all. 1246 * @see #setRetainLast(int) 1247 */ 1248 public void setRetainFirst(int retainFirst) { 1249 this.retainFirst = retainFirst; 1250 } 1251 1252 /** 1253 * Specifies to only retain the last n'th number of received {@link Exchange}s. 1254 * <p/> 1255 * This is used when testing with big data, to reduce memory consumption by not storing 1256 * copies of every {@link Exchange} this mock endpoint receives. 1257 * <p/> 1258 * <b>Important:</b> When using this limitation, then the {@link #getReceivedCounter()} 1259 * will still return the actual number of received {@link Exchange}s. For example 1260 * if we have received 5000 {@link Exchange}s, and have configured to only retain the last 1261 * 20 {@link Exchange}s, then the {@link #getReceivedCounter()} will still return <tt>5000</tt> 1262 * but there is only the last 20 {@link Exchange}s in the {@link #getExchanges()} and 1263 * {@link #getReceivedExchanges()} methods. 1264 * <p/> 1265 * When using this method, then some of the other expectation methods is not supported, 1266 * for example the {@link #expectedBodiesReceived(Object...)} sets a expectation on the first 1267 * number of bodies received. 1268 * <p/> 1269 * You can configure both {@link #setRetainFirst(int)} and {@link #setRetainLast(int)} methods, 1270 * to limit both the first and last received. 1271 * 1272 * @param retainLast to limit and only keep the last n'th received {@link Exchange}s, use 1273 * <tt>0</tt> to not retain any messages, or <tt>-1</tt> to retain all. 1274 * @see #setRetainFirst(int) 1275 */ 1276 public void setRetainLast(int retainLast) { 1277 this.retainLast = retainLast; 1278 } 1279 1280 public int isReportGroup() { 1281 return reportGroup; 1282 } 1283 1284 /** 1285 * A number that is used to turn on throughput logging based on groups of the size. 1286 */ 1287 public void setReportGroup(int reportGroup) { 1288 this.reportGroup = reportGroup; 1289 } 1290 1291 public boolean isCopyOnExchange() { 1292 return copyOnExchange; 1293 } 1294 1295 /** 1296 * Sets whether to make a deep copy of the incoming {@link Exchange} when received at this mock endpoint. 1297 * <p/> 1298 * Is by default <tt>true</tt>. 1299 */ 1300 public void setCopyOnExchange(boolean copyOnExchange) { 1301 this.copyOnExchange = copyOnExchange; 1302 } 1303 1304 // Implementation methods 1305 // ------------------------------------------------------------------------- 1306 private void init() { 1307 expectedCount = -1; 1308 counter = 0; 1309 defaultProcessor = null; 1310 processors = new HashMap<>(); 1311 receivedExchanges = new CopyOnWriteArrayList<>(); 1312 failures = new CopyOnWriteArrayList<>(); 1313 tests = new CopyOnWriteArrayList<>(); 1314 latch = null; 1315 sleepForEmptyTest = 0; 1316 resultWaitTime = 0; 1317 resultMinimumWaitTime = 0L; 1318 assertPeriod = 0L; 1319 expectedMinimumCount = -1; 1320 expectedBodyValues = null; 1321 actualBodyValues = new ArrayList<>(); 1322 expectedHeaderValues = null; 1323 actualHeaderValues = null; 1324 expectedPropertyValues = null; 1325 actualPropertyValues = null; 1326 retainFirst = -1; 1327 retainLast = -1; 1328 } 1329 1330 protected synchronized void onExchange(Exchange exchange) { 1331 try { 1332 if (reporter != null) { 1333 reporter.process(exchange); 1334 } 1335 Exchange copy = exchange; 1336 if (copyOnExchange) { 1337 // copy the exchange so the mock stores the copy and not the actual exchange 1338 copy = ExchangeHelper.createCopy(exchange, true); 1339 } 1340 performAssertions(exchange, copy); 1341 } catch (Throwable e) { 1342 // must catch java.lang.Throwable as AssertionError extends java.lang.Error 1343 failures.add(e); 1344 } finally { 1345 // make sure latch is counted down to avoid test hanging forever 1346 if (latch != null) { 1347 latch.countDown(); 1348 } 1349 } 1350 } 1351 1352 /** 1353 * Performs the assertions on the incoming exchange. 1354 * 1355 * @param exchange the actual exchange 1356 * @param copy a copy of the exchange (only store this) 1357 * @throws Exception can be thrown if something went wrong 1358 */ 1359 protected void performAssertions(Exchange exchange, Exchange copy) throws Exception { 1360 Message in = copy.getIn(); 1361 Object actualBody = in.getBody(); 1362 1363 if (expectedHeaderValues != null) { 1364 if (actualHeaderValues == null) { 1365 actualHeaderValues = getCamelContext().getHeadersMapFactory().newMap(); 1366 } 1367 if (in.hasHeaders()) { 1368 actualHeaderValues.putAll(in.getHeaders()); 1369 } 1370 } 1371 1372 if (expectedPropertyValues != null) { 1373 if (actualPropertyValues == null) { 1374 actualPropertyValues = getCamelContext().getHeadersMapFactory().newMap(); 1375 } 1376 actualPropertyValues.putAll(copy.getProperties()); 1377 } 1378 1379 if (expectedBodyValues != null) { 1380 int index = actualBodyValues.size(); 1381 if (expectedBodyValues.size() > index) { 1382 Object expectedBody = expectedBodyValues.get(index); 1383 if (expectedBody != null) { 1384 // prefer to convert body early, for example when using files 1385 // we need to read the content at this time 1386 Object body = in.getBody(expectedBody.getClass()); 1387 if (body != null) { 1388 actualBody = body; 1389 } 1390 } 1391 actualBodyValues.add(actualBody); 1392 } 1393 } 1394 1395 // let counter be 0 index-based in the logs 1396 if (LOG.isDebugEnabled()) { 1397 String msg = getEndpointUri() + " >>>> " + counter + " : " + copy + " with body: " + actualBody; 1398 if (copy.getIn().hasHeaders()) { 1399 msg += " and headers:" + copy.getIn().getHeaders(); 1400 } 1401 LOG.debug(msg); 1402 } 1403 1404 // record timestamp when exchange was received 1405 copy.setProperty(Exchange.RECEIVED_TIMESTAMP, new Date()); 1406 1407 // add a copy of the received exchange 1408 addReceivedExchange(copy); 1409 // and then increment counter after adding received exchange 1410 ++counter; 1411 1412 Processor processor = processors.get(getReceivedCounter()) != null 1413 ? processors.get(getReceivedCounter()) : defaultProcessor; 1414 1415 if (processor != null) { 1416 try { 1417 // must process the incoming exchange and NOT the copy as the idea 1418 // is the end user can manipulate the exchange 1419 processor.process(exchange); 1420 } catch (Exception e) { 1421 // set exceptions on exchange so we can throw exceptions to simulate errors 1422 exchange.setException(e); 1423 } 1424 } 1425 } 1426 1427 /** 1428 * Adds the received exchange. 1429 * 1430 * @param copy a copy of the received exchange 1431 */ 1432 protected void addReceivedExchange(Exchange copy) { 1433 if (retainFirst == 0 && retainLast == 0) { 1434 // do not retain any messages at all 1435 } else if (retainFirst < 0 && retainLast < 0) { 1436 // no limitation so keep them all 1437 receivedExchanges.add(copy); 1438 } else { 1439 // okay there is some sort of limitations, so figure out what to retain 1440 if (retainFirst > 0 && counter < retainFirst) { 1441 // store a copy as its within the retain first limitation 1442 receivedExchanges.add(copy); 1443 } else if (retainLast > 0) { 1444 // remove the oldest from the last retained boundary, 1445 int index = receivedExchanges.size() - retainLast; 1446 if (index >= 0) { 1447 // but must be outside the first range as well 1448 // otherwise we should not remove the oldest 1449 if (retainFirst <= 0 || retainFirst <= index) { 1450 receivedExchanges.remove(index); 1451 } 1452 } 1453 // store a copy of the last n'th received 1454 receivedExchanges.add(copy); 1455 } 1456 } 1457 } 1458 1459 protected void waitForCompleteLatch() throws InterruptedException { 1460 if (latch == null) { 1461 fail("Should have a latch!"); 1462 } 1463 1464 StopWatch watch = new StopWatch(); 1465 waitForCompleteLatch(resultWaitTime); 1466 long delta = watch.taken(); 1467 LOG.debug("Took {} millis to complete latch", delta); 1468 1469 if (resultMinimumWaitTime > 0 && delta < resultMinimumWaitTime) { 1470 fail("Expected minimum " + resultMinimumWaitTime 1471 + " millis waiting on the result, but was faster with " + delta + " millis."); 1472 } 1473 } 1474 1475 protected void waitForCompleteLatch(long timeout) throws InterruptedException { 1476 // Wait for a default 10 seconds if resultWaitTime is not set 1477 long waitTime = timeout == 0 ? 10000L : timeout; 1478 1479 // now let's wait for the results 1480 LOG.debug("Waiting on the latch for: {} millis", timeout); 1481 latch.await(waitTime, TimeUnit.MILLISECONDS); 1482 } 1483 1484 protected void assertEquals(String message, Object expectedValue, Object actualValue) { 1485 if (!ObjectHelper.equal(expectedValue, actualValue)) { 1486 fail(message + ". Expected: <" + expectedValue + "> but was: <" + actualValue + ">"); 1487 } 1488 } 1489 1490 protected void assertTrue(String message, boolean predicate) { 1491 if (!predicate) { 1492 fail(message); 1493 } 1494 } 1495 1496 protected void fail(Object message) { 1497 if (LOG.isDebugEnabled()) { 1498 List<Exchange> list = getReceivedExchanges(); 1499 int index = 0; 1500 for (Exchange exchange : list) { 1501 LOG.debug("{} failed and received[{}]: {}", getEndpointUri(), ++index, exchange); 1502 } 1503 } 1504 throw new AssertionError(getEndpointUri() + " " + message); 1505 } 1506 1507 public int getExpectedMinimumCount() { 1508 return expectedMinimumCount; 1509 } 1510 1511 public void await() throws InterruptedException { 1512 if (latch != null) { 1513 latch.await(); 1514 } 1515 } 1516 1517 public boolean await(long timeout, TimeUnit unit) throws InterruptedException { 1518 if (latch != null) { 1519 return latch.await(timeout, unit); 1520 } 1521 return true; 1522 } 1523 1524 public boolean isSingleton() { 1525 return true; 1526 } 1527 1528 public boolean isLenientProperties() { 1529 return true; 1530 } 1531 1532 private Exchange getReceivedExchange(int index) { 1533 if (index <= receivedExchanges.size() - 1) { 1534 return receivedExchanges.get(index); 1535 } else { 1536 return null; 1537 } 1538 } 1539 1540}