các scalaz dòng tương đương với Chơi Enumerator.fromCallback Khung là gì

Question

thư viện iteratee

Chơi Khung của định nghĩa một phương pháp Enumerator.fromCallback cho phép các yếu tố được tạo ra dựa trên kết quả của một tương lai:

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] 

bạn có thể thấy một ví dụ tốt đẹp của nó được sử dụng để cung cấp kết quả phân trang từ một dịch vụ Web đây:

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

Mã luồng tương tự như thế nào để thực hiện tương tự? Tôi khá chắc chắn rằng nó có thể được thực hiện bằng cách sử dụng Process.emit hoặc Process.await hoặc có thể Process.eval, nhưng tôi rất muốn xem một ví dụ làm việc. Điều này cũng có thể yêu cầu nâng tương lai scala vào một scalaz Task, mà có một câu trả lời ở đây:

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

Nếu nó làm cho mọi thứ đơn giản hơn, chúng ta có thể bỏ qua các scala Future vs scalaz công tác bit và giả sử chúng ta có một Bài tập.

Answer 1

Để có được scalaz.concurrent.Task từ scala.concurrent.Future bạn có thể sử dụng Task.async, khi bạn đã có nhiệm vụ trong tay của bạn, bạn có thể làm theo cách này:

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)