CAMEL-7833 Added convenience method ReactiveCamel.to(...).

Changed test to highlight the idea of creating routes with RX.
apache · Apr 3, 2015 · 3c3b0ed · 3c3b0ed
1 parent c67392f
commit 3c3b0ed
Show file tree

Hide file tree

Showing 2 changed files with 27 additions and 4 deletions.
diff --git a/components/camel-rx/src/main/java/org/apache/camel/rx/ReactiveCamel.java b/components/camel-rx/src/main/java/org/apache/camel/rx/ReactiveCamel.java
@@ -92,6 +92,20 @@ public <T> void sendTo(Observable<T> observable, Endpoint endpoint) {
         }
     }
 
+    /**
+     * Convenience method for creating CamelOperator instances
+     */
+    public CamelOperator to(String uri) throws Exception {
+        return new CamelOperator(camelContext, uri);
+    }
+
+    /**
+     * Convenience method for creating CamelOperator instances
+     */
+    public CamelOperator to(Endpoint endpoint) throws Exception {
+        return new CamelOperator(endpoint);
+    }
+
     public CamelContext getCamelContext() {
         return camelContext;
     }

diff --git a/components/camel-rx/src/test/java/org/apache/camel/rx/CamelOperatorTest.java b/components/camel-rx/src/test/java/org/apache/camel/rx/CamelOperatorTest.java
@@ -24,6 +24,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import rx.Observable;
+import rx.Subscription;
+import rx.observables.ConnectableObservable;
 
 /**
  */
@@ -39,12 +41,16 @@ public void testCamelOperator() throws Exception {
         mockEndpoint2.expectedMessageCount(1);
         mockEndpoint3.expectedMessageCount(1);
 
-        Observable<Message> result = reactiveCamel.toObservable("direct:start")
-            .lift(new CamelOperator(camelContext, "mock:results1"))
+        ConnectableObservable<Message> route = reactiveCamel.toObservable("direct:start")
+            .lift(new CamelOperator(mockEndpoint1))
             .lift(new CamelOperator(camelContext, "log:foo"))
             .debounce(1, TimeUnit.SECONDS)
-            .lift(new CamelOperator(mockEndpoint2));
-        reactiveCamel.sendTo(result, "mock:results3");
+            .lift(reactiveCamel.to(mockEndpoint2))
+            .lift(reactiveCamel.to("mock:results3"))
+            .publish();
+
+        // Start the route
+        Subscription routeSubscription = route.connect();
 
         // Send two test messages
         producerTemplate.sendBody("direct:start", "<test/>");
@@ -53,5 +59,8 @@ public void testCamelOperator() throws Exception {
         mockEndpoint1.assertIsSatisfied();
         mockEndpoint2.assertIsSatisfied();
         mockEndpoint3.assertIsSatisfied();
+
+        // Stop the route
+        routeSubscription.unsubscribe();
     }
 }