Skip to content

Instantly share code, notes, and snippets.

@ubourdon
Created August 24, 2016 08:44
Show Gist options
  • Select an option

  • Save ubourdon/b9c24bdb25badac1c8363c7a3d27da12 to your computer and use it in GitHub Desktop.

Select an option

Save ubourdon/b9c24bdb25badac1c8363c7a3d27da12 to your computer and use it in GitHub Desktop.
feed a scalaz Process with recursive async function
/**
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]
}
@ubourdon

Copy link
Copy Markdown
Author

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

@ubourdon

Copy link
Copy Markdown
Author
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
    })

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment