Scalaで Twitter ストリーミングAPI

ソース

class Streaming(id: String, pw: String) {

  import _root_.java.io.InputStream
  import _root_.java.net.{ Authenticator, PasswordAuthentication, URL }

  import _root_.scala.collection.mutable.ListBuffer
  import _root_.scala.io.Source

  import _root_.scala.actors.{ Actor, TIMEOUT }

  val url = new URL("http://stream.twitter.com/1/statuses/sample.xml?delimited=length")

  Authenticator.setDefault(new Authenticator {
    override def getPasswordAuthentication = {
      new PasswordAuthentication(id, pw.toCharArray)
    }
  })

  val handlers = ListBuffer[String => Unit]()

  private var in: InputStream = null

  def start {
    streaming.getState match {
      case Actor.State.New => {
        in = url.openStream
        streaming.start
      }
      case Actor.State.Terminated => throw new IllegalStateException("Streaming was closed.")
      case _ => throw new IllegalStateException("Streaming has already been started.")
    }
  }

  def close {
    try {
      streaming.getState match {
        case Actor.State.New => throw new IllegalStateException("Streaming has not been started yet.")
        case Actor.State.Terminated => throw new IllegalStateException("Streaming was closed.")
        case _ => {
          streaming !? CLOSE
        }
      }
    }
    finally {
      if(in != null) {
        in.close
        in = null
      }
    }
  }

  private case class CLOSE()

  private object streaming extends Actor {
    def act = {
      val source = Source.fromInputStream(in)

      def readState = {
        val length = source.getLine(1).replaceAll("\\s", "").toInt
        source.take(length) mkString
      }

      loop {
        receiveWithin(0) {
          case TIMEOUT => {
            val state = readState
            handlers foreach { handler => handler(state) }
          }
          case CLOSE => {
            reply()
            exit
          }
        }
      }
    }
  }

}

使い方

val twitter = new Streaming("userid", "password")
twitter.handlers += { xml => println(xml) } // ハンドラ { (String) => Unit } 設定
twitter.start

・・・

twitter.close