生成器/块到迭代器/流的转换

Question

基本上我想转换这个：

def data(block: T => Unit)

到一个流（dataToStream 是一个进行这种转换的假设函数）：

val dataStream: Stream[T] = dataToStream(data)

我想这个问题可以通过延续来解决：

// let's assume that we don't know how data is implemented
// we just know that it generates integers
def data(block: Int => Unit) { for (i <- 0 to 10) block(i) }

// here we can print all data integers
data { i => println(i) }

// >> but what we really want is to convert data to the stream <<

// very dumb solution is to collect all data into a list
var dataList = List[Int]()
data { i => dataList = i::dataList }
// and make a stream from it
dataList.toStream

// but we want to make a lazy, CPU and memory efficient stream or iterator from data
val dataStream: Stream[Int] = dataToStream(data)
dataStream.foreach { i => println(i) }

// and here a black magic of continuations must be used
// for me this magic is too hard to understand
// Does anybody know how dataToStream function could look like?

谢谢，大卫

Answer 1

已编辑：修改示例以显示 traversable.view 的惰性

scala> def data(f : Int => Unit) = for(i <- 1 to 10) {    
     |   println("Generating " + i)
     |   f(i)
     | }
data: (f: (Int) => Unit)Unit

scala> def toTraversable[T]( func : (T => Unit) => Unit) = new Traversable[T] {
     |   def foreach[X]( f : T => X) = func(f(_) : Unit)                       
     | }                                                                       
toTraversable: [T](func: ((T) => Unit) => Unit)java.lang.Object with Traversable[T]

toTraversable 方法会将您的数据函数转换为 Traversable 集合。就其本身而言，它没什么大不了的，但您可以将其转换为惰性的 TraversableView。这是一个例子：

scala> toTraversable(data).view.take(3).sum
Generating 1
Generating 2
Generating 3
Generating 4
res1: Int = 6

take 方法的不幸性质是它必须超过最后一个生成的值才能正常工作，但它会提前终止。如果没有“.view”调用，上面的代码看起来是一样的。然而，这里有一个更引人注目的例子：

scala> toTraversable(data).view.take(2).foreach(println)
Generating 1
1
Generating 2
2
Generating 3

因此，总而言之，我相信您正在寻找的集合是 TraversableView，它最容易创建创建 Traversable 的视图，然后在其上调用“视图”。如果你真的想要 Stream 类型，这里有一个在 2.8.0.final 中工作的方法，它会生成一个没有线程的“Stream”：

scala> def dataToStream( data : (Int => Unit) => Unit) = {
     |   val x = new Traversable[Int] {                     
     |     def foreach[U](f : Int => U) = {                 
     |        data( f(_) : Unit)                            
     |     }
     |   }
     |   x.view.toList.toStream                             
     | }
dataToStream: (data: ((Int) => Unit) => Unit)scala.collection.immutable.Stream[Int]

scala> dataToStream(data)
res8: scala.collection.immutable.Stream[Int] = Stream(0, ?)

这种方法的不幸之处在于它会在创建流之前遍历整个可遍历对象。这也意味着所有值都需要在内存中缓冲。唯一的选择是诉诸线程。

顺便说一句：这是更喜欢 Traversables 作为 scalax.io.File 方法的直接返回的动机：“lines”、“chars”和“bytes”。

Answer 2

这是一个简单的解决方案，它产生一个使用数据的线程。它将数据发布到 SynchronousQueue。创建并返回从队列中提取数据的流：

 def generatortostream[T](f: (T=>Unit)=>Unit): Stream[T] = {
  val queue = new java.util.concurrent.SynchronousQueue[Option[T]]
  val callbackthread = new Runnable {
    def run() { f((Some(_:T)) andThen (queue.put(_))); queue.put(None) }
  }   
  new Thread(callbackthread).start()
  Stream.continually(queue.take).takeWhile(_.isDefined).map(_.get)
}   

Answer 3

我仍然需要自己弄清楚如何做到这一点。我怀疑答案就在这里：

• Jim McBeath 的博客，例如 Standalone Generic Scala Generator
• 另一个博客 - Notas van de Vos Generators in Scala

编辑：删除了显示如何解决不同问题的代码。

Edit2：使用最初发布http://gist.github.com/574873 的代码http://gist.github.com/580157，您可以这样做：

object Main {
  import Generator._

  def data = generator[Int] { yld =>
    for (i <- suspendable(List.range(0, 11))) yld(i)
  }

  def main(args: Array[String]) {
    for( i <- data.toStream ) println(i)
  }
}

data 不采用块代码，但我认为这很好，因为通过继续，调用者可以处理块。生成器的代码可以在 github 上的 gist 中看到。

Answer 4

这是一个基于分隔的基于延续的实现，改编自 @Geoff Reedy 的产品：

import Stream._
import scala.util.continuations._
import java.util.concurrent.SynchronousQueue

def toStream[A](data: (A=>Unit)=>Unit):Stream[A] = reset {
    val queue = new SynchronousQueue[Option[A]]
    queue.put(Some(shift { k: (A=>Unit) =>
        new Thread() { 
            override def run() {
                data(k)
                // when (if) the data source stops pumping, add None 
                // to signal that the stream is dead
                queue.put(None)
            }
        }.start()
        continually(queue.take).takeWhile(_.isDefined).map(_.get)
    })
}