Sodium で関数型リアクティブプログラミング2 - skip・take 処理

前回に続き、Sodium を試してみます。

• Sodium

今回は 「RxJava で行単位のファイル処理 - Groovy, Java Lambda」 で実装したものと同等の処理を Sodium を使って実装してみました。

ソースは http://github.com/fits/try_samples/tree/master/blog/20141209/

skip・take 処理

Sodium には、RxJava の際に使った skip （指定した処理数だけ無視する） や take （指定した処理数だけ取り出す） のようなメソッドが用意されていないようなので、Event と Behavior を組み合わせて実装してみる事にします。

Event クラスには gate メソッドがあり、このメソッドを使えば Behavior オブジェクトの値が true の場合だけイベントを発生するような Event オブジェクトを作成できます。

つまり、指定した数のイベントを受け取れば true と false が反転する Behavior オブジェクトを用意して gate メソッドへ与えてやれば skip や take の処理を実現した Event オブジェクトを作成できそうです。

なお、skip と take の違いは、false から true への変化か true から false への変化かの違いしかありませんので、下記のサンプルでは batch メソッドとして共通化し、判定処理を引数 Lambda1<Integer, Boolean> cond として渡すようにしました。

batch メソッドは以下のような処理内容となっています。

• (1) イベントの発生数をカウントアップする Behavior を用意 （実際は BehaviorSink を使用）
• (2) イベント発生時に (1) をカウントアップ
• (3) (1) のカウント値が条件に合致した場合のみイベントを発生させる Event を作成

ReadLineFile.java

import sodium.*;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.function.*;

class ReadLineFile {
    public static void main(String... args) throws Exception {
        // skip 処理
        Function<Integer, Function<Event<String>, Event<String>>> skip = n -> {
            return ev -> batch( v -> v >= n, n, ev);
        };
        // take 処理
        Function<Integer, Function<Event<String>, Event<String>>> take = n -> {
            return ev -> batch( v -> v < n, n, ev);
        };

        // 1行スキップして 3行取得する Event を作成する処理を合成 (b) (c)
        Function<Event<String>, Event<String>> skipAndTake3 = skip.apply(1).andThen( take.apply(3) );

        // (a)
        EventSink<String> es = new EventSink<>();
        // (d)
        Listener esl = skipAndTake3.apply(es).map( v -> "# " + v ).listen( System.out::println );
        // ファイルを行単位で処理
        readFileLines(args[0], es);

        esl.unlisten();
    }

    private static void readFileLines(String fileName, EventSink<String> es) throws IOException {
        // ファイルを行単位で EventSink へ send
        try (BufferedReader br = new BufferedReader(new FileReader(fileName))) {
            br.lines().forEach( es::send );
        }
    }
    // skip・take の共通処理
    private static Event<String> batch(Lambda1<Integer, Boolean> cond, int n, Event<String> ev) {
        // (1) イベント発生数をカウントする Behavior を用意
        BehaviorSink<Integer> counter = new BehaviorSink<>(0);
        // (2) イベント発生時に counter をカウントアップ
        ev.listen( v -> counter.send(counter.sample() + 1) );
        // (3) counter の値が条件に合致した場合のみイベント発生する Event を作成
        return ev.gate(counter.map(cond));
    }
}

上記は、以下の 4つの Event が (a) -> (b) -> (c) -> (d) の順でイベントを伝播するような構成となっており、(d) に到達した場合のみ System.out::println を実行するようになっています。

• (a) EventSink
• (b) skip で作成した Event
• (c) take で作成した Event
• (d) 先頭に "# " を付与する Event

また、batch メソッドの実装内容に関しては counter の無駄なカウントアップを防止するため、下記のように unlisten するようにした方が望ましいかもしれません。

unlisten する処理を追加した batch メソッドの例

    private static Event<String> batch(Lambda1<Integer, Boolean> cond, int n, Event<String> ev) {
        BehaviorSink<Integer> counter = new BehaviorSink<>(0);
        final ArrayList<Listener> list = new ArrayList<>(1);

        list.add(ev.listen( v -> {
            int newValue = counter.sample() + 1;
            counter.send(newValue);

            if (newValue >= n) {
                list.stream().forEach( li -> li.unlisten() );
            }
        }));

        return ev.gate(counter.map(cond));
    }

実行

それでは、下記ファイルを使って実行してみます。

test1.txt

1a
2b
3c
4d
5e

実行結果は以下の通りです。 1行スキップした後、3行を先頭に "# " を付与して出力しています。

実行結果

> java -cp .;sodium.jar ReadLineFile test1.txt

# 2b
# 3c
# 4d

test1.txt を処理した際の (a) ～ (d) の Event に対するイベント伝播状況をまとめると以下のようになっていると考えられます。（○ は伝播する、× は伝播しない）

対象行	値	(a)	(b)	(c)	(d)
1行目	1a	○	○	×	×
2行目	2b	○	○	○	○
3行目	3c	○	○	○	○
4行目	4d	○	○	○	○
5行目	5e	○	○	○	×

(d) まで到達した 2 ～ 4行目だけを出力する結果となります。

Sodium で関数型リアクティブプログラミング

関数型リアクティブプログラミング（FRP）用のライブラリ Sodium を試してみました。

• Sodium

Sodium には現時点で Java・Haskell・C++・C# 用のライブラリが用意されていますが（Embedded-C や Rust 用のライブラリも実装中の模様）、今回は Java 用のモジュールを使います。

今回のソースは http://github.com/fits/try_samples/tree/master/blog/20141123/

はじめに

Sodium の Java 用モジュールは Maven のセントラルリポジトリ等で配布されていないようなので、ソースを取得してビルドする事にします。

ビルドには Java 8 と Apache Ant を使います。（build.xml の source・target が 1.8 となっています）

ビルド例

$ git clone https://github.com/SodiumFRP/sodium.git
・・・
$ cd sodium/java
$ ant

ビルドに成功すると sodium/sodium.jar ファイルが生成されます。

Event と Behavior

Sodium では下記のような Event と Behavior を組み合わせて処理を組み立てます。

クラス	特徴	現在値の取得（sample メソッド）	イベント受信（listen メソッド）
Event	離散的なストリームを扱う	×	○
Behavior	連続的なストリームを扱う	○	×

Event の処理内容

まずは Event を単純に listen するだけの処理を実装してみます。

Event に何らかの値を送信する（イベントを発火させる）には Event のサブクラスである EventSink の send メソッドを使います。

なお、今回のようなサンプルでは Listener を unlisten する必要は無いのですが、一応入れています。

EventSample.java

import sodium.*;

class EventSample {
    public static void main(String... args) {

        EventSink<String> es = new EventSink<>();
        Listener esl = es.listen(System.out::println);

        es.send("ES1");

        System.out.println("---");

        es.send("ES2");

        esl.unlisten();
    }
}

実行結果は下記の通りです。

EventSink へ send した値 （ES1 と ES2） が listen の処理 （System.out::println） へ渡されています。

ビルドと実行

> javac -cp sodium.jar EventSample.java

> java -cp .;sodium.jar EventSample

ES1
---
ES2

Behavior の処理内容

次は、Behavior のカレント値の変更を listen してみます。

Behavior を直接 listen する事はできませんが、updates や value メソッドを使えば Behavior の値の変更に対応した Event を取得できます。

updates と value の違いは、取得した Event が listen 時にカレント値を含むかどうかの違いです。

動作としては RxJava の PublishSubject と BehaviorSubject にそれぞれ該当すると思います。

Behavior の Event 取得メソッド	listen 時のカレント値の扱い	RxJava の類似クラス
updates	含まない	rx.subjects.PublishSubject
value	含む	rx.subjects.BehaviorSubject

Event と同様に Behavior のカレント値を変更するには BehaviorSink の send メソッドを使います。

BehaviorSample.java

import sodium.*;

class BehaviorSample {
    public static void main(String... args) {
        updatesSample();

        System.out.println("");

        valueSample();
    }
    // updates メソッドのサンプル
    private static void updatesSample() {
        System.out.println("*** Behavior.updates sample ***");

        BehaviorSink<String> bh = new BehaviorSink<>("BH1");
        Listener bhl = bh.updates().listen( msg -> System.out.println("behavior: " + msg) );

        bh.send("BH2");

        System.out.println("---");

        bh.send("BH3");

        bhl.unlisten();
    }
    // value メソッドのサンプル
    private static void valueSample() {
        System.out.println("*** Behavior.value sample ***");

        BehaviorSink<String> bh = new BehaviorSink<>("BH1");
        Listener bhl = bh.value().listen( msg -> System.out.println("behavior: " + msg) );

        bh.send("BH2");

        System.out.println("---");

        bh.send("BH3");

        bhl.unlisten();
    }
}

value メソッドの場合のみ、初期値として設定した値 （BH1） を出力しています。

実行結果

> java -cp .;sodium.jar BehaviorSample

*** Behavior.updates sample ***
behavior: BH2
---
behavior: BH3

*** Behavior.value sample ***
behavior: BH1
behavior: BH2
---
behavior: BH3

Event の各種メソッド

最後に Event クラスの map・merge・hold・snapshot メソッドを簡単に試してみます。

map

map メソッドによって元の Event で発火した値を加工した値を発火する Event を作成できます。

import sodium.*;

class EventMethodSample {
    public static void main(String... args) {
        mapSample();
        ・・・
    }

    private static void mapSample() {
        System.out.println("*** Event.map sample ***");

        EventSink<String> es = new EventSink<>();
        Listener esl = es.listen( msg -> System.out.println("event sink: " + msg) );

        // 元の値に !!! を付ける Event 作成
        Event<String> me = es.map( msg -> msg + "!!!" );
        Listener mel = me.listen( msg -> System.out.println("mapped event: " + msg) );

        es.send("ME1");
        es.send("ME2");

        mel.unlisten();
        esl.unlisten();
    }
    ・・・
}

ちなみに、上記では使っていませんが、Listener は append する事が可能です。 （append で単一の Listener へまとめれば unlisten を個々に実施しなくても済みます）

実行結果

> java -cp .;sodium.jar EventMethodSample

*** Event.map sample ***
event sink: ME1
mapped event: ME1!!!
event sink: ME2
mapped event: ME2!!!
・・・

merge

merge メソッドによって二つの Event をマージできます。下記ではどちらの Event が発火しても発火する Event を作成しています。

import sodium.*;

class EventMethodSample {
    public static void main(String... args) {
        ・・・
        mergeSample();
        ・・・
    }
    ・・・
    private static void mergeSample() {
        System.out.println("*** Event.merge sample ***");

        EventSink<String> es1 = new EventSink<>();
        Listener es1l = es1.listen( msg -> System.out.println("event sink1: " + msg) );

        EventSink<String> es2 = new EventSink<>();
        Listener es2l = es2.listen( msg -> System.out.println("event sink2: " + msg) );

        Event<String> me = es1.merge(es2);
        Listener mel = me.listen( msg -> System.out.println("merged event: " + msg) );

        es1.send("ES1-1");

        System.out.println("---");

        es2.send("ES2-1");

        System.out.println("---");

        es1.send("ES1-2");

        mel.unlisten();
        es2l.unlisten();
        es1l.unlisten();
    }
    ・・・
}

実行結果

> java -cp .;sodium.jar EventMethodSample
・・・

*** Event.merge sample ***
event sink1: ES1-1
merged event: ES1-1
---
event sink2: ES2-1
merged event: ES2-1
---
event sink1: ES1-2
merged event: ES1-2

・・・

hold

hold メソッドによって Event の発火した値でカレント値が変化する Behavior を作成できます。

import sodium.*;

class EventMethodSample {
    public static void main(String... args) {
        ・・・
        holdSample();
        ・・・
    }
    ・・・
    private static void holdSample() {
        System.out.println("*** Event.hold sample ***");

        EventSink<String> es = new EventSink<>();
        Listener esl = es.listen( msg -> System.out.println("event sink: " + msg) );

        Behavior<String> bh = es.hold("BH1");
        Listener bhl = bh.value().listen( msg -> System.out.println("behavior: " + msg) );

        es.send("ES1");

        System.out.println("bh current value: " + bh.sample());

        System.out.println("---");

        es.send("ES2");

        System.out.println("bh current value: " + bh.sample());

        esl.unlisten();
        bhl.unlisten();
    }
    ・・・
}

bh の初期値は BH1 ですが、send した値 （ES1 や ES2） によって sample メソッドの結果が変化しています。

実行結果

> java -cp .;sodium.jar EventMethodSample
・・・

*** Event.hold sample ***
behavior: BH1
event sink: ES1
behavior: ES1
bh current value: ES1
---
event sink: ES2
behavior: ES2
bh current value: ES2

・・・

snapshot

snapshot によって Event 発火時に任意の Behavior のカレント値を発火する Event を作成できます。

import sodium.*;

class EventMethodSample {
    public static void main(String... args) {
        ・・・
        snapshotSample();
    }
    ・・・
    private static void snapshotSample() {
        System.out.println("*** Event.snapshot sample ***");

        EventSink<String> es = new EventSink<>();
        Listener esl = es.listen( msg -> System.out.println("event sink: " + msg) );

        Behavior<Integer> bh = new Behavior<>(1);
        Listener bhl = bh.value().listen( msg -> System.out.println("behavior: " + msg) );

        Event<Integer> se = es.snapshot(bh);
        Listener sel = se.listen( i -> System.out.println("snapshot event: " + i) );

        es.send("ES1");

        System.out.println("bh current value: " + bh.sample());

        System.out.println("---");

        es.send("ES2");

        System.out.println("bh current value: " + bh.sample());

        sel.unlisten();
        esl.unlisten();
        bhl.unlisten();
    }
}

snapshot で作成した Event （se） は EventSink へ send した値 （ES1 と ES2） に関わらず、bh のカレント値 （1） を発火しています。

実行結果

> java -cp .;sodium.jar EventMethodSample
・・・

*** Event.snapshot sample ***
behavior: 1
event sink: ES1
snapshot event: 1
bh current value: 1
---
event sink: ES2
snapshot event: 1
bh current value: 1

・・・

snapshot した Event で発火するのは Behavior のカレント値であることを確認するため、上記の Behavior を BehaviorSink へ変更し ES2 を send する前にカレント値を 2 へ変更してみました。

import sodium.*;

class EventMethodSample {
    public static void main(String... args) {
        ・・・
        snapshotSample2();
    }
    ・・・
    private static void snapshotSample2() {
        ・・・
        // BehaviorSink へ変更
        BehaviorSink<Integer> bh = new BehaviorSink<>(1);
        Listener bhl = bh.value().listen( msg -> System.out.println("behavior: " + msg) );

        Event<Integer> se = es.snapshot(bh);
        ・・・
        System.out.println("---");

        // bh のカレント値を 2 へ変更
        bh.send(2);
        es.send("ES2");

        System.out.println("bh current value: " + bh.sample());
        ・・・
    }
}

Behavior のカレント値を 2 へ変更した後、snapshot の Event は 2 の値を発火している事を確認できます。

実行結果

> java -cp .;sodium.jar EventMethodSample
・・・

*** Event.snapshot sample2 ***
behavior: 1
event sink: ES1
snapshot event: 1
bh current value: 1
---
behavior: 2
event sink: ES2
snapshot event: 2
bh current value: 2