Reactor Core 2.5: もう一つのJava向けReactive Extensions実装

Reactive Extensionの調査中に[Reactor Core 2.5](http://projectreactor.io/)が面白いことに気がついたのでまとめ。

* [JavaDoc](http://projectreactor.io/core/docs/api/)
* [GitHub](https://github.com/reactor/reactor-core)
* [A lite Rx API for the JVM by Sébastien Deleuze](https://speakerdeck.com/sdeleuze/a-lite-rx-api-for-the-jvm) (Spring I/O 2016での発表資料)

# Reactor Core (2.5) = Reactive Streams + Reactive Extensions(の一部)

Reactor(〜2.0)は効率的な非同期プログラミングのためのツールキットで、基本的な関数型(Java8以前に開発されたため)に始まって高効率なスケジューラー上に実装された[Reactive Streams](http://www.reactive-streams.org/)やその[Reactive Extensions](http://reactivex.io/)(Rx)、非同期のネットワークライブラリ等を提供していました。

ところがReactor 2.5を開発するにあたって[プロジェクトの構成をガラガラポン](https://spring.io/blog/2016/02/16/reactor-2-5-a-second-generation-reactive-foundation-for-the-jvm)したらしく、結果としてReactor Core(2.5)単体はReactive Streamsに準拠した軽量なRxライブラリに仕上がっています(依存性はreactive-streamsのみ)。

(RxJavaの主要コミッターであるDávid Karnok曰く現在のRxJavaは第二世代のリアクティブライブラリなのに対して[Reactor 2.5+は第四世代](https://github.com/google/agera/issues/20#issuecomment-212007539)との解説)

Spring Frameworkの次期バージョン5ではこのReactorがSpring MVCのリアクティブ対応に用いられるようです。

現在のバージョンはマイルストーンリリース4で、利用するにはspring.ioのスナップショットリポジトリを参照する必要があります。


```xml:pom.xml
  <repositories>
    <repository>
      <id>spring-snapshot</id>
      <name>Spring Snapshot Repository</name>
      <url>https://repo.spring.io/snapshot</url>
      <snapshots>
        <enabled>true</enabled>
      </snapshots>
    </repository>
    ...
  </repositories>

  <dependencies>
    <dependency>
      <groupId>io.projectreactor</groupId>
      <artifactId>reactor-core</artifactId>
      <version>2.5.0.M4</version>
    </dependency>
    ...
  </dependencies>
```

リファレンスマニュアル等が依然として書かれていないのがプロジェクトの進捗としては気になるところですが、今後充実すると期待したいところです。

# ``Flux``と``Mono``: Reactive Streams準拠な``Observable``

Rxの``Observable``の概念を説明するのによく以下のような[表](http://reactivex.io/intro.html)が用いられます。

|       | Single item   | Multiple items    |
|-------|---------------|-------------------|
| 同期  | ``T``         | ``Iterable<T>``   |
| 非同期 | ``Future<T>`` | ``Observable<T>`` |

つまり``Observable``は非同期に出力されるデータの列を抽象するクラスです。

これに対して、Reactor Core(2.5)で提供されるクラスは以下のように対応します。

|                       | Single item | Multiple items |
|-----------------------|-------------|----------------|
| Reactor core (非同期) | ``Mono<T>`` | ``Flux<T>``    |

``Observable``相当の``Flux``に加えて、最大1個の出力を返す``Mono``が用意されているのがポイントです。共にRxで定義されているAPIのサブセットを実装しています。

最も重要な点として、``Flux``も``Mono``も、Reactive Streamsの``Publisher``を実装しています。

```java
public abstract class Flux<T> implements Publisher<T>, Introspectable, Backpressurable{...}

public abstract class Mono<T> implements Publisher<T>, Backpressurable, Introspectable {...}
```

``Flux``も``Mono``も``Publisher``から直接作成出来るほか、``flatMap``等の重要な操作の多くも``Flux``等ではなく``Publisher``を引数に用います。

```java
public static <T> Flux<T> from(Publisher<? extends T> source) {...}

public final <R> Flux<R> flatMap(Function<? super T, ? extends Publisher<? extends R>> mapper) {...}
```

Reactor CoreはReactive Streamsの上にReactive ExtensionsのAPIを被せるラッパーのように見ることが出来るかと思います。

# ``Mono``は便利

``Mono``は最大1個の値を出力する``Publisher``です。前掲の表でも対応していたJDKのFutureのように、単発の値を返す非同期処理の戻り値として使うと便利です。

## ``Mono``はPromise風

最近は特にAndroidアプリの開発にRxJavaを導入する事例が多く報告されていますが、その理由の一つとして``Observable``をPromiseとして用いることでUI周りの非同期処理の書き方を統一したいという事情があるかと思います。

``Mono``はこの用途に適しています。単一の出力値しか扱わないため``Observable``よりもAPIがコンパクトであり、加えて他の言語のPromiseに多く用いられる``then``メソッドも提供しています。

``Mono``を作る方法は複数ありますが、コールバックを用いた古典的な非同期処理は``Mono.create``でラップするのが簡単です。

```java
class SampleService{

  private final Timer timer = new Timer();

  // タイマーを用いて1秒後に入力値+1を計算して出力
  Mono<Integer> increment(int v){

    return Mono.create(e -> timer.schedule(new TimerTask() {
      @Override
      public void run() { e.complete(v + 1); }
    }, 1000));
  }
}
```

``Mono.create``は[``MonoEmitter``](https://projectreactor.io/core/docs/api/reactor/core/publisher/MonoEmitter.html)を引数とした関数から``Mono``を作成します。
コールバックの中で``MonoEmitter``の[``complete(v)``](https://projectreactor.io/core/docs/api/reactor/core/publisher/MonoEmitter.html#complete-T-)を呼び出すことで値を下流のSubscriberに伝達します。

``MonoEmitter``はキャンセル処理登録の``setCancellation(c)``や例外捕捉の``fail(e)``といったメソッドも持ちます。

```java
// より丁寧に実装したincrementメソッドの例
  Mono<Integer> increment(int v) {

    return Mono.create(e -> {
      
      TimerTask t = new TimerTask() {
        @Override
        public void run() { e.complete(v + 1);}
      };
      
      e.setCancellation(t::cancel); // 下流でキャンセルされたとき
      
      try {
        timer.schedule(t, 1000);
      }catch (Exception ex){
        e.fail(ex); // タイマー開始時の例外を下流にerrorとして転送
      }
    });
  }
```



このように定義された非同期処理は``subscribe``されて初めて実行されます。

```java
SampleService service = new SampleService();

service.increment(0).subscribe(System.out::println);

System.out.println("Subscribed"); // 即座に"Subscribed"、1秒後に"1"を表示
```

現在のスレッドをブロックして同期的に値を得たいときは代わりに``block``を使います。

```java
int result = service.increment(0).block();
System.out.println(result); // 1秒後に"1"を表示
```

また``Mono``を``then``で連鎖することで非同期処理を逐次実行することが出来ます。

```java
service.increment(0)
  .then(service::increment)
  .then(service::increment)
  .then(service::increment)
  .map(v -> "Result: " + v)
  .subscribe(System.out::println);

System.out.println("Subscribed"); // 即座に"Subscribed"、4秒後に"Result: 4"と表示
```

「``Mono``の``then``」 = 「``Mono``を返す関数を``flatMap``」と殆ど等価だと思うのですが、JavaScript等のPromiseを読み慣れている場合はより読みやすいかと思います。

他方で``Mono``を返さない同期処理を繋ぎたいときには``then``ではなく``map``を使わなければならないのは現状イマイチなところです。

複数の``Mono``の同期を取る関数も用意されています。

```java
Mono.when(                           // 全てのMonoの完了を持つ
  service.increment(0),
  service.increment(0).then(service::increment),
  service.increment(0).then(service::increment).then(service::increment)
).subscribe(System.out::println);    // 3秒後に"1, 2, 3"を表示
```

RxJavaの``Observable``をPromise代わりに用いる場合、例えば単一の値を返す``Observable``をメソッドの返値として用いる運用を行うかと思います。
この場合、メソッドのシグネチャからだけでは返値が一つの値なのかストリームなのか判別が出来ません。

```java
interface AsyncService {
  Observable<Integer> action();
  Observable<Integer> stream();
}
```

この点、Reactorでは``Mono``と``Flux``を使い分けることで区別出来ます。また後述の通り使い分けても相互運用はスムーズです。

```java
interface AsyncService {
  Mono<Integer> action();
  Flux<Integer> stream();
}
```

## ``Mono``は``Flux``との相互運用が容易

RxJavaにも非同期に一個の値を出力するクラスとして``Single``が定義されています。``Single``のAPIは``Observable``風ではありますがそれ自体は``Observable``では無く、インターフェイスも共有していません。
そのため、例えば``Single``を返す非同期処理を``Observable``に適用するには一旦``Single``を``Observable``に変換する手順が必要になります。

```java
interface RxService {
  Single<String> action(int v);
  Observable<Integer> stream();
}

RxService rxService = ...;
rxService.stream().flatMap(v -> rxService.action(v).toObservable());
```

他方で``Flux``と``Mono``は共にReactive Streamの``Publisher``であり、そのAPIも多くは共通のインターフェイスである``Publisher``を引数に用いる事が出来ます。
ですので一方を他方に直接適用することが可能です。

```java
interface RoService{
  Mono<String> action(int v);
  Flux<Integer> stream();
}

RoService roService = ...;
roService.stream().flatMap(roService::action);
```

単発の値を返す非同期処理には``Mono``、ストリームには``Flux``と使い分けて、前者を後者に``flatMap``するという書き方が多くなるかと思います。

## ``Mono``は四文字

微妙に重要(個人的には)。

実際のところJDK8にも従来の``Future``の拡張として``Promise``風に使える``CompletableFuture``が導入されています。しかし17文字です。

例として、非同期にレスポンスを返すサービスのインターフェイスを定義してみましょう。

```java
interface AsyncService{
  CompletableFuture<Integer> action1();
  CompletableFuture<Integer> action2();
  CompletableFuture<Integer> action3();
  CompletableFuture<Integer> action4();
  Observable<Integer> stream();
}
```

返値のシグネチャを見ているだけで目が微妙にチカチカと・・・``Mono``だとこうなります。

```java
interface AsyncService{
  Mono<Integer> action1();
  Mono<Integer> action2();
  Mono<Integer> action3();
  Mono<Integer> action4();
  Flux<Integer> stream();
}
```

実際、非同期メッセージングを用いるコードを書き始めると非同期レスポンスの型は頻繁に記述することになります。その度に``CompletableFuture``と書くのと(命名者出てこいっ)``Mono``4文字で済むのではコードの可読性にも差が出ると思います。

また上記の例ではストリームの返値として``Observable``と``Flux``も比べてみました。``Flux``も4文字で、しかも``Mono``と文字数が揃っているのが個人的には好ましいです。

# ``Flux``は過不足無し

・・・と思います。RxJavaの``Observable``を使い慣れていれば``Flux``も難なく使えるはずです。

Reactorが提供するのはRxのサブセットですが必要なAPIは過不足無くカバーしているとの[開発サイドの主張](https://spring.io/blog/2016/02/16/reactor-2-5-a-second-generation-reactive-foundation-for-the-jvm)です。

> Under its new scope and close ties with Reactive Streams Commons, Reactor Core offers just enough Rx coverage to build reactive apps or libraries alike, e.g. [Spring Reactive Web support](https://github.com/spring-projects/spring-reactive).

ただし``Observable``以上のメリットが現在あるかというとやや微妙です。

最大の違いは``Flux``はReactive Streamsの``Publisher``を実装しているという点ですが、現状Reactive Streams準拠のライブラリやクライアント等が少ない以上、Reactive Streamsとの互換性はメリットとしては実質あまり大きくないと思います。

現時点での``Flux``のメリットは専ら前述の``Mono``とインターフェイスを共有しているという設計の筋の良さ、それに伴う相互運用の良さになるかと思います。それと文字数。

というわけで、Androidアプリ開発への適用としては既に実績も支援ライブラリも豊富なRxJavaから移行する積極的な理由は少ないと思います(現状ではAndroid対応にはJava Time API周りの問題があるようです [issue](https://github.com/reactor/reactor-core/issues/98))。

ただ今回はサーバーサイドでSpringと共に使う事を目的として調査したのですが、他のSpringライブラリの対応が増えるようだと今後楽しみだと思います。