Chapter 272. Reactive Streams Component

reactive-streams://stream?[options]

Where stream is a logical stream name used to bind Camel routes to the external stream processing systems.

The Reactive Streams component supports 4 options, which are listed below.

The Reactive Streams endpoint is configured using URI syntax:

reactive-streams:stream

with the following path and query parameters:

The component supports 7 options, which are listed below.

The library is aimed to support all the communication modes needed by an application to interact with Camel data:

In order to subscribe to data flowing from a Camel route, exchanges should be redirected to a named stream, like in the following snippet:

from("timer:clock")
.setBody().header(Exchange.TIMER_COUNTER)
.to("reactive-streams:numbers");

Routes can also be written using the XML DSL.

In the example, an unbounded stream of numbers is associated to the name numbers. The stream can be accessed using the CamelReactiveStreams utility class.

CamelReactiveStreamsService camel = CamelReactiveStreams.get(context);

// Getting a stream of exchanges
Publisher<Exchange> exchanges = camel.fromStream("numbers");

// Getting a stream of Integers (using Camel standard conversion system)
Publisher<Integer> numbers = camel.fromStream("numbers", Integer.class);

The stream can be used easily with any reactive streams compatible library. Here is an example of how to use it with RxJava 2 (although any reactive framework can be used to process events).

Flowable.fromPublisher(integers)
    .doOnNext(System.out::println)
    .subscribe();

The example prints all numbers generated by Camel into System.out.

CamelReactiveStreamsService camel = CamelReactiveStreams.get(context);

// Get a stream from all the files in a directory
Publisher<String> files = camel.from("file:folder", String.class);

// Use the stream in RxJava2
Flowable.fromPublisher(files)
    .doOnNext(System.out::println)
    .subscribe();

When an external library needs to push events into a Camel route, the Reactive Streams endpoint must be set as consumer.

from("reactive-streams:elements")
.to("log:INFO");

A handle to the elements stream can be obtained from the CamelReactiveStreams utility class.

CamelReactiveStreamsService camel = CamelReactiveStreams.get(context);

Subscriber<String> elements = camel.streamSubscriber("elements", String.class);

The subscriber can be used to push events to the Camel route that consumes from the elements stream.

Here is an example of how to use it with RxJava 2 (although any reactive framework can be used to publish events).

Flowable.interval(1, TimeUnit.SECONDS)
    .map(i -> "Item " + i)
    .subscribe(elements);

String items are generated every second by RxJava in the example and they are pushed into the Camel route defined above.

CamelReactiveStreamsService camel = CamelReactiveStreams.get(context);

// Send two strings to the "seda:queue" endpoint
Flowable.just("hello", "world")
    .subscribe(camel.subscriber("seda:queue", String.class));

Routes defined in some Camel DSL can be used within a reactive stream framework to perform a specific transformation (the same mechanism can be also used to eg. just send data to a http endpoint and continue).

The following snippet shows how RxJava functional code can request the task of loading and marshalling files to Camel.

CamelReactiveStreamsService camel = CamelReactiveStreams.get(context);

// Process files starting from their names
Flowable.just(new File("file1.txt"), new File("file2.txt"))
    .flatMap(file -> camel.toStream("readAndMarshal", String.class))
    // Camel output will be converted to String
    // other steps
    .subscribe();

In order this to work, a route like the following should be defined in the Camel context:

from("reactive-streams:readAndMarshal")
.marshal() // ... other details

CamelReactiveStreamsService camel = CamelReactiveStreams.get(context);

// Process files starting from their names
Flowable.just(new File("file1.txt"), new File("file2.txt"))
    .flatMap(file -> camel.to("direct:process", String.class))
    // Camel output will be converted to String
    // other steps
    .subscribe();

from("direct:process")
.marshal() // ... other details

While a reactive streams Publisher allows exchanging data in a unidirectional way, Camel routes often use a in-out exchange pattern (eg. to define REST endpoints and, in general, where a reply is needed for each request).

In these circumstances, users can add a reactive processing step to the flow, to enhance a Camel route or to define the entire transformation using the reactive framework.

For example, given the following route:

from("timer:clock")
.setBody().header(Exchange.TIMER_COUNTER)
.to("direct:reactive")
.log("Continue with Camel route... n=${body}");

A reactive processing step can be associated to the "direct:reactive" endpoint:

CamelReactiveStreamsService camel = CamelReactiveStreams.get(context);

camel.process("direct:reactive", Integer.class, items ->
    Flowable.fromPublisher(items) // RxJava2
        .map(n -> -n)); // make every number negative

Data flowing in the Camel route will be processed by the external reactive framework then continue the processing flow inside Camel.

This mechanism can also be used to define a In-Out exchange in a completely reactive way.

CamelReactiveStreamsService camel = CamelReactiveStreams.get(context);

// requires a rest-capable Camel component
camel.process("rest:get:orders", exchange ->
                    Flowable.fromPublisher(exchange)
                            .flatMap(ex -> allOrders())); // retrieve orders asynchronously

See Camel examples (camel-example-reactive-streams) for details.

A starter module is available to spring-boot users. When using the starter, the CamelReactiveStreamsService can be directly injected into Spring components.

To use the starter, add the following to your spring boot pom.xml file:

<dependency>
    <groupId>org.apache.camel</groupId>
    <artifactId>camel-reactive-streams-starter</artifactId>
    <version>${camel.version}</version> <!-- use the same version as your Camel core version -->
</dependency>