Hello For each item in a list, I am requesting an API and based on the response, I decide to call the same API again, or proceed . I have something like that: items |> Enum.map(&Task.async(fn -> item =&1) HTTPoison.get(url) end)) |> Enum.map(&Task.await/1) |> Enum.map(fn(response) -> case respo…

[image] sabri: How can I do a repeat while inside a pipe above? or what’s the recommended approach here? I think what is happening here is that you have broken down your tasks too far. Your first task ends with: HTTPoison.get(url) but your code suggests that the “task” isn’t really done …

In addition to what @peerreynders was suggesting you might want to take a look at Task.async_stream/3 which offers some useful options to deal with concurrent tasks that are started from an enumerable.

Thanks, I think using a genserver to handle the blocking call and return the final response would give more flexibility to handle possible errors inside the genserver

When does the stream resulted from Task.async_stream/3 actually run? is it when iterated into enumerable? like the example at documentation: Enum.reduce(stream, 0, fn {:ok, num}, acc -> num + acc end) ?

[image] sabri: is it when iterated into enumerable? like the example at documentation: Yes. Streams are lazily evaluated, namely only when you actually try to use them like with Enum.each or Enum.map.

Stream.transform/4 has a way to halt when your conditions are met. https://hexdocs.pm/elixir/Stream.html#transform/4

[image] sabri: I think using a genserver to handle the blocking call and return the final response would give more flexibility to handle possible errors inside the genserver Not necessarily, using Task.async_stream gives you the same, you just have to handle error inputs in the function that …

[image] dimitarvp: Using GenServer is additional and unnecessary complexity. Using Task.async_stream also parallelizes work amongst all CPU cores transparently As far as I understand it Task.async_stream/5 simply offers concurrency control and the :max_concurrency setting simply defaults to S…

[image] dimitarvp: Using GenServer is additional and unnecessary complexity. Using Task.async_stream also parallelizes work amongst all CPU cores transparently I am using Enum.map with GenServer in parallel as follows: vectors= Enum.take(destinations, rows) |> Enum.with_index |> …

[image] sabri: but how would I handle errors or process crashes? Look into Task.Supervisor . For a contrived code example look here .

Repeat previous pipe while condition?

Questions & Help Questions

peerreynders July 18, 2018, 2:19pm 11

Look into Task.Supervisor. For a contrived code example look here.