273.5. 从 Camel 获取数据

为了订阅来自 Camel 路由的数据流，应将交换重定向到命名流，如以下片段所示：

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

也可以使用 XML DSL 编写路由。

在示例中，未绑定的数字流与名称 号 关联。该流可以使用 CamelReactiveStreams 实用程序类来访问。

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);

流可用于任何兼容被动流的被动流。以下是如何将它与 RxJava 2 搭配使用的示例（尽管任何被动框架可用于处理事件）。

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

这个示例将 Camel 生成的所有数字打印到 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();