akka.actor.IOManagerでエコーサーバ

akka-actorのライブラリにはIOManagerというクラスがあり、こちらを用いることでActorSystemの裏側でネットワークI/Oを管理してくれます。
通常ネットワークI/Oを利用する場合、C言語ならread()やwrite()などの関数を使い、JavaならSocketからInputStream, OutputStreamを取り出して、プログラム上の読み書きするタイミングを明示的にpull型の操作で指定することになります。select, poll, epollなどでI/O多重化させる場合も基本的には同様です。

IOManagerの場合は、裏のActorが非同期にI/Oを管理してイベントがある度に表のActorに通知(IOMessage)を飛ばしてきます。表のActorはこれらに応答することで、push型の処理を実行することができます。

参考：http://doc.akka.io/docs/akka/2.0.4/scala/io.html

以下で想定しているバージョンは Scala 2.9.2 と Akka-Actor 2.0.4 です。

Echoサーバ

単純なEchoサーバを作ってみます。

import akka.actor.{Actor, ActorSystem, Props}
import akka.actor.{IO, IOManager}
import akka.util.ByteString

/**
 * エコーサーバ
 */
class EchoActor extends Actor {

  /* Actor生成時にループバックの8080ポートでlisten開始 */
  override def preStart { IOManager(context.system).listen("127.0.0.1",8080) }
  
  /* アクセスがあった時の処理 */
  def receive = {
    
    /* 新しいコネクション要求があった時 */
    case IO.NewClient(serverSocket) => serverSocket.accept

    /* データが送られてきた時 */
    case IO.Read(socket, bytes) => {
       /* 送られてきたバイト列をそのまま返す */
       socket.asWritable.write(bytes)
       /* QUITが来た場合はコネクションを切る */
       if (bytes.utf8String.startsWith("QUIT")) socket.close
    }

  }
}

/**
 * Main : Actorの起動のみ
 */
object EchoServer extends App {
  ActorSystem().actorOf(Props[EchoActor])
}

コンパイルと起動に必要なライブラリの設定をbuild.sbtなどに記述します。

resolvers += "Typesafe Repository" at "http://repo.typesafe.com/typesafe/releases/"

libraryDependencies += "com.typesafe.akka" % "akka-actor" % "2.0.4"

試しに sbt から起動してみます。(Actorを停止させる処理を記述していないので、Ctrl-C で強制的に終了してください。）

$ sbt run

telnetで8080ポートにアクセスします。
書き込んだ内容がそのまま返ってるくのが確認できました。

$ telnet 127.0.0.1 8080
Trying 127.0.0.1...
Connected to localhost.
Escape character is '^]'.
hoge
hoge
fuga
fuga
QUIT
QUIT
Connection closed by foreign host.

しかし、このエコーサーバではakka.actor.IOの肝である Iteratee に触れていません。
次回以降説明していきます。