[CAMEL-7833] create an extension of the RX Observable<T> to add more of the Camel DSL in there (e.g. to() or to go back to the general camel DSL)

• Type: New Feature
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 2.16.0
• Estimated Complexity: Unknown
• Assignee: Willem Jiang
• Reporter: James Strachan
• Created: 17/Sep/14 12:50
• Updated: 06/Apr/15 13:53
• Resolved: 02/Apr/15 07:52

Description

with Camel RX and java 8 we can do some nice lambdas and typesafe filtering and transformation:

ReactiveCamel rx = new ReactiveCamel(camelContext);
Observable<Order> observable = rx.toObservable("seda:orders", Order.class);
 
// now lets filter and map using Java 8
Observable<String> largeOrderIds = observable.
  filter(order -> order.getAmount() > 100.0).
  map(order -> order.getId());
 
rx.sendTo(observable, "activemq:MyQueue");

however the DSL isn't quite as nice as Camel's due to the lack of the camel verbs like "to()". It'd be nice to provide an extended Observable<T> interface which adds more of Camel's DSL in there too; so you can still use things the RX way; but can also reuse the camel DSL too (in a typesafe way).

e.g. something like this

// Observable comes from RX; it'd be nice to have a camel extended version so we can add camel DSL stuff in there like to() etc....
CamelStream<Order> orders = rx.stream("seda:orders", Order.class);
 
// now lets filter and map using Java 8
orders.
  filter(order -> order.getAmount() > 100.0).
  map(order -> order.getId()).
  to("activemq:MyQueue");

Issue Links

• relates to: CAMEL-7832 create a java 8 demo of RX to show typesafe filtering and transforming of messages