Chat room and bots using Scalaz Stream

chatbot.scala

import scalaz.stream._
import scalaz.concurrent.Task
import scalaz.stream.async

object ChatBot {
  val queue = async.boundedQueue[String](10)

  val reader =
    new Thread {
      override def run(): Unit = {
        while(true) {
          val line = scala.io.StdIn.readLine()
          queue.enqueueOne(line).run
        }
      }
    }


  def excellentBot(input: Process[Task,String]): Process[Task,String] = {
    input.
      filter(line => line.contains("excellent")).
      map(line => "Be excellent to each other. And... PARTY ON, DUDES!")
  }

  def highFiveBot(input: Process[Task,String]): Process[Task,String] = {
    input.
      filter(line => line.contains("o/")).
      map(line => "\\o")
  }

  val read: Process[Task,String] = queue.dequeue
  val write: Sink[Task,String] = sink.lift((msg: String) => Task.delay(println(msg)))
  val console = Exchange(read, write)

  val room = async.topic[String](console.read)

  room.subscribe // Process[Task,String]

  def run() = {
    val bots = List(excellentBot _, highFiveBot _).map(_(room.subscribe))
    val mergedBots = merge.mergeN(Process.emitAll(bots))

    reader.start()
    ((room.subscribe merge mergedBots) to console.write).run.run
  }
}