Maxximiliann
Learning how to use concurrency - Task.async_stream
defmodule Foo do
def remove_single_item_lists(enumerable) do
Enum.flat_map(enumerable, fn
{_, [_, _ | _] = map} -> map
_ -> []
end)
end
end
cars =
%{
"Ferrari" => [
%{color: "Blue", make: "Ferrari", mileage: 120012.481},
%{color: "Red", make: "Ferrari", mileage: 29831.021},
%{color: "Black", make: "Ferrari", mileage: 24030.674},
%{color: "Cobalt", make: "Ferrari", mileage: 412.811}
],
"Koenigsegg" => [
%{color: "Blue", make: "Koenigsegg", mileage: 250.762},
%{color: "Cobalt", make: "Koenigsegg", mileage: 1297.76},
%{color: "Titanium", make: "Koenigsegg", mileage: 5360.336}
],
"Maserati" => [%{color: "Blue", make: "Maserati", mileage: 255.78}],
"Mclaren" => [%{color: "Red", make: "Mclaren", mileage: 15641.469}]
}
iex(5)> Foo.remove_single_item_lists(cars)
[
%{color: "Blue", make: "Ferrari", mileage: 120012.481},
%{color: "Red", make: "Ferrari", mileage: 29831.021},
%{color: "Black", make: "Ferrari", mileage: 24030.674},
%{color: "Cobalt", make: "Ferrari", mileage: 412.811},
%{color: "Blue", make: "Koenigsegg", mileage: 250.762},
%{color: "Cobalt", make: "Koenigsegg", mileage: 1297.76},
%{color: "Titanium", make: "Koenigsegg", mileage: 5360.336}
]
iex(6)> cars |> Task.async_stream(Foo, :remove_single_item_lists, []) |> Enum.map(fn {:ok, val} -> val end)
16:08:35.086 [error] Task #PID<0.357.0> started from #PID<0.394.0> terminating
** (Protocol.UndefinedError) protocol Enumerable not implemented for {"Ferrari", [%{color: "Blue", make: "Ferrari", mileage: 120012.481}, %{color: "Red", make: "Ferrari", mileage: 29831.021}, %{color: "Black", make: "Ferrari", mileage: 24030.674}, %{color: "Cobalt", make: "Ferrari", mileage: 412.811}]} of type Tuple
(elixir 1.10.3) lib/enum.ex:1: Enumerable.impl_for!/1
(elixir 1.10.3) lib/enum.ex:141: Enumerable.reduce/3
(elixir 1.10.3) lib/enum.ex:3383: Enum.flat_map/2
(elixir 1.10.3) lib/task/supervised.ex:90: Task.Supervised.invoke_mfa/2
(elixir 1.10.3) lib/task/supervised.ex:35: Task.Supervised.reply/5
(stdlib 3.12.1) proc_lib.erl:249: :proc_lib.init_p_do_apply/3
Function: &Foo.remove_single_item_lists/1
Args: [{"Ferrari", [%{color: "Blue", make: "Ferrari", mileage: 120012.481}, %{color: "Red", make: "Ferrari", mileage: 29831.021}, %{color: "Black", make: "Ferrari", mileage: 24030.674}, %{color: "Cobalt", make: "Ferrari", mileage: 412.811}]}]
** (EXIT from #PID<0.357.0>) shell process exited with reason: an exception was raised:
** (Protocol.UndefinedError) protocol Enumerable not implemented for {"Ferrari", [%{color: "Blue", make: "Ferrari", mileage: 120012.481}, %{color: "Red", make: "Ferrari", mileage: 29831.021}, %{color: "Black", make: "Ferrari", mileage: 24030.674}, %{color: "Cobalt", make: "Ferrari", mileage: 412.811}]} of type Tuple
(elixir 1.10.3) lib/enum.ex:1: Enumerable.impl_for!/1
(elixir 1.10.3) lib/enum.ex:141: Enumerable.reduce/3
(elixir 1.10.3) lib/enum.ex:3383: Enum.flat_map/2
(elixir 1.10.3) lib/task/supervised.ex:90: Task.Supervised.invoke_mfa/2
(elixir 1.10.3) lib/task/supervised.ex:35: Task.Supervised.reply/5
(stdlib 3.12.1) proc_lib.erl:249: :proc_lib.init_p_do_apply/3
iex(7)> cars |> Task.async_stream(Foo, :remove_single_item_lists, []) |> Tuple.to_list() |> Enum.map(fn {:ok, val} -> val end)
** (ArgumentError) argument error
:erlang.tuple_to_list(#Function<1.67315895/2 in Task.build_stream/3>)
What am I missing? Thanks for all your help! 
Marked As Solved
al2o3cr
The way Task.async_stream calls the function you give it is equivalent to Enum.map. So:
Task.async_stream(cars, Foo, :remove_single_item_lists, [])
# calls `Foo.remove_single_item_lists` just like:
Enum.map(cars, &Foo.remove_single_item_lists/1)
You could refactor your functions like this:
defmodule Foo do
def remove_single_item_lists(enumerable) do
Enum.flat_map(enumerable, &remove_single_item_list/1)
end
def remove_single_item_list({_key, [_, _ | _] = map}), do: map
def remove_single_item_list(_), do: []
end
cars = ...
# note the function being passed here
cars |> Task.async_stream(Foo, :remove_single_item_list, []) |> Enum.flat_map(fn {:ok, val} -> val end)
1
Also Liked
AstonJ
It’s great to hear that you found Matt’s post helpful @Maxximiliann, however the wording of your reply isn’t really fair on everyone else who took the time to try and help you.
If you find a reply doesn’t quite answer your query, it’s usually best to thank the poster for their attempt to help you and to simply say that you still don’t understand, and whether they would kindly elaborate for you. You are more likely to get help this way, not just in that thread but future threads too 
7
Maxximiliann
That is a wonderful suggestion! I’ll certainly keep that in mind for the future.
Thank you!
3
Popular in Questions
Hello, how can I check the Phoenix version ?
Thanks !
New
Hello, can anybody help here..? I have a list of players and I what to delete an element, but every for loop the list is reverting to ori...
New
For some reason my phoenix channels are working for me in my local dev environment, but as soon as I deploy via Docker, I get a 403 error...
New
In templates/appointment/index.html.eex:
<%= for appointment <- @appointments do %>
<tr>
<td><%= appoi...
New
I’m not a pro in using Regex and can’t figure out why the following behaviour happens, especially if we take into account the difference ...
New
Currently suffering from paralysis by [HTTP client] analysis. This is rather unusual in Elixirland as there tends to be consensus on the ...
New
I tried installing
elixir 1.11.2
erlang 23.3.4
via asdf in my zsh shell. Enabled the versions locally and globally.
When I list them ...
New
Other popular topics
Hi,
I am new to Elixir. I am trying to use the DateTime component to insert a date into MySQL however the there seems to be no way to fo...
New
Hi!
In PHP: $_SERVER[‘SERVER_ADDR’] - in Elixir?
Searched the docs for ip address and the web, no good results.
Thanks!
New
Hello, how can I check the Phoenix version ?
Thanks !
New
I am trying to figure out how Mix knows whether the environment is test, dev, or prod – where is this set?
Thanks.
New
For some reason my phoenix channels are working for me in my local dev environment, but as soon as I deploy via Docker, I get a 403 error...
New
Hi. I’ve noticed that Windows Powershell has it’s own IEX command and you cannot access Elixir’s IEX due to the conflict. This isn’t a cr...
New
Latest on Elixir Forum
Sponsor Spotlight
Error tracking, logs, APM, uptime and cron monitoring in one dev-friendly platform. Find and fix problems before your users notice.
Categories:
Sub Categories:
Forums
Popular Tags
- #ecto
- #liveview
- #troubleshooting
- #learning-elixir
- #deployment
- #library
- #erlang
- #testing
- #genserver
- #mix
- #absinthe
- #remote-other
- #otp
- #plug
- #how-to-question
- #macros
- #postgres
- #channels
- #elixirconf
- #exunit
- #discussion
- #code-sync
- #javascript
- #podcasts
- #onsite
- #dialyzer
- #docker
- #authentication
- #umbrella
- #full-time-contract
- #podcasts-by-brainlid
- #ecto-query
- #elixir-ls
- #phoenix_html
- #iex
- #blog-post
- #graphql
- #genstage
- #ai
- #websockets
- #supervisor
- #elixirconf-us
- #advent-of-code
- #distillery
- #processes
- #api
- #forms
- #metaprogramming
- #security
- #hex
Our Sponsors
Build Elixir applications with speed and confidence.
Supporting innovation across the BEAM ecosystem.
We build reliable cloud platforms for business-critical systems.
Catch errors, track performance, monitor hosts and more.
Error tracking, logs, uptime, and Just Enough APM. Built for developers.
Develop your skills with books, videos, and courses.
Producing high quality Elixir screencasts since 2017.
The deployment platform built for Elixir: PaaS ease, VPS control.
The team behind Membrane, Popcorn, LiveDebugger. Available for hire.
Courses that'll move you from confusion to "Aha, now I get it!"