feed a scalaz Process with recursive async function

Feed.scala

/**
My problem is: 

I have a function getData[A](params ...): Task[List[A]] that return asynchronously data
But this data is big so I cut the call in several step playing with params

In term of API, I want expose a function stream[A]: Process[Task, A]

I see Process.repeatEval & stream.async.unboundedQueue API from scalaz 
but I don't understand how to use it to produce this result
*/
object Feed[A] {
  // return a piece of needed data
  private def getData(start: Int, chunk: Int): Task[List[A]] = ???
  
  // the public API I want to expose
  def stream: Process[Task, A]
}

I try this :

import scalaz.stream.Process
import scalaz.stream.async
import scalaz.stream.async.mutable.Queue

object Feed[A] {
  private def getData(start: Int, chunk: Int): Task[List[A]] = ???

  private def rec(q: Queue[A], start: Int, chunk: Int, end: Boolean): Task[Unit] = end match {
    case false => {
      getData(start, chunk)
        .flatMap { q.enqueueAll }
        .flatMap { _ => rec(q, start + chunk, chunk, isEnd(end)) }
    }
    case true => q.close()
  }

  def stream: Process[Task, A] = {
    val q: Queue[A] = async.unboundedQueue[E]

    val result = rec(q, 0, 250, false).map { _ => q.dequeue }

    Process eval(result) flatMap(x => x)
  }
}

The problem is my Process reference is render only when all my call to getData are Done and so I stream nothing here.

Have you any idea how i can peform this ?

I think my pb is to exec rec in other thread than q.dequeue, so don't use map function but I don't know how to do this.

Two things:

• No need for queues for something like this (although it would work fine with it, let's not complicate things too much)
• You are doing your recursion at the Task level, that is, within a single Task! Remember, a Task is like a Future, it only returns a value once, at the end of whatever computation it was doing, hence why you have nothing actually streaming.

This should explain better:

import scalaz._
import scalaz.concurrent.Task
import scalaz.stream.Process
import scalaz.stream.async
import scalaz.stream.async.mutable.Queue

trait Feed[A] {

  protected def getData(start: Int, chunk: Int): Task[List[A]]

  protected def isEnd(i: Int): Boolean


  // we do the recursion at the Process level instead, halting when necessary
  private def getChunks(start: Int, chunk: Int): Process[Task, List[A]] =
    if (isEnd(start)) Process.halt  
    else Process.eval(getData(start, start + chunk)) ++ getChunks(start + chunk, chunk)

  // let's flatten the feed
  def stream: Process[Task, A] = getChunks(0, 250).flatMap(Process.emitAll)
}

object TestFeed extends Feed[Int] {
  protected def getData(start: Int, chunk: Int): Task[List[Int]] = Task { 
    println("debug: getting data!")
    Range(start,chunk).toList 
  }

  protected def isEnd(i: Int) = i >= 1000
}

object Run extends App {
  // this will output the integer 0 to 999 with "getting data!" interleaved in between
  // instead of block of it at the beginning
  TestFeed.stream.map(i => print(s"$i;")).run.run
}

Si getData ne renvoi plus que une liste de E mais un résultat plus complexe qui contient notamment la condition de sorti de la récursion.

J'ai tenté ça. J'ai bon ?

trait Feed[A] {
  case class ComplexResult[A](result: List[A], nextChunk: Int, endOfData: Boolean)

  protected def getData(start: Int, chunk: Int): Task[ComplexResult[A]]

  protected def isEnd(i: Int): Boolean

  // we do the recursion at the Process level instead, halting when necessary
  private def getChunks(start: Int, chunk: Int, end: Boolean): Process[Task, A] =
    if (end) Process.halt  
    else {  
      val chunk: Task[ReadStreamEventsCompleted] = processChunk(streamUid, nextEventNumber, chunkSize)

      chunk.map { result =>
        Process.emitAll(readEvent(result)) ++ getChunks(streamUid, result.nextChunk, chunkSize, result.endOfData)
      } |> flattenTaskProcess 
    }

  private def flattenTaskProcess[A](f: Task[Process[Task, A]]): Process[Task, A] = Process eval(f) flatMap(x => x)

  def stream: Process[Task, A] = getChunks(0, 250, false)
}

When the end of recursion is decide by getData result :

private def getChunks(start: Int, chunk: Int): Process[Task, ComplexResult] =
    Process.await(getData(start, start + chunk))({ c: ComplexResult =>
      val next = if (c.endOfData) Process.halt else getChunks(start + chunk, chunk)
      Process.emit(c) ++ next
    })

private def getChunks(start: Int, chunk: Int): Process[Task, ComplexResult] =
    Process.await(getData(start, start + chunk))({ c: ComplexResult =>
        val next = if (c.endOfData) Process.halt else getChunks(start + chunk, chunk)
        Process.emit(c) ++ next
    })