generator2stream

object Main {
  def generator2stream[T](generator: (T=>Unit)=>Unit): Stream[T] = {
    val queue = new java.util.concurrent.SynchronousQueue[Option[T]]

    val callbackthread = new Runnable {
      def run() {
        generator { (Some(_:T)) andThen (queue.put(_)) }
        queue.put(None)
      }
    }   

    new Thread(callbackthread).start()
    Stream.continually(queue.take).takeWhile(_.isDefined).map(_.get)
  }

  def bench(issue: String)(block: => Unit) {
    val t1 = System.nanoTime
    block
    val t2 = System.nanoTime

    println("Duration for " + issue + " " + (t2-t1)/1000.0/1000.0)
  }

  def main(args: Array[String]) {
    def data(block: Int => Unit) {
      for (i <- 0 to 1000000) block(i)
    }

    for (i <- 1 to 3) {
      bench("generator") {
        data { i => i }
      }

      bench("stream") {
        generator2stream(data).foreach { i => i }
      }
    }
  }
}