scalaz.zio.streamのAPI弾丸ツアー

## scalaz.zio.stream
scalaz.zioのストリーム処理用package
`scalaz-zio-streams` に含まれている
ZIOについては[マイクロサイト](https://scalaz.github.io/scalaz-zio/)と他の方が[書いてくださっている記事](https://qiita.com/tags/zio)を読めばだいたい理解できそうな感じだったので
本記事ではStreamに重点を置き、各APIを解説してみる
ZIOについても必要な部分では少し触れる

## 全体
zioのStreamは主に `ZStream` [[source](https://github.com/scalaz/scalaz-zio/blob/master/streams/shared/src/main/scala/scalaz/stream/ZStream.scala)] と `ZSink`[[source](https://github.com/scalaz/scalaz-zio/blob/master/streams/shared/src/main/scala/scalaz/stream/ZSink.scala)] と言うクラスによって構成されている
簡単に説明するならば `ZStream` はPublisherで `ZSink` はConsumerだ
それぞれ `Stream` `Sink` という型第一引数を `Any` にしたエイリアスが提供されている(型第一引数はZIOのDI的機能を利用するときに必要だが、使わない場合は気にしなくていい、本記事では説明しない)

## Streamの実行
基本的には `Stream#run` 関数に `Sink` を渡すことで `ZIO` インスタンスを得ることができる
`ZIO` インスタンスとは `Runtime#unsafeRun` に渡すなどで実行することのできる再利用可能なタスクのようなもの
ZIOアプリケーション( `scalaz.zio.App` を継承したメインクラス)ではアプリケーション全体がひとつの大きなZIOインスタンスとなる(これはHaskellのmainメソッドが大きなIOモナドであることと同じ)

## Publisher
この項ではStreamのファクトリメソッドを紹介していく

- **fromIterable**  

```scala
def fromIterable = {
  Stream.fromIterable(1 to 3).foreach{ i => putStr(s"$i ") } // run task => 1 2 3
}
```
引数の `scala.collection.Iterable` の要素を一つずつ渡すStreamを生成する
すべての要素を渡し終えるとStreamが終了する
`foreach` はStreamを実行可能状態にする一番カンタンな方法(多分)

- **bracket**  

```scala
def bracket(filePath: String) = {
  Stream.bracket(openFile(filePath)) { bf => IO.effectTotal(bf.close()) } { bf => IO.effect(readNext(bf)) }
    .foreach(putStrLn) // output file content for std out
}
```
外部リソースから要素を取得するStreamを作成する
第一引数はリソースの確保、第二引数はリソースの開放処理、第三引数はリソースから要素を取得する処理になる
第三引数の関数の戻り値は `ZIO[R, E, Option[A]]` となり、取得した要素をOptionで渡し、値がNoneの場合はStreamが終了する

- **fromQueue**  

```scala
def fromQueue = {
  for {
    q <- Queue.unbounded[Int]
    f <- Stream.fromQueue(q).take(3).foreach{ i => putStr(s"$i ") } .fork
    _ <- q.offer(1) *> q.offer(2) *> q.offer(3) *> q.offer(4)
    _ <- f.join
  } yield () // run task => 1 2 3
}
```
`ZQueue` より要素を取得する( `ZQueue` はZIOコンテキストで使えるインメモリキュー)
`ZQueue` がシャットダウンした場合にStreamが終了するが、基本的に永続的に流れ続けるStreamの上流として使うことが多いだろう
`ZIO#fork` は対象ZIOタスクを非同期で行うためのもの(Streamの実行はそのままではBlocking処理のため)

- **fromEffect**  

```scala
def fromEffect = {
  val effect = IO(100)
  Stream.fromEffect(effect).foreach{ i => putStr(s"$i ") } // run task => 100
}
```
ZIOインスタンスの結果値より要素を取得する
`ZIO[R, E, List[A]]` を引数に渡しても `List[A]` 型の要素が一つだけ渡されるStreamが生成されることになる
`Stream[E, List[A]]` を `Stream[A]` に変換するのは後述の `Stream#mapConcat` を使うといい

## Consumer
この項ではStreamに流れてくる要素を消費/収集するConsumerであるSinkのファクトリメソッドを紹介する
- **collect**  

```scala
def collect_sink = {
  pub.run(Sink.collect[Int]) // run task => List(1, 2, 3)
}
```
流れてきた要素をすべて収集し、Listとして返すコンシューマ

- **readWhile**  

```scala
def readWhile = {
  pub.run(Sink.readWhile[Int](_ < 3)) // run task => List(1, 2)
}
```
引数の関数がfalseになるまで収集するコンシューマ
`Stream#run` の引数として渡している場合は、falseになった場合Streamが残っていても処理を終了する
(後述するがSinkは `Stream#run` のほか `Stream#transduce` に渡して使用することもある)

- **drain**  

```scala
def drain_sink = {
  pub.tap { i => putStr(s"$i ") }.run(Sink.drain) // run task => List(1, 2)
}
```
流れてくる要素をすべて捨てる
Stream中に行われる副作用のみを必要としているときなどに使う

- **identity**  

```scala
def identity = {
  pub.run(Sink.identity[Int]) // run task => 1
}
```
要素を一つだけ収集するコンビネータ
`Stream#run` の引数として渡している場合は、1要素取得した時点でStreamが終了する

- **foldLeft**  

```scala
def foldLeft = {
  pub.run(Sink.foldLeft[Int, Int, Int](0)(_ + _)) // run task => 6
}
```
流れてくる要素に対して畳み込みを実行するコンビネータ

## Operator
この項ではStreamのに流れる要素を加工するコンビネータ関数を紹介していく
- **map**  

```scala
def map = {
  pub.map(_ + 1).foreach{ i => putStr(s"$i ") } // run task => 2 3 4
}
```
いわゆる `Functor#map`
説明するまでもないが `A => B` の関数を引数に渡すことで流れてくる要素を加工することができる

- **filter**  

```scala
def filter = {
  pub.filter(_ % 2 != 0).foreach{ i => putStr(s"$i ") } // run task => 1 3
}
```
`A => Boolean` の関数を渡すことで流れてくる要素をフィルタリングする

- **collect**  

```scala
def collect_stream = {
    pub.collect {
      case 1 => "one"
      case 2 => "two"
    }.foreach{ i => putStr(s"$i ") } // run task => one two
  }
```
パターンマッチを使うことでFilterと値の加工の両方をすることができる

- **take/drop**  

```scala
def take = {
  pub.take(2).foreach{ i => putStr(s"$i ") } // run task => 1 2
}
def drop = {
  pub.drop(2).foreach{ i => putStr(s"$i ") } // run task => 3
}
```
先頭からn個の要素のみを取得するStreamに変換する/先頭からn個の要素を捨てるStreamに変換する

- **drain**  

```scala
def drain_stream = {
  pub.tap { i => putStr(s"$i ") }.drain.run(Sink.collect) // run task => 1 2 3 (return is Nil)
}
```
流れてきた要素を捨てる(下流に流れなくする)

- **++**  

```scala
def ++ = {
  (pub ++ Stream.fromIterable(4 to 6)).foreach{ i => putStr(s"$i ") } // run task => 1 2 3 4 5 6
}
```
ふたつのStreamを結合する、ただし、先頭Streamが流れ切らない限り、後続Streamは流れない(直列の合成)

- **zip**  

```scala
def zip = {
  pub.zip(Stream.fromIterable(4 to 6)).foreach{ i => putStr(s"$i ") } // run task => (1, 4) (2, 5) (3, 6)
}
```
ふたつのStreamを結合する、要素は２つのStreamから一つずつ取得したペアになる

- **merge**  

```scala
def merge = {
  import scalaz.zio.duration._

  def countup(start: Ref[Int], q: Queue[Int]) = {
    for {
      i <- start.get
      _ <- q.offer(i)
      _ <- start.set(i + 1)
    } yield ()
  }
  for {
    q1 <- Queue.unbounded[Int]
    q2 <- Queue.unbounded[Int]
    s1Start <- Ref.make(1)
    s2Start <- Ref.make(4)
    _ <- countup(s1Start, q1).repeat(Schedule.spaced(1000.millis) && Schedule.recurs(4)).fork
    _ <- countup(s2Start, q2).delay(500.millis).repeat(Schedule.spaced(1000.millis) && Schedule.recurs(4)).fork
    _ <- Stream.fromQueue(q1).merge(Stream.fromQueue(q2)).take(6).foreach{ i => putStr(s"$i ") } // run task => 1 4 2 5 3 4
  } yield ()
}
```
ふたつのStreamを結合する、Trigger自体もマージされ、流れる順番は保証されない(並列の合成)

- **mapAccum**  

```scala
def mapAccum = {
  pub.mapAccum(0) { case (s, i) => i -> s"$s -> $i"}.foreach{ i => putStr(s"$i, ") } // run task => 0 -> 1, 1 -> 2, 2 -> 3
}
```
Stream中に更新可能な状態付きのmap処理

- **mapConcat**  

```scala
def mapConcat = {
  Stream.fromIterable(List("abcd", "cdef")).mapConcat(s => Chunk.fromArray(s.toCharArray)).run(Sink.collect[Char]) // run task => List(a, b, c, d, c, d, e, f)
}
```
`A => Chunk[B]` の関数を渡すことで大きな塊の流れてくるStreamを小さな値のStreamに変換することができる
`Chunk` はZIOライブラリにて提供されている効率的な `Array` のようなもの

- **transduce**  

```scala
def transduce = {
  (pub ++ Stream(4)).transduce {
    Sink.fold[Int, Int, List[Int]](Nil) { case (xs, i) => if (xs.length < 2) Step.more(i :: xs) else Step.done(xs, Chunk(i)) }.map(_.sum)
  }.foreach{ i => putStr(s"$i ") } // run task => 3 7

  //    (pub ++ Stream(4)).transduce { // same behavior
  //      for {
  //        a <- Sink.identity[Int]
  //        b <- Sink.identity[Int]
  //      } yield a + b
  //    }.foreach{ i => putStr(s"$i ") } // run task => 3 7
  }
```
Sinkを引数にわたすことでStream内の複数の値の合成ができるコンビネータ
Sinkを工夫することで様々な凝った加工ができる
一部のSinkはdoシンタックスで書ける（つまり、flatMapを実装している）ためコメントアウト箇所の用に直感的に記述することも可能になっている


- **mapM**  

```scala
def mapM = {
  for {
    q <- Queue.unbounded[Int]
    _ <- pub.mapM { i => q.offer(i).map(_ => s"offer $i") }.foreach{ i => putStr(s"$i ") }
    res <- q.takeAll
  } yield res // run task => offer 1 offer 2 offer 3 (return List(1, 2, 3))
}
```
通常のmapから渡す関数の戻り値が `ZIO[R, E, A]` になったもの
map時にZIOコンテキストでの処理を行うことができる
mapに限らず、他の関数にも渡す関数の戻り値が `ZIO` になっている末尾 `M` versionの関数が多く存在している、使い方は一緒

---

他にも様々な関数が存在しているが、ひとまず自分が調べたのは以上だ
最後に各種関数を実行するメインメソッドとそれを含むソースコードを貼り付けておく、よければ手元で動かしてみて頂けると嬉しいです、それによりさらなる理解が得られることを願います
- [ソースコード](https://gist.github.com/keiSunagawa/a6ad2eb3066e0dd3d5d270f4ab11331d)
 
```scala
object Main extends App {
  import Examples._
  def run(args: List[String]) = {
    def withLog[R <: Clock with Console, E, A](name: String, f: ZIO[R, E, A]): ZIO[R, E, Unit] = {
      for {
        _ <- putStrLn(s"-- ${name}")
        res <- f
        _<- putStrLn(s"${name} res: $res \n")
      } yield ()
    }
    (for {
      _ <- withLog("fromIterable", fromIterable)
      _ <- withLog("bracket", bracket("test.txt"))
      _ <- withLog("fromQueue", fromIterable)
      _ <- withLog("fromEffect", fromEffect)
      _ <- withLog("collect_sink", collect_sink)
      _ <- withLog("readWhile", readWhile)
      _ <- withLog("drain_sink", drain_sink)
      _ <- withLog("identity", identity)
      _ <- withLog("foldLeft", foldLeft)
      _ <- withLog("map", map)
      _ <- withLog("filter", filter)
      _ <- withLog("collect_stream", collect_stream)
      _ <- withLog("take", take)
      _ <- withLog("drop", drop)
      _ <- withLog("++", ++)
      _ <- withLog("zip", zip)
      _ <- withLog("mapM", mapM)
      _ <- withLog("mapConcat", mapConcat)
      _ <- withLog("merge", merge)
      _ <- withLog("mapAccum", mapAccum)
      _ <- withLog("transduce", transduce)
    } yield ()).fold(_ => 1, _ => 0)
  }
}
```

- 結果出力

```
-- fromIterable
1 2 3 fromIterable res: ()

-- bracket
aaa
bbb
ccc
bracket res: ()

-- fromQueue
1 2 3 fromQueue res: ()

-- fromEffect
100 fromEffect res: ()

-- collect_sink
collect_sink res: List(1, 2, 3)

-- readWhile
readWhile res: List(1, 2)

-- drain_sink
1 2 3 drain_sink res: ()

-- identity
identity res: 1

-- foldLeft
foldLeft res: 6

-- map
2 3 4 map res: ()

-- filter
1 3 filter res: ()

-- collect_stream
one two collect_stream res: ()

-- take
1 2 take res: ()

-- drop
3 drop res: ()

-- ++
1 2 3 4 5 6 ++ res: ()

-- zip
(1,4) (2,5) (3,6) zip res: ()

-- mapM
offer 1 offer 2 offer 3 mapM res: List(1, 2, 3)

-- mapConcat
mapConcat res: List(a, b, c, d, c, d, e, f)

-- merge
1 4 2 5 3 4 merge res: ()

-- mapAccum
0 -> 1, 1 -> 2, 2 -> 3, mapAccum res: ()

-- transduce
3 7 transduce res: ()
```