генератор / блок в итератор / преобразование потока

EDITED: изменены примеры, чтобы показать лень 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, а затем вызвав для него "view". Если вам действительно нужен тип 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".

Вот простое решение, которое порождает поток, потребляющий данные. Он отправляет данные в 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)
}   

Мне все еще нужно придумать, как это сделать. Я подозреваю, что ответ лежит где-то здесь:

• Блог Джима Макбита, например Автономный универсальный генератор Scala
• Другой блог - Notas van de Vos Генераторы на Scala

Изменить: удален код, показывающий, как решить другую проблему.

Edit2: использование кода http://gist.github.com/580157, который изначально был опубликован http://gist.github.com/574873, вы можете сделать это:

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.

Вот реализация на основе продолжений с разделителями, адаптированная из предложения @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)
    })
}