Akka-StreamのRunnableGraphの構築パターン

[Akka-Stream](http://doc.akka.io/docs/akka/current/scala/stream/)でアプリケーションを実装する場合、
SourceとFlowとSinkを組み合わせて`RunnableGraph`を構築し、`run`して実行する。

本記事ではAkka-Streamの部品のメインとなるSource, Flow, Sinkを、
[Akka-Actor](http://doc.akka.io/docs/akka/current/scala/actors.html)および[ActorPublisher](https://github.com/akka/akka/blob/master/akka-stream/src/main/scala/akka/stream/actor/ActorPublisher.scala), [ActorSubscriber](https://github.com/akka/akka/blob/master/akka-stream/src/main/scala/akka/stream/actor/ActorSubscriber.scala)を使う場合/使わない場合でシンプルに実装してみる。

# Actorを使わずにRunnableGraphを構築する

まず、Actorを使わないAkka-Streamなアプリケーションを実装する。
コード全体はこちら
[AkkaStreamStandard.scala](https://github.com/petitviolet/akka-stream-prac/blob/master/src/main/scala/net/petitviolet/ex/akka/stream/AkkaStreamStandard.scala)

今回取り扱うデータ型はこれだけ。

```scala
case class Message(value: String) extends AnyVal
```

この`Message`を追加されたら加工しながら出力するgraphを実装する。

## Source

後からデータを追加できるように`SourceQueueWithComplete`を使う。
`Source.queue[T]`で楽に作れて便利。

```scala
val source: Source[Message, SourceQueueWithComplete[Message]] =
  Source.queue[Message](100, OverflowStrategy.backpressure)
```

## Flow

`Message`を受け取って値を書き換えるものと、畳み込んで`String`にする2つのFlow。
`Flow[T].map`と`Flow[T].fold`で実装する。

```scala
val flow: Flow[Message, Message, NotUsed] =
  Flow[Message].map { r => r.copy(value = s"(Mapped: ${r.value})") }

val accumulater: Flow[Message, String, NotUsed] =
  Flow[Message].fold("init") { (acc, rep) => s"$acc :: ${rep.value}" }
```

## Sink

先程のFlowで作成した`String`を`println`するだけの簡単なSink。

```scala
val sink: Sink[String, Future[Done]] = Sink.foreach[String] { println }
```

## Graph

上で作った`Source`, `Flow`, `Sink`を組み合わせて`RunnableGraph`を構築する。
`via`と`to`でつなげるだけ。

```scala
// simple graph
val graph: RunnableGraph[SourceQueueWithComplete[Message]] =
  source via flow via accumulater to sink
```

## 実行

上の`graph`に`run`を呼んだら手に入る`queue`に対して`offer`でメッセージを詰めるとstreamを流れていく。
`complete`で終了。

```scala
val queue: SourceQueueWithComplete[Message] = graph.run()

queue offer Message("hello!") 
queue offer Message("100") 
queue offer Message("good") 
queue complete
```

実行するとこんなログが出る。

> [info] init :: (Mapped: hello!) :: (Mapped: 100) :: (Mapped: good)

# Actor{Publisher,Subscriber}を使ってRunnableGraphを構築する

コード全体はこちら
[AkkaStreamPracWithActor.scala](https://github.com/petitviolet/akka-stream-prac/blob/master/src/main/scala/net/petitviolet/ex/akka/stream/AkkaStreamPracWithActor.scala)

使用するデータ型は、
先程の`Message`と名前を変えた`Letter`と`queue.complete`に相当するメッセージとして`Finish`を用意した。

```scala
case class Letter(value: String) extends AnyVal
case object Finish
```

## Source

### Source用のActorを実装

Sourceに使用するActorは`ActorPublisher[T]`を実装すれば良い。
publishするメッセージは`receive`内で`onNext(msg)`し、`onComplete`で終了する。

```scala
class PublishActor extends ActorPublisher[Letter] {
  // publish [[Letter]] or OnComplete
  override def receive: Actor.Receive = {
    case s: String =>
      onNext(Letter(s"Nice: $s"))
    case i: Int =>
      onNext(Letter(s"Great: ${i * 100}"))
    case Finish =>
      onComplete()
  }
}
```

### ActorPublisherを使ってSourceを作る

先程の`PublishActor`の`ActorRef`インスタンスを`ActorPublisher.apply`で`Publisher[T]`にしてから、
`Source.fromPublisher`でSourceに変換する

```scala
// publisher actor
val publishActorRef = system.actorOf(Props[PublishActor])

// source with actor
val source: Source[Letter, NotUsed] = {
  val publisher: Publisher[Letter] = ActorPublisher(publishActorRef)
  Source.fromPublisher(publisher)
}
```

## Flow

### Flow用のActorを実装

これは普通に`Actor`をextendsした普通のActor。
返り値を使いたいので`sender`にメッセージを送る。

```scala
class FlowActor extends Actor {
  // subscribe and publish
  override def receive: Actor.Receive = {
    case Letter(msg) => sender() ! Letter(s"(Mapped: $msg)")
    case any        => println(s"??? => $any")
  }
}
```

### Actorを使ってFlowを作る

[StackOverFlow](http://stackoverflow.com/a/39250275/3893741)を参考に実装。
先程の`FlowActor`の`ActorRef`インスタンスに`ask`(`?`)で得られる`Future[T]`を`Flow.mapAsync[T]`でFlowにするだけ。

```scala
// flow
val flow: Flow[Letter, Letter, NotUsed] = {
  import scala.concurrent.duration._
  implicit val timeout: Timeout = 1.second
  val flowActor = system.actorOf(Props[FlowActor])
  def flowWithActor(reply: Letter): Future[Letter] = (flowActor ? reply).mapTo[Letter]

  Flow[Letter].mapAsync[Letter](3)(flowWithActor)
}

// another flow without actor
val accumulater: Flow[Letter, String, NotUsed] =
  Flow[Letter].fold("init") { (acc, rep) => s"$acc :: ${rep.value}" }
```

Actorを使わないFlowもあわせて使ってみる。

## Sink

### Sink用のActorを実装

`ActorSubscriber`を実装したActorにする。
`OnNext`と`OnComplete`が送られてくるので、それを`receive`すればいい。

```scala
class SubscribeActor extends ActorSubscriber {
  override protected def requestStrategy: RequestStrategy = OneByOneRequestStrategy

  // just subscribe
  override def receive: Actor.Receive = {
    case OnNext(any) => println(s"subscribed: $any")
    case OnComplete  => println(s"finish process!")
  }
}
```

### ActorSubscriberを使ってSinkを作る

先程の`SubscribeActor`の`ActorRef`を`ActorSubscriber.apply`したものを`Sink.fromSubscriber`でSourceにする。

```scala
// sink with actor
val sink: Sink[String, NotUsed] = {
  val printActor = system.actorOf(Props[SubscribeActor])
  Sink.fromSubscriber[String](ActorSubscriber[String](printActor))
}
```

## RunnableGraph

これは特に変わらない。
`via`と`to`でSource, Flow, Sinkをつなげればいい。

```scala
// simple graph
val graph: RunnableGraph[NotUsed] = source via flow via accumulater to sink
```

## 実行

`RunnableGraph#run`する。
そして、Sourceの元となった`publishActorRef`に対してメッセージを`tell`すればstreamを流れていく。
`run`して即座にメッセージを流すと`IllegalStateException`が出てしまうのでとりあえず`Thread.sleep`してごまかしている。

```scala
graph.run
// wait preparing graph
Thread.sleep(100L)

publishActorRef ! "hello!"
publishActorRef ! 100
publishActorRef ! "good"
publishActorRef ! Finish
```

実行するとこんなログが出る。

> [info] subscribed: init :: (Mapped: Nice: hello!) :: (Mapped: Great: 10000) :: (Mapped: Nice: good)

# ActorRefを使ってRunnableGraphを構築する

上記では`ActorPublisher`/`ActorSubscriber`を使って実装したが、次は`Source.actorRef`, `Sink.actorRef`を使ってみる。
コード全体はこちら
[AkkaStreamPracWithActorRef.scala](https://github.com/petitviolet/akka-stream-prac/blob/master/src/main/scala/net/petitviolet/ex/akka/stream/AkkaStreamPracWithActorRef.scala)

## Source

`Source.actorRef`でSourceを作る。

```scala
// source with actorRef
val source: Source[Message, ActorRef] = Source.actorRef[Message](65536, OverflowStrategy.fail)
```

これだけ。
Sourceとして使用する`ActorRef`インスタンスを用意する必要すらない。
第一引数はメッセージバッファのサイズで、第二引数はそれがoverflowした時の挙動。
ちなみにこの場合はBackPressureはサポートされていない。

## Flow

先ほどと全く同じものを使うのでここでは省略。

## Sink

### Sink用のActor

まずSink用のActorを実装する。
ただ`println`するだけのシンプルなActor。

```scala
class SubscribeActor extends Actor {
  // just subscribe
  override def receive: Actor.Receive = {
    case Finish => println(s"finish process!")
    case any => println(s"subscribed: $any")
  }
}
```

### Sink.actorRefでSinkを作る

先程のActorの`ActorRef`インスタンスを`Sink.actorRef`に渡すだけで良い。

```scala
// subscriber
val subscribeActor = system.actorOf(Props[SubscribeActor])
// sink with actorRef
val sink: Sink[Any, NotUsed] = Sink.actorRef(subscribeActor, onCompleteMessage = Finish)
```

第二引数の`onCompleteMessage`は、Sink内部の`ActorSubscriber`に対して`OnComplete`が`tell`された時にこの`ActorRef`に対して`tell`される。
[ActorRefSinkActor.scala](https://github.com/akka/akka/blob/master/akka-stream/src/main/scala/akka/stream/impl/ActorRefSinkActor.scala#L38)

## RunnableGraph

`via`と`to`でSource, Flow, Sinkをつなげる点は同じ。
ただ、`RunnableGraph[Mat]`の`Mat`が`ActorRef`になる。

```scala
// simple graph
val graph: RunnableGraph[ActorRef] = source via flow via accumulator to sink
```

## 実行

先程の`graph`に対して`run`を実行すると`ActorRef`が返ってくる。
```scala
val sourceActorRef: ActorRef = graph.run
```

これに対してメッセージを`tell`すればそのメッセージがSourceとしてStreamを流れていく。

```scala
// wait preparing graph
Thread.sleep(100L)

sourceActorRef ! Letter("hello!")
sourceActorRef ! Letter("100")
sourceActorRef ! Letter("good")

// force complete upstream source
sourceActorRef ! PoisonPill
```

今回は途中のFlowで`fold`を使っているため、上流となる`sourceActorRef`を終了させることで次にstreamを無理やり流している。
実行結果は以下。

> [info] subscribed: init :: (Mapped: hello!) :: (Mapped: 100) :: (Mapped: good)

# まとめ

RunnableGraphの構築にActorを使わない, ActorPublisher/ActorSubscriber, ActorRefを使う3パターンで実装した。

Actorを使うやり方だとそもそもActorを多少知らないと使えない点ではコスト高にはなるが、
開発アーキテクチャのベースとしてActorを使用しているケースではStreamでも統一してActorを使えるというのは利点になるはず。

ただ、[Konradさん(@ktosopl)](https://twitter.com/ktosopl)が[StackOverFlow](http://stackoverflow.com/a/39132662/3893741)で

> Please don't use ActorPublisher and ActorSubscriber. They're too low level and you might end up implementing them in such a way that's violating the Reactive Streams specification. 

と言っているので`ActorPublisher`/`ActorSubscriber`を使って実装するのは避けた方が良さそう。
なので、これらを使うくらいなら`ActorRef`を使うべきだが、
Akka-StreamがActorの存在をせっかく隠してくれているので明示的に使わずにすむならそれがベターかと。
必要のないものを無理に使わない、という当たり前の感想となってしまった。