scala - 什么是与 Play Framework 的 Enumerator.fromCallback 等效的 scalaz-stream

Question

Play Framework 的 iteratee 库定义了一个方法 Enumerator.fromCallback它允许根据 Future 的结果生成元素:

http://www.playframework.com/documentation/2.2.x/Enumerators

def fromCallback[E](
  retriever: () => Future[Option[E]],
  onComplete: () => Unit = () => (),
  onError: (String, Input[E]) => Unit = (_: String, _: Input[E]) => ()
): Enumerator[E]

您可以在此处看到一个很好的示例，它用于从 Web 服务提供分页结果:

http://engineering.klout.com/2013/01/iteratees-in-big-data-at-klout/

def pagingEnumerator(url:String):Enumerator[JsValue]={
  var maybeNextUrl = Some(url) //Next url to fetch
  Enumerator.fromCallback[JsValue] ( retriever = {
    val maybeResponsePromise =
      maybeNextUrl map { nextUrl=>  
        WS.url(nextUrl).get.map { reponse =>
          val json = response.json
          maybeNextUrl = (json \ "next_url").asOpt[String]
          val code = response.status //Potential error handling here
          json
        }   
      }

    /* maybeResponsePromise will be an Option[Promise[JsValue]].
     * Need to 'flip' it, to make it a Promise[Option[JsValue]] to
     * conform to the fromCallback constraints */
    maybeResponsePromise match {
      case Some(responsePromise) => responsePromise map Some.apply
      case None => PlayPromise pure None
    }
  })
}

执行相同操作的等效 scalaz-stream 代码是什么？我很确定它可以使用 Process.emit 来完成或 Process.await或者 Process.eval ，但我很想看到一个有效的例子。这可能还需要将 scala Future 提升为 scalaz 任务，对此有一个答案:

Convert scala 2.10 future to scalaz.concurrent.Future // Task

如果它使事情更简单，我们可以忽略 scala Future 与 scalaz Task 位并假设我们有一个 Task。

Answer 1

要从 scala.concurrent.Future 获取 scalaz.concurrent.Task，您可以使用 Task.async，当您手上有任务时，您可以这样做:

  import java.util.concurrent.atomic.AtomicInteger
  import scalaz.concurrent.Task
  import scalaz.stream.Process.End
  import scalaz.stream._

  val cnt = new AtomicInteger(0)

  val task: Task[String] = Task {
    if (cnt.incrementAndGet() <= 10) s"Task ${cnt.get}" else throw End
  }

  Process.repeatEval(task).runLog.run.foreach(println)