Reactive streams with Camel, knowing when the flow completes (normally or with error)

[Artur Jablonski <aj...@ravenpack.com>]

Hello,

I was following the Camel In Action 2 (version 14) chapter on Reactive
streams and Camel (chapter 21).

I followed section 21.1 about bridging Camel routes and JavaRx reactive
streams, did some experimenting and there's one thing that I scratch my
head over.

Namely, is there a way to wait/be notified when camel routes turned into
reactive stream complete (normally or with an exception).

So if I have these routes:

from("direct:in1")
    .routeId("1")
    .to("reactive-streams:out1")
    .log("1 done");


from("reactive-streams:in2")
    .routeId("2")
    .to("reactive-streams:out2")
    .log("2 done");


from("reactive-streams:in3")
    .routeId("3")
    .log("3 done");

and then if I glue them using JavaRx and CamelReactiveStreamService:

Flowable.fromPublisher(rxCamel.fromStream("out1"))
        .subscribe(rxCamel.streamSubscriber("in2"));

Flowable.fromPublisher(rxCamel.fromStream("out2"))
        .subscribe(rxCamel.streamSubscriber("in3"));

and now if I call the route "1" using ProducerTemplate:

template.requestBody("direct:in1", "payload");
logger.info("requestBody() returned");

the output is:


*1 done*
*requestBody() returned*
*2 done*
*3 done*

So the requestBody() returns after the route "1" passes the data to
"reactive-streams:out1".

What if I want to wait/be notified that the exchange reached the end of the
reactive stream and all went good or if somewhere on the way there has been
an error? Is that possible? Where do I look?

Disclaimer: I am new to reactive streams.

Best,
Artur

PS: Thanks Claus and Jonathan for the book. It's awesome!