Play Framework の Iteratee/Enumerator/Enumeratee の使用例

はじめに

概要

Play Framework の Iteratee/Enumerator/Enumeratee って分かりにくいですね。
公式ドキュメントを読んでもよく分からないし、ググっても数学的な説明が書いてあったりして、余計に分かりにくい。

本記事では、いくつかの例を示しながら理解を深めていこうと思います。
なお、参考記事は、Understanding Play2 Iteratees for Normal Humansで、Play Framework 2.1.0、Scala 2.10 を対象としています。

Iteratee/Enumerator/Enumeratee とは？

Iteratee/Enumerator/Enumeratee を簡単にいうと:

名称	説明
Iteratee[E, A]	ループの中身。型 E から型 A を生成する。
Enumerator[E]	コレクションを汎化したもので 型 E を列挙する。無限に列挙(Streaming)することもできる。
Enumeratee[E, A]	滅多に使わないので今は考えなくていい。

Iterator と Iteratee の違い

Scala には、Iterator[E] というおなじみのインターフェースがあります。
Iterator[E] と Iteratee[E, A] の違いは、Iterator[E] は、何らかのコレクションから生成されるのに対して、Iteratee[E, A] は、コレクションとは独立していて、不変(immutable)で、非ブロック非同期処理で、入力の型 E と出力の型 A が静的に定義されているという特長に加え、結果が必要になるまで何ら実行されないという特長があります。

とはいうものの、Iteratee で実現できることのほとんどは、Iterator を使用しても実現できます。
どちらを使うかは、好みの問題です。

例1: Int型の要素をもつ Enumerator[Int] とそれの合計を計算する Iteratee[Int, Int]

// 合計を計算する Iteratee (初期値は 0)
val initialValue: Int = 0
val sumIteratee: Iteratee[Int, Int] = Iteratee.fold(initialValue) { (total, e) => total + e }
// 1, 234, 455, 987 を要素にもつ Enumerator
val intEnumerator1: Enumerator[Int] = Enumerator(1, 234, 455, 987)

// intEnumerator1 の合計を計算する
val futureTotal1: Future[Int] = intEnumerator1.run(sumIteratee)
val total1: Int = Await.result(futureTotal1, Duration.Inf)
println("total=" + total1.toString)

// 431, 57, 668 を要素にもつ Enumerator
val intEnumerator2: Enumerator[Int] = Enumerator(431, 57, 668)

// intEnumerator2 の合計を計算する
val futureTotal2: Future[Int] = intEnumerator2.run(sumIteratee)
val total2: Int = Await.result(futureTotal2, Duration.Inf)
println("total=" + total2.toString)

次のような結果が出力されます。

total=1677
total=1156

Iteratee[E,A] は、コレクションと独立しているため、intEnumerator1 と intEnumerator2 というそれぞれ異なるコレクションに対して実行でき、Iteratee[E,A] は、非ブロック非同期で処理を実行するため、Future[Int] が得られます。

Future[Int] から結果を取り出すため、わざわざ Await.result() メソッドを使用しなければならないため、少しめんどくさいと感じるでしょう。
私もそう思います。

例2: 無限 の Enumerator (Streaming Enumerator)

Enumerator は無限に列挙することができます。

// 500 ミリ秒ごとに文字列 "current time %s" を生成し続ける Enumerator
val stringGenerator: Enumerator[String] = Enumerator.generateM(
  play.api.libs.concurrent.Promise.timeout(
    Some("current time %s".format((new java.util.Date()))),
    500
  )
)

// コンソールに出力する Iteratee
val printIteratee: Iteratee[String, Unit] = Iteratee.foreach { println _ }

// 5 秒間だけ、stringGenerator が生成する要素それぞれに対して printIteratee を実行する
val future: Future[Unit] = stringGenerator.run(printIteratee)
Await.result(future, Duration(5, "seconds")

次のような結果が出力されます。

current time Tue Feb 26 11:55:58 JST 2013
current time Tue Feb 26 11:55:58 JST 2013
current time Tue Feb 26 11:55:59 JST 2013
current time Tue Feb 26 11:55:59 JST 2013
current time Tue Feb 26 11:56:00 JST 2013
current time Tue Feb 26 11:56:00 JST 2013
current time Tue Feb 26 11:56:01 JST 2013
current time Tue Feb 26 11:56:01 JST 2013
current time Tue Feb 26 11:56:02 JST 2013
current time Tue Feb 26 11:56:02 JST 2013

例3: Iteratee[E,A] と Future[Iteratee[E,A]]

Iteratee[E,A] から Future[Iteratee[E,A]] へ変換し、逆に Future[Iteratee[E,A]] から Iteratee[E,A] に変換することができます。
両者は可換です。

val initialValue: Int = 0
val sumIterator: Iteratee[Int, Int] = Iteratee.fold(initialValue) { (total, e) => total + e }
val futureSumIterator: Future[Iteratee[Int, Int]] = sumIterator.unflatten.map(_.it)
val sumIteratorAgain: Iteratee[Int, Int] = Iteratee.flatten(futureSumIterator)

例4: Iteratee[E, A].feed と Input

Iteratee[E, A].feed() メソッドを使うと、Enumerator[E] を使わずに、要素を 1 つずつ Iteratee[E, A] に与えることができます。
Iteratee[E, A].feed() メソッドには、要素を汎化した Input クラスのサブクラスである El, Empty, EOF のいずれかを与えます。
次に、Input の 3 つのサブクラス El, Empty, EOF の概要を示します。

名称	説明
Input.El[E]	型 E の要素を表します。
Input.Empty	空の要素を表します。
Input.EOF	イテレーションを終了します。

次の例は、Enumerator を使わずに、Iteratee[E, A].feed() メソッドを用いて、要素列 1, 234, 455, 987 を先頭から順番に 1 つずつ与えます。

val initialValue: Int = 0
var sumIteratee: Iteratee[Int, Int] = Iteratee.fold(initialValue) { (total, e) => total + e }
var futureSumIterator: Future[Iteratee[Int, Int]] = null

// 1 を Iteratee に与える
futureSumIterator = sumIteratee.feed(Input.El(1))
sumIteratee = Iteratee.flatten(futureSumIterator)

// 234 を Iteratee に与える
futureSumIterator = sumIteratee.feed(Input.El(234))
sumIteratee = Iteratee.flatten(futureSumIterator)

// 455 を Iteratee に与える
futureSumIterator = sumIteratee.feed(Input.El(455))
sumIteratee = Iteratee.flatten(futureSumIterator)

// 987 を Iteratee に与える
futureSumIterator = sumIteratee.feed(Input.El(987))
sumIteratee = Iteratee.flatten(futureSumIterator)

// 合計を計算する
val futureTotal: Future[Int] = sumIteratee.run
val total: Int = Await.result(futureTotal, Duration.Inf)
println("0+1+234+455+987=" + total.toString)

次のような結果が出力されます。

0+1+234+455+987=1677

もし、あなたが Future 好きなら

Future[Iteratee[E, A]] を Iteratee[E, A] にわざわざ変換しなくても、Future[Iteratee[E, A]] のまま同等の処理を実行することができます。

val initialValue: Int = 0
var sumIteratee: Iteratee[Int, Int] = Iteratee.fold(initialValue) { (total, e) => total + e }
var futureSumIteratee: Future[Iteratee[Int, Int]] = null

// 1 を Iteratee に与える
futureSumIteratee = sumIteratee.feed(Input.El(1))

// 234 を Iteratee に与える
futureSumIteratee = futureSumIteratee.flatMap(_.feed(Input.El(234)))

// 455 を Iteratee に与える
futureSumIteratee = futureSumIteratee.flatMap(_.feed(Input.El(455)))

// 987 を Iteratee に与える
futureSumIteratee = futureSumIteratee.flatMap(_.feed(Input.El(987)))

// 合計を計算する
val totalFuture: Future[Int] = futureSumIteratee.flatMap(_.run)
val total = Await.result(totalFuture, Duration.Inf)
println("0+1+234+455+987=" + total.toString)

例5: Iteratee[E, A] の不変性 (Immutability)

Iteratee[E, A] はインスタンス化された後、不変で内部状態を変化させません。それを見ていきます。

val initialValue: Int = 0
val sumIteratee0: Iteratee[Int, Int] = Iteratee.fold(initialValue) { (total, e) => total + e }

// sumIteratee0 に 1 を与える
val sumIteratee1: Iteratee[Int, Int] = Iteratee.flatten(sumIteratee0.feed(Input.El(1)))
// sumIteratee1 に 234 を与える
val sumIteratee2: Iteratee[Int, Int] = Iteratee.flatten(sumIteratee1.feed(Input.El(234)))
// sumIteratee2 に 583 を与える
val sumIteratee2_1: Iteratee[Int, Int] = Iteratee.flatten(sumIteratee2.feed(Input.El(583)))
// sumIteratee2 に 162 を与える
val sumIteratee2_2: Iteratee[Int, Int] = Iteratee.flatten(sumIteratee2.feed(Input.El(162)))

println("0=" + Await.result(sumIteratee0.run, Duration.Inf))
println("0+1=" + Await.result(sumIteratee1.run, Duration.Inf))
println("0+1+234=" + Await.result(sumIteratee2.run, Duration.Inf))
println("0+1+234+583=" + Await.result(sumIteratee2_1.run, Duration.Inf))
println("0+1+234+162=" + Await.result(sumIteratee2_2.run, Duration.Inf))

次のような結果が出力されます。

0=0
0+1=1
0+1+234=235
0+1+234+583=818
0+1+234+162=397

例6: Iteratee[E, A] は結果が必要になるまで何ら実行されない

Iteratee[E, A] は、本当に結果が必要になるまで何も実行されません。

val initialValue: Int = 0
val sumIteratee0: Iteratee[Int, Int] = Iteratee.fold(initialValue) { (total, e) => {
    println("e=" + e.toString)
    total + e
  }
}
// feed しただけでは計算されない
val sumIteratee1 = Iteratee.flatten(sumIteratee0.feed(Input.El(1)))
val sumIteratee2 = Iteratee.flatten(sumIteratee1.feed(Input.El(234)))
val sumIteratee3 = Iteratee.flatten(sumIteratee2.feed(Input.El(455)))
val sumIteratee4 = Iteratee.flatten(sumIteratee3.feed(Input.El(987)))
// まだ計算されない
val futureTotal: Future[Int] = sumIteratee4.run

println("caclulate result")
val total = Await.result(futureTotal, Duration.Inf)
println("0+1+234+455+987=" + total.toString)

次のような結果が出力されます。

caclulate result
e=1
e=234
e=455
e=987
0+1+234+455+987=1677

例7: Input.Empty と Input.EOF

Input.Empty を Iteratee[E, A] に与えると、単に無視されます。
Input.EOF を Iteratee[E, A] に与えると、イテレーションを停止し、何を与えても結果は同じになります。

val initialValue: Int = 0
val sumIteratee0: Iteratee[Int, Int] = Iteratee.fold(initialValue) { (total, e) => total + e }

val sumIteratee1 = Iteratee.flatten(sumIteratee0.feed(Input.El(1)))
val sumIteratee2 = Iteratee.flatten(sumIteratee1.feed(Input.Empty))
val sumIteratee3 = Iteratee.flatten(sumIteratee2.feed(Input.El(234)))
val sumIteratee4 = Iteratee.flatten(sumIteratee3.feed(Input.EOF))
val sumIteratee5 = Iteratee.flatten(sumIteratee4.feed(Input.El(455)))
val sumIteratee6 = Iteratee.flatten(sumIteratee5.feed(Input.El(987)))
// Input.Empty を feed した後に feed した値は計算される
println("0+1+234=" + Await.result(sumIteratee3.run, Duration.Inf))
// Input.EOF を feed して以降に feed した値は計算されない
println("0+1+234+455=" + Await.result(sumIteratee5.run, Duration.Inf))
println("0+1+234+455+987=" + Await.result(sumIteratee6.run, Duration.Inf))

このプログラムを実行すると、次のような結果が出力されます。

0+1+234=235
0+1+234+455=235
0+1+234+455+987=235

結果から分かるように、Input.EOF を Iteratee[E, A] に与えるとイテレーションが停止するため、Input.EOF を与えた後に Input.El(455) や Input.El(987) は与えても合計は変化していません。また、Input.EOF をフィードすると、イテレーションを停止することが確認できます。