svn commit: r985535 - /servicemix/smx4/features/trunk/camel/servicemix-camel/src/test/java/org/apache/servicemix/camel/nmr/CamelAsyncRouteTest.java

[ge...@apache.org]

Author: gertv
Date: Sat Aug 14 16:54:09 2010
New Revision: 985535

URL: http://svn.apache.org/viewvc?rev=985535&view=rev
Log:
Fix CI build failures for CamelAsyncRouteTest

Modified:
    servicemix/smx4/features/trunk/camel/servicemix-camel/src/test/java/org/apache/servicemix/camel/nmr/CamelAsyncRouteTest.java

Modified: servicemix/smx4/features/trunk/camel/servicemix-camel/src/test/java/org/apache/servicemix/camel/nmr/CamelAsyncRouteTest.java
URL: http://svn.apache.org/viewvc/servicemix/smx4/features/trunk/camel/servicemix-camel/src/test/java/org/apache/servicemix/camel/nmr/CamelAsyncRouteTest.java?rev=985535&r1=985534&r2=985535&view=diff
==============================================================================
--- servicemix/smx4/features/trunk/camel/servicemix-camel/src/test/java/org/apache/servicemix/camel/nmr/CamelAsyncRouteTest.java (original)
+++ servicemix/smx4/features/trunk/camel/servicemix-camel/src/test/java/org/apache/servicemix/camel/nmr/CamelAsyncRouteTest.java Sat Aug 14 16:54:09 2010
@@ -32,7 +32,9 @@ import org.apache.servicemix.nmr.api.Sta
 public class CamelAsyncRouteTest extends AbstractComponentTest {
 
     private static final String HANDLED_BY_THREAD = "HandledByThread";
+    
     private static final int COUNT = 1000;
+    private static final long DELAY = 60000;
 
     /* Latch to count NMR Done Exchanges */
     private CountDownLatch done;
@@ -45,40 +47,41 @@ public class CamelAsyncRouteTest extends
     }
 
     public void testCamelThreads() throws InterruptedException {
-        MockEndpoint mock = getMockEndpoint("mock:threads");
-        mock.expectedMessageCount(COUNT);
-
-        getMockEndpoint("mock:sent").expectedMessageCount(COUNT);
-
+        expectDefaultMessageCount("mock:sent");
+        expectDefaultMessageCount("mock:threads").whenAnyExchangeReceived(new AssertHandledByCamelThreadProcessor());
+        
         for (int i = 0 ; i < COUNT ; i++) {
-            template.asyncSendBody("direct:threads", "Simple message body");
+            template.asyncSendBody("direct:threads", "Simple message body " + i);
         }
 
         assertMockEndpointsSatisfied();
 
         assertTrue("All NMR exchanges should have been marked DONE",
-                   done.await(20, TimeUnit.SECONDS));                
-        
-        for (Exchange exchange : mock.getExchanges()) {
-            Thread thread = exchange.getProperty(HANDLED_BY_THREAD, Thread.class);
-            assertTrue("processor should have been called from the Camel 'threads' thread pool instead of " + thread.getName(),
-                       thread.getName().contains("Camel") && thread.getName().contains("Thread"));
-        }
-
+                   done.await(DELAY, TimeUnit.MILLISECONDS));
     }
 
-    public void testCamelSeda() throws InterruptedException {       
-        getMockEndpoint("mock:sent").expectedMessageCount(COUNT);
-        getMockEndpoint("mock:seda").expectedMessageCount(COUNT);
+    public void testCamelSeda() throws InterruptedException {
+        expectDefaultMessageCount("mock:sent");
+        expectDefaultMessageCount("mock:seda");
 
         for (int i = 0 ; i < COUNT ; i++) {
-            template.asyncSendBody("seda:seda", "Simple message body");
+            template.asyncSendBody("seda:seda", "Simple message body " + i);
         }
 
         assertMockEndpointsSatisfied();
 
         assertTrue("All NMR exchanges should have been marked DONE",
-                   done.await(20, TimeUnit.SECONDS));
+                   done.await(DELAY, TimeUnit.MILLISECONDS));
+    }
+
+    /*
+     * Configure the mock endpoint to expect {@value #COUNT} messages to arrive in {@value #DELAY}ms
+     */
+    private MockEndpoint expectDefaultMessageCount(String endpoint) {
+        final MockEndpoint mock = getMockEndpoint(endpoint);
+        mock.setResultWaitTime(DELAY);
+        mock.expectedMessageCount(COUNT);
+        return mock;
     }
 
     @Override
@@ -111,4 +114,17 @@ public class CamelAsyncRouteTest extends
             done.countDown();
         }
     }
+
+    /*
+     * Processor to ensure that the exchange has been handled by a Camel thread instead of an NMR thread
+     */
+    private static final class AssertHandledByCamelThreadProcessor implements Processor {
+
+        public void process(Exchange exchange) throws Exception {
+            Thread thread = exchange.getProperty(HANDLED_BY_THREAD, Thread.class);
+            assertTrue("processor should have been called from the Camel 'threads' thread pool instead of " + thread.getName(),
+                       thread.getName().contains("Camel") && thread.getName().contains("Thread"));
+        }
+
+    }
 }