Apache Camelとリアクティブプログラミング

# 概要

- リアクティブプログラミングとは
- Camel + Reactive Streams
- Camel + Vert.x

----

# リアクティブプログラミングとは

https://www.reactivemanifesto.org/ja

- リアクティブシステムの特徴
    - **即応性（Responsive）**― 即時に応答を返す
    - **耐障害性（Resilient）**― 障害があっても応答する
    - **弾力性（Elastic）**― 高負荷であっても応答する
    - **メッセージ駆動（Message Driven）**― 疎結合な非同期通信に基づく

----

## バックプレッシャー（Back pressure）

https://www.reactivemanifesto.org/ja/glossary#Back-Pressure

> ❝あるコンポーネントが全体に追いつけなくなった場合、システム全体として何らかの対処をする必要がある。過負荷状態のコンポーネントが壊滅的にクラッシュしたり、制御無くメッセージを損失することは許されない。処理が追いつかなくなっていて、かつクラッシュすることも許されないならば、コンポーネントは上流のコンポーネント群に自身が過負荷状態であることを伝えて負荷を減らしてもらうべきだ。このバック・プレッシャー (back-pressure) と呼ばれる仕組みは、過負荷の下でシステムを崩壊させず緩やかに応答を続ける重要なフィードバック機構だ。バック・プレッシャーはユーザまで転送してもよく、その場合、即応性 (resilient) は低下するが負荷の下でのシステムの耐障害性が保証される。また、システムがその情報を使って自身に他のリソースを振り向け、負荷分散を促すこともできる。❞

----

## Reactive Streamsの例

http://www.reactive-streams.org/

- JDK 9の`java.util.concurrent.Flow`と1:1に対応
- back pressureのサポート

```java
import org.reactivestreams.*;
import io.reactivex.rxjava3.core.*;

// Hello world
Flowable.just("Hello World").subscribe(System.out::println);

// Publisher/Subscriber example
Publisher<Integer> publisher = Flowable.just(1, 2, 3, 4, 5);
Flowable.fromPublisher(publisher)
    .map(i -> i*i)
    .doOnNext(System.out::println)
    .subscribe();
```

----

# Camel + Reactive Streams

https://camel.apache.org/components/latest/reactive-streams-component.html

- CamelルートとReactive Streamsフローとの双方向の連携を可能にする
  - `reactive-streams:xxxxx[?options]`
- 以下のリアクティブフレームワーク実装に対応
    - [Reactor Core](https://github.com/reactor/reactor-core)
    - [RxJava](https://github.com/ReactiveX/RxJava)
- Back pressureにも対応

----

## Camel (pub) → Reactive Streams (sub)

- Camelのプロデューサーエンドポイント（`to`）からReactive Streamsのフローへ

```java
CamelContext camel = new DefaultCamelContext();
CamelReactiveStreamsService rsCamel = CamelReactiveStreams.get(camel);
camel.start();

// Camel (pub) → Reactive Streams (sub)
//Publisher<Integer> publisher = rsCamel.from("seda:input", Integer.class);
Publisher<Integer> publisher = rsCamel.fromStream("input", Integer.class);
Flowable.fromPublisher(publisher)
    .map(i -> i*i)
    .doOnNext(System.out::println)
    .subscribe();

FluentProducerTemplate template = camel.createFluentProducerTemplate();
IntStream.rangeClosed(1, 5).forEach(i ->
    //template.withBody(i).to("seda:input").send());
    template.withBody(i).to("reactive-streams:input").send());
```

----

## Reactive Streams (pub) → Camel (sub)

- Reactive StreamsのフローからCamelのコンシューマーエンドポイント（`from`）へ

```java
CamelContext camel = new DefaultCamelContext();
camel.addRoutes(new RouteBuilder() {
    public void configure() {
        from("reactive-streams:input")
            .to("stream:out");
    }
});
CamelReactiveStreamsService rsCamel = CamelReactiveStreams.get(camel);
camel.start();

// Reactive Streams (pub) → Camel (sub)
Flowable.just(1, 2, 3, 4, 5)
    .map(i -> i*i)
    .subscribe(rsCamel.streamSubscriber("input", Integer.class));
```

----

## CamelルートをReactive Streamsフローで実装

- ルートとフローのミックス実装もできる

```java
CamelContext camel = new DefaultCamelContext();
camel.addRoutes(new RouteBuilder() {
    public void configure() {
        from("reactive-streams:input")
            .to("stream:out");
        from("direct:output")
            .log("Done: ${body}");
    }
});
CamelReactiveStreamsService rsCamel = CamelReactiveStreams.get(camel);
camel.start();

// Camel route & Reactive Streams flow
Flowable.fromPublisher(rsCamel.from("seda:input"))
    .map(e -> {
        int i = e.getMessage().getBody(Integer.class);
        e.getMessage().setBody(i * i);
        return e;
    })
    .doOnNext(e -> rsCamel.to("stream:out", e))
    .subscribe(rsCamel.subscriber("direct:output"));

FluentProducerTemplate template = camel.createFluentProducerTemplate();
IntStream.rangeClosed(1, 5).forEach(i ->
    template.withBody(i).to("seda:input").send());
```

----

## CamelのBack pressureサポート

- 処理中（in-flight）エクスチェンジの数を制限することでback pressureを実現できる

### プロデューサー側

- `ThrottlingInflightRoutePolicy`ポリシー
    - スロットル中はルートが一時停止する

```java
ThrottlingInflightRoutePolicy policy = new ThrottlingInflightRoutePolicy();
policy.setMaxInflightExchanges(10);

from("jms:queue1")
    .routePolicy(policy)
    .to("reactive-streams:flow1");
```
- `backpressureStrategy`オプション
    - `BUFFER`, `LATEST`, `OLDEST`

```java
from("jms:queue2")
    .to("reactive-streams:flow2?backpressureStrategy=LATEST");
```

### コンシューマー側

- `maxInflightExchanges`オプション

```java
from("reactive-streams:numbers?maxInflightExchanges=10")
    .to("direct:endpoint");
```

----

# Camel + Vert.x

https://vertx.io/

- Vert.xのEventBusとイベントのやり取りを可能にする
  - `vertx:xxxxx[?options]`

----

## Vert.xとは

- JVM版のNode.jsのようなもの
  - Verticle（= Actor）とイベントバス

```java
import io.vertx.core.Vertx;
import io.vertx.core.eventbus.EventBus;

class HttpServer {
    public static void main(String[] args) {
        Vertx vertx = Vertx.vertx();
        vertx.deployVerticle(new LoggingVerticle());
        EventBus eventBus = vertx.eventBus();
        vertx.createHttpServer()
            .requestHandler(req -> {
                eventBus.send("logging", "Got request: " + req.method() + " " + req.uri());
                req.response().end("Hello!");
            })
            .listen(8888);

        // ... クローズ処理
    }
}

class LoggingVerticle extends AbstractVerticle {
    public void start() {
        vertx.eventBus().<String>consumer("logging",
            message -> log.info(message.body()));
    }
}
```

----

## Camel Vert.x

- CamelからVert.xのイベントバス、Vert.xイベントバスからCamelへの双方向のメッセージングが可能

```java
class CamelHttpServer {
    public static void main(String[] args) {
        Vertx vertx = Vertx.vertx();
        vertx.deployVerticle(new LoggingVerticle());
        EventBus eventBus = vertx.eventBus();
        vertx.createHttpServer()
            .requestHandler(req -> {
                eventBus.send("camel-logging", "Got request: " + req.method() + " " + req.uri());
                req.response().end("Hello!");
            })
            .listen(8888);

        CamelContext camel = new DefaultCamelContext();
        camel.getComponent("vertx", VertxComponent.class).setVertx(vertx);
        camel.addRoutes(new RouteBuilder() {
            public void configure() {
                // Vert.x → Camel
                from("vertx:camel-logging")
                    .log("body = ${body}")
                    .to("direct:logging");
                // Camel → Vert.x
                from("direct:logging")
                    .to("vertx:logging");
            }
        });
        camel.start();

        // ... クローズ処理
    }
}

class LoggingVerticle extends AbstractVerticle {
    public void start() {
        vertx.eventBus().<String>consumer("logging",
            message -> log.info(message.body()));
    }
}
```

----

## Vert.x関連コンポーネント

| コンポーネント | 説明 |
|:-------------|:----|
| [camel-vertx-http](https://camel.apache.org/components/latest/vertx-http-component.html) | [Vert.x Web Client](https://vertx.io/docs/vertx-web-client/java/)によるHTTPエンドポイント実装（Producerのみ） |
| [camel-vertx-kafka](https://camel.apache.org/components/latest/vertx-kafka-component.html) | [Vert.x Kafka Client](https://vertx.io/docs/vertx-kafka-client/java/)によるKafkaとの接続  |
| [camel-vertx-websocket](https://camel.apache.org/components/latest/vertx-websocket-component.html) | Vert.x WebSocket機能によるWebSocketエンドポイントの実装 |
| [Vert.x Camel Bridge](https://vertx.io/docs/vertx-camel-bridge/java/) | CamelでなくVert.x側からのCamelへのブリッジ（ちなみに作者はCamelのClaus） |

----

# 参考文献

- [Camel in Action, 2nd Edition](https://www.manning.com/books/camel-in-action-second-edition) &mdash; Chapter 20: Reactive Camel

----

# END