Qual è lo scalaz-stream equivalente a giocare Enumerator.fromCallback del quadro

Question

biblioteca iteratee

Play Framework definisce un metodo Enumerator.fromCallback che permette gli elementi per essere generati sulla base dei risultati di un futuro:

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] 

Puoi vedere un bell'esempio che viene utilizzato per fornire risultati impaginati da un servizio Web qui:

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 
    } 
    }) 
} 

Qual è il codice scalaz-stream equivalente per fare lo stesso? Sono abbastanza sicuro che può essere fatto usando Process.emit o Process.await o forse Process.eval, ma mi piacerebbe vedere un esempio elaborato. Questo potrebbe anche richiedere il sollevamento del futuro scala in uno scalaz compito, per il quale c'è una risposta qui:

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

Se fa le cose più semplici, si può ignorare la Scala futuro vs scalaz Task bit e supponiamo di avere una Compito.

Answer 1

Per ottenere risultati scalaz.concurrent.Task da scala.concurrent.Future è possibile utilizzare Task.async, quando hai compito in mano si può fare in questo modo:

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)