THIS IS A TEST INSTANCE. ALL YOUR CHANGES WILL BE LOST!!!!
...
You can then create an Observable<Message> from any endpoint using the ReactiveCamel helper class.
Code Block |
---|
import org.apache.camel.rx.*;
ReactiveCamel rx = new ReactiveCamel(camelContext);
Observable<Message> observable = rx.toObservable("activemq:MyMessages");
// we can now call filter/map/concat etc
filtered = observable.filter(m -> m.getHeader("foo") != null);
|