Merge pull request #128 from smecsia/master

Migrate to newer pessimistic mongo
camelot-framework · Nov 20, 2015 · bd73ca9 · bd73ca9
2 parents 783e9b5 + 368b0d1
commit bd73ca9
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 4 deletions.
diff --git a/camelot-ext/camelot-mongodb/pom.xml b/camelot-ext/camelot-mongodb/pom.xml
@@ -10,7 +10,7 @@
     <name>Camelot MongoDB Extension</name>
     <artifactId>camelot-mongodb</artifactId>
     <properties>
-        <pessimistic-mongodb.version>1.2</pessimistic-mongodb.version>
+        <pessimistic-mongodb.version>1.3</pessimistic-mongodb.version>
     </properties>
 
     <dependencies>

diff --git a/camelot-test/src/main/java/ru/yandex/qatools/camelot/test/core/CamelotTestListener.java b/camelot-test/src/main/java/ru/yandex/qatools/camelot/test/core/CamelotTestListener.java
@@ -27,6 +27,7 @@
 
 import static java.lang.String.format;
 import static java.lang.System.currentTimeMillis;
+import static java.lang.Thread.sleep;
 import static java.util.stream.Collectors.toList;
 import static org.apache.camel.util.CamelContextHelper.getEndpointInjection;
 import static org.apache.camel.util.ObjectHelper.isEmpty;
@@ -163,9 +164,14 @@ private void clearContext(TestContext testContext) throws Exception { //NOSONAR
             while (!camelContext.getInflightRepository().browse().isEmpty() &&
                     currentTimeMillis() - waitStartedTime < MAX_INFLIGHT_WAIT_MS) {
                 camelContext.getInflightRepository().browse().stream().collect(toList()).forEach(e -> {
-                    camelContext.getInflightRepository().remove(e.getExchange());
-                    logger.warn("Removing inflight exchange {} for route {}",
-                            e.getExchange().getExchangeId(), e.getRouteId());
+                    try {
+                        camelContext.getInflightRepository().remove(e.getExchange());
+                        logger.warn("Removing inflight exchange {} for route {}",
+                                e.getExchange().getExchangeId(), e.getRouteId());
+                        sleep(20); // Giving some time for camel to fetch new inflights
+                    } catch (InterruptedException ignored) {
+                        throw new RuntimeException(ignored);
+                    }
                 });
             }
             try {