OutputRangeの利用と応用（OutputRangeからReactive Extensionsへ至る道）

D言語Advent Calendar 2016の15日目の記事です。

今回はOutputRangeを実際どうやって使っていくかという話をまとめます。

結構細かい話になるので、その１を読んでいただけると理解しやすいと思います。

あと最終的に無理やり解釈でOutputRangeからReactive Extensionsに持って行く話をしたいと思いますので、ちょっと長いですがお付き合いいただければと思います。

OutputRangeの使われどころ

ではまず標準ライブラリであるPhobosで、OutputRangeどのように使われているのか確認していきます。
OutputRangeの使われどころは大きく2か所です。

1. オブジェクトを文字列やバイト列に変換する処理
   各種toStringやstd.base64, std.outbufferなど
2. OutputRangeを拡張するコンセプト
   std.digest.*

それぞれ簡単にですが、どんな意図でどうやって使っているのか掘り下げていきます。
※AppenderはOutputRangeそのものの話になって長くなりすぎるのでちょっと省きます

オブジェクトを文字列やバイト列に変換する処理

比較的素直なstd.base64から簡単な例を挙げてみたいと思います。

std.base64とは、文字通りBase64のエンコードやデコードを行うモジュールです。

このモジュールには、そのままズバリBase64.encodeというバイト列を文字列にする関数があるのですが、ここで文字列の格納先としてcharのOutputRangeが指定できるようになっています。（decodeもubyteのOutputRangeが指定できます）

言ってしまえば、作る文字列の長さが事前に求めづらいため、Appender!stringのようなOutputRangeによろしく入れさせてもらおう、というのが動機であると考えられます。
JavaやC#で言うStringBuilderと全く同じ用途ですね。decode側でも使えるので、OutputRangeはもう少し汎用的な概念になります。

とりあえずサンプルを動かしてみましょう。

Base64.encodeのサンプル

import std.stdio;
import std.base64;
import std.array;

void main()
{
    ubyte[] data = [0, 1, 2, 3];
    auto buf = appender!string;
    Base64.encode(data, buf); //第2引数はOutputRangeを受け付ける
    write(buf.data); //AAECAw==
}

はい。書いてある通りですね。

しかし、わずか4行のとても単純なこの処理も、一度OutputRangeを細かく学んでいると少し冗長ではないでしょうか？
最終的に標準出力に出すだけなのに結果を一度メモリに保持している…
標準出力もOutputRangeなのに…！

ということで少し処理をハックしてみます。以下の通りです。

Base64.encodeの結果を直接標準出力に流す

import std.stdio;
import std.base64;

void main()
{
    ubyte[] data = [0, 1, 2, 3];
    Base64.encode(data, &write!char); //AAECAw==
}

はい、すっきりしました。writeをcharで特殊化することでOutputRangeとして扱えるんでしたね。
やはりOutputRangeにしておくとすごい便利ですね！（IOの回数から目をそむける）

OutputRangeを受け付ける関数の書き方

ここで触れたかったのは、OutputRangeを受け付ける関数の書き方です。
思った要素が渡せるOutputRangeを受け付ける（多分テンプレートだろう）関数はどう書くか？ということです。

encode関数などの実装を見ると分かりますが、OutputRangeを受け取る関数は、テンプレートの型をif条件で絞り込める「Template Constraints」を使って以下のように書くことが多いです。

何か文字列をOutputRangeに入れる関数

void putMessage(R)(auto ref R range) //テンプレートで受け取る
if (isOutputRange!(R, char)) //putを使うところの要素型に合わせる
{
    put(range, "Hello");
}

OutputRange!intなどのインターフェースで受け付けることも可能ですが、テンプレートを使えば構造体でも関数でも好きなものが渡せるのでお手軽スマートだーという感じです。

なお、isOutputRangeでチェックする要素型については、最適化を考えればputするところの型に合わせるのがベストかと思いますが、インターフェースは一度決めたら変えるなとか、まぁ色々あるので最終的な要素型にしておくのが無難なようです。（理解不足）

OutputRangeを拡張するコンセプト

OutputRange自体はかなり汎用的なコンセプトなので、もう少し具体的な用途に落とし込んで使うこともあります。標準ライブラリの中でもstd.digestがそういったことをしています。

std.digestは、バイト列に対して色々な関数を使ったハッシュ値を計算するモジュール群です。サブモジュールとして、MD5とかCRC32とかSHA各種などの実装が含まれます。
細かくは説明しませんが、ちょっと珍しいところだとmurmurhashなんかも標準で提供されています。

そしてstd.digestでは「バイト列からハッシュ値を生成する関数」として、OutputRangeから派生したDigestというコンセプトを提供します。
ハッシュ関数は「可変長のubyteをどんどん入れていってハッシュ値を計算する関数」なので、ubyteのOutputRangeを拡張して作れそうだ、というわけです。

そしてOutputRangeのときと同様、isDigestという判別用のテンプレートがあるので抜粋してきます。以下のようになっています。

isDigestの定義

template isDigest(T)
{
    import std.range : isOutputRange;
    enum bool isDigest = isOutputRange!(T, const(ubyte)[]) && isOutputRange!(T, ubyte) &&
        is(T == struct) &&
        is(typeof(
        {
            T dig = void; //Can define
            dig.put(cast(ubyte)0, cast(ubyte)0); //varags
            dig.start(); //has start
            auto value = dig.finish(); //has finish
        }));
}

これをざっくり読み下すと、以下の特徴を要求するコンセプトになっています。

• 変数として宣言できる
• ubyteとconst(ubyte)[]を受け付けるOutputRangeである
• 構造体である
• 2つのubyteを受け取るput関数を実装している
• startメソッド、戻り値のあるfinishメソッドを実装する

実際これを愚直に実装してもある程度は動くのですが、一部基本編でご紹介したputのメンバー呼び出しをしているようで、最低限のputで何とかしようと思うと動作しないケースが見つかりました。
OutputRange警察としては何とか直してほしいところです。

Digestの上手な書き方

ではPhobosの単体テストや実際のDigestはどうなってるの？ということで見てみると、大変効率的な書き方をされています。
どうやらDigestはこうやって書くんだぞ！という話のようなので単体テストから抜粋してご紹介しておきます。以下の通りです。

struct ExampleDigest
{
public:
    @trusted void put(scope const(ubyte)[] data...) { }

    @trusted void start() { }

    @trusted ubyte[16] finish()
    {
        return (ubyte[16]).init;
    }
}

見た通りですが、putが可変長引数でconst(ubyte)を受け取るようになっているので、ubyteが1つ、ubyteが2つ、ubyteの配列、といった呼び出しを1つのputメソッドで実装してしまっているわけです。かしこい。

そしてこれが普通だとすると、Issue内容のコーナーケースすぎる感がすごい…
とはいえ、型がOKなのに内部でコンパイルエラーが起きるのは実装が悪いはず。
OutputRangeとしての誇りを大事にしてほしい。

と、使う人は良く使うハッシュ関数ですが、標準ライブラリとしてOutputRangeを応用している、ということでちょっと身近に感じてもらえれば良いかなと思います。

OutputRangeを拡張するときの書き方

実際OutputRange＋αを判定するテンプレートはisDigestと同じように以下の形式でテンプレートを用意すれば良いことが分かります。

OutputRange+αを判定するテンプレート

template isHoge(R, E) {
    enum isHoge = isOutputRange!(R, E) && is(typeof({
        R r = void;
        //チェック内容を書く
    }));
}

さて、これでOutputRangeを拡張して使う準備ができました。

Reactive Extensionsへ至る道（自作ライブラリ紹介）

ここまでで調べた「OutputRangeを受け付ける関数」と「OutputRangeを拡張するコンセプト」という2つの知識が揃うとD言語ネイティブっぽいReactive Extensionsが作れます。（本当か？）

そして、これらを踏まえてrxというライブラリを書いているので簡単に紹介します。

• GitHubリポジトリ: https://github.com/lempiji/rx
• dubパッケージ：http://code.dlang.org/packages/rx

なにをするライブラリか？

用途としては、非同期処理やUI関連のイベントを扱いやすくするためのライブラリで、形式的にはReactiveXやReactive Extensionsと呼ばれる実装のD言語版です。

Rx.NET, RxJava, RxJSなどなど、各言語向けに似たような実装がたくさん存在していますので知名度自体はそこそこあるはず。たぶん。

• ReactiveX - Languages

※ReactiveXやReactive Extensionsについては、長いので以下総称して「Rx」とします。

一瞬Rxの話

まずRxには、ObserverやObservableといった最も基本的な共通インターフェースがあります。
どの実装も大差はないのですが、Rx.NET(C#)でいくと以下のような感じです。

Rx.NET(C#版)

interface IObserver<T>
{
    void OnNext(T obj);
    void OnCompleted();
    void OnError(Exception e);
}

interface IObservable<T>
{
    IDisposable Subscribe(IObserver<T> observer);
}

このインターフェースのうちObserverは、たくさんのデータが流れてくるいわゆる「ストリーム」に対して、
- 個々の要素がOnNextに渡される
- 一通り終わるとOnCompletedが呼ばれる
- 途中でエラーになるとOnErrorが呼ばれる
といった処理を表すインターフェースです。

この単純なインターフェースですが、C#におけるRangeのような存在であるIEnumerableに対し、（数学的な意味の）双対であるとか何とか言われており、なかなかすごそうなインターフェースだったりします。

ライブラリのアイデアと設計

当時D言語er脳だった私は上記の双対に関する言及を見て、「え？それってInputRangeに対するOutputRange的な？」と考えました。（今思えば当たらずとも遠からず）

これOutputRangeに対応付ければD言語ネイティブな実装が作れそうな気がする！ということでいろいろ試行錯誤し、やっと形になってきたのが本ライブラリ、というわけです。
（元からRx.NETをD言語で使えたらなーとは思っていた）

というわけで、ぼくのかんがえたさいきょうのRxD言語版のRxを目指すこのライブラリでは、ObserverをOutputRangeの拡張として再定義しています。
もうお分かりと思いますが、さきほどのインターフェースを見てOnNextがputだと考えれば、あとはOnCompletedとOnErrorの分をisDigestのときみたいに拡張すれば作れるじゃん、ということです。

ただし後のことを考えて、OnCompletedとOnError相当の部分はInputRangeのhasLengthのようにオプション要素としておきます。これは最終的に「配列も関数も全部OutputRangeで、それはそのままObserverにも変換できる」ということがしたかったからです。これによって実際に使うときに色々と楽ができます（が、実装は多少苦労します）。

試し書き

rxでは、この記事で書いているOutputRangeを受け付ける関数の書き方やOutputRangeを拡張するコンセプトをバリバリ使って書いています。

ちょっとそれにならってObserverやObservableを書いてみると以下のようになります。

Observerな構造体

struct TestObserver
{
    void put(int n) { writeln(n); }
    void completed() { writeln("completed"); }
    void failure(Exception e) { writeln("failure"); }
}

import std.range : isOutputRange;
static assert(isOutputRange!(TestObserver, int)); //少なくともOutputRange

import rx;
static assert(hasCompleted!TestObserver); //completedはオプションなので無くても良い
static assert(hasFailure!TestObserver); //failureもオプションなので無くても良い
static assert(isObserver!(TestObserver, int)); //全部あるかどうかも一応チェックできる

Observableな構造体

struct SomeObservable
{
    alias ElementType = int;

    Disposable subscribe(T)(auto ref T observer)
    if (isOutputRange!(T, ElementType)) //Base64.encodeなどの応用
    {
        return null;
    }
}

import rx;
static assert(isObservable!SomeObservable); //Observableの特性を満たすかチェック

動作サンプル

と、ここまでおおまかにアイデアと設計の話をしましたが、実際に使ったサンプルも貼り付けておきます。

Rxもメソッドチェインが売りのライブラリなので、UFCSを使ったサンプルです。
std.algorithmと同様にOutputRangeの値が色々加工される雰囲気を感じ取ってもらえれば、という感じです。

import rx;
import std.range : put, iota;
import std.conv : to;

unittest
{
    //intを受け取るOutputRangeで、他のOutputRangeを登録すると値を分配する機能を持つ
    auto subject = new SubjectObject!int;

    //std.algorithmのようにUFCSでOutputRangeに入ってきた値を加工するよう定義
    auto evenStrings = subject.filter!"a % 2 == 0".map!(to!string);

    //最終的にsubjectから値を受け取るObserver(ここではただのOutputRange)を用意
    auto buf = appender!string();

    //bufを登録して、実際にsubjectに値を入れると、加工されてappenderに入る
    evenStrings.doSubscribe(buf);
    put(subject, iota(10));

    import std.algorithm : equal;
    assert(equal(buf.data, ["0", "2", "4", "6", "8"])); //偶数のみで文字列化されている
}

動作（非同期）

とはいえ、これではイマイチ何が良いのかわかりづらいので、UIに絡めて実際に非同期の処理も書いてみます。

GitHubのExamplesから主なところだけ抜粋したものですが、比較的メジャーなUIライブラリのDlangUIと連携させて、テキストボックスの値をラベルに反映する例です。

入力中で頻繁に変更がある間はイベントをスルーして、500msec操作がなく落ち着いた頃にラベルに反映、というコードがdebounceという関数を使って以下のように書けます。

extern (C) int UIAppMain(string[] args)
{
    auto window = createAppWindow();

    //非同期処理をメインループで処理させるための補助クラスを登録しておく（定型）
    auto scheduler = new DlangUIScheduler();
    window.mainWidget.addChild(scheduler);
    currentScheduler = scheduler;

    auto label = window.mainWidget.childById!TextWidget("label");
    auto edit = window.mainWidget.childById!EditLine("edit");

    //EditLineの変更イベントを見て、500msec発火しなかったら最後のイベントを処理
    edit.contentChange
        .asObservable()
        .debounce(dur!"msecs"(500))
        .doSubscribe((EditableContent _) {
            label.text = edit.text;
        });

    //以下省略
}

ここまでユースケースにハマる例もそうそうないですが、手で書くことを想像すると非常に楽で良い感じです。
このあたりの時間が絡む処理もObserverとして抽象化することでライブラリにお任せできるのがRxの強みですね。

と、これ以上は長すぎるので、他の色々な例はGitHubのリポジトリを見ていただければと思います！

まとめ

ここまでの内容をまとめると以下の通りです。

• OutputRangeの用途として以下のようなものがある
  • StringBuilderのような可変長のバッファー（toStringやstd.base64）
  • 可変長のデータを受け付けるコンセプトを作るベースにする（std.digest, Rx）
• OutputRangeを受け付ける関数は「Template Constraint」を使って書くと良い感じ
• 非同期処理を書くのにRxはとても便利
• Rxたのしい

長くなりましたが、誤り/マサカリ/勘違い、もろもろのフィードバックお待ちしております！

以上！