Keep Task.async_stream backwards compatible

Author: José Valim

lib/elixir/lib/task/supervised.ex

@@ -210,8 +210,8 @@ defmodule Task.Supervised do
       # this response when the replying task dies (we'll notice in the :down
       # message).
       {{^monitor_ref, position}, reply} ->
-        %{^position => {pid, :running, element}} = waiting
-        waiting = Map.put(waiting, position, {pid, {:ok, reply}, element})
+        %{^position => {pid, :running}} = waiting
+        waiting = Map.put(waiting, position, {pid, {:ok, reply}})
         stream_reduce({:cont, acc}, max, spawned, delivered, waiting, next, config)
 
       # The task at position "position" died for some reason. We check if it
@@ -223,20 +223,20 @@ defmodule Task.Supervised do
           case waiting do
             # If the task replied, we don't care whether it went down for timeout
             # or for normal reasons.
-            %{^position => {_, {:ok, _} = ok, _}} ->
+            %{^position => {_, {:ok, _} = ok}} ->
               ok
             # If the task exited by itself before replying, we emit {:exit, reason}.
-            %{^position => {_, :running, element}} when kind == :down ->
-              {:exit, reason, element}
+            %{^position => {_, :running}} when kind == :down ->
+              {:exit, reason}
             # If the task timed out before replying, we either exit (on_timeout: :exit)
             # or emit {:exit, :timeout} (on_timeout: :kill_task) (note the task is already
             # dead at this point).
-            %{^position => {_, :running, element}} when kind == :timed_out ->
+            %{^position => {_, :running}} when kind == :timed_out ->
               if on_timeout == :exit do
                 stream_cleanup_inbox(monitor_pid, monitor_ref)
-                exit({:timeout, {__MODULE__, :stream, [element, timeout]}})
+                exit({:timeout, {__MODULE__, :stream, [timeout]}})
               else
-                {:exit, :timeout, element}
+                {:exit, :timeout}
               end
           end
 
@@ -364,7 +364,7 @@ defmodule Task.Supervised do
     receive do
       {:spawned, {^monitor_ref, ^spawned}, pid} ->
         send(pid, {self(), {monitor_ref, spawned}})
-        Map.put(waiting, spawned, {pid, :running, value})
+        Map.put(waiting, spawned, {pid, :running})
       {:DOWN, ^monitor_ref, _, ^monitor_pid, reason} ->
         stream_cleanup_inbox(monitor_pid, monitor_ref)
         exit({reason, {__MODULE__, :stream, [timeout]}})

lib/elixir/test/elixir/task/supervisor_test.exs

@@ -221,7 +221,7 @@ defmodule Task.SupervisorTest do
       assert supervisor
              |> Task.Supervisor.async_stream(1..4, &exit(Integer.to_string(&1)), @opts)
              |> Enum.to_list ==
-             [{:exit, "1", 1}, {:exit, "2", 2}, {:exit, "3", 3}, {:exit, "4", 4}]
+             [exit: "1", exit: "2", exit: "3", exit: "4"]
     end
 
     test "shuts down unused tasks", %{supervisor: supervisor} do
@@ -277,7 +277,7 @@ defmodule Task.SupervisorTest do
       assert supervisor
              |> Task.Supervisor.async_stream_nolink(1..4, &exit/1, @opts)
              |> Enum.to_list ==
-             [{:exit, 1, 1}, {:exit, 2, 2}, {:exit, 3, 3}, {:exit, 4, 4}]
+             [exit: 1, exit: 2, exit: 3, exit: 4]
     end
 
     test "shuts down unused tasks", %{supervisor: supervisor} do

lib/elixir/test/elixir/task_test.exs

@@ -492,7 +492,7 @@ defmodule TaskTest do
   describe "async_stream/2" do
     test "timeout" do
       assert catch_exit([:infinity] |> Task.async_stream(&sleep/1, [timeout: 0]) |> Enum.to_list) ==
-             {:timeout, {Task.Supervised, :stream, [:infinity, 0]}}
+             {:timeout, {Task.Supervised, :stream, [0]}}
       refute_received _
     end
 
@@ -509,7 +509,7 @@ defmodule TaskTest do
     test "streams an enumerable with ordered: false, on_timeout: :kill_task" do
       opts = [max_concurrency: 4, ordered: false, on_timeout: :kill_task, timeout: 50]
       assert [100, 1, 100, 1] |> Task.async_stream(&sleep/1, opts) |> Enum.to_list() ==
-             [{:ok, 1}, {:ok, 1}, {:exit, :timeout, 100}, {:exit, :timeout, 100}]
+             [ok: 1, ok: 1, exit: :timeout, exit: :timeout]
       refute_received _
     end
   end
@@ -543,7 +543,7 @@ defmodule TaskTest do
       test "streams an enumerable with exits" do
         Process.flag(:trap_exit, true)
         assert 1..4 |> Task.async_stream(&exit/1, @opts) |> Enum.to_list ==
-               [{:exit, 1, 1}, {:exit, 2, 2}, {:exit, 3, 3}, {:exit, 4, 4}]
+               [exit: 1, exit: 2, exit: 3, exit: 4]
         refute_received {:EXIT, _, _}
       end
 
@@ -586,7 +586,7 @@ defmodule TaskTest do
       test "with inner halt on failure" do
         Process.flag(:trap_exit, true)
         assert 1..8 |> Stream.take(4) |> Task.async_stream(&exit/1, @opts) |> Enum.to_list ==
-               [{:exit, 1, 1}, {:exit, 2, 2}, {:exit, 3, 3}, {:exit, 4, 4}]
+               [exit: 1, exit: 2, exit: 3, exit: 4]
       end
 
       test "with inner halt and slowest first" do
@@ -602,7 +602,7 @@ defmodule TaskTest do
       test "with outer halt on failure" do
         Process.flag(:trap_exit, true)
         assert 1..8 |> Task.async_stream(&exit/1, @opts) |> Enum.take(4) ==
-               [{:exit, 1, 1}, {:exit, 2, 2}, {:exit, 3, 3}, {:exit, 4, 4}]
+               [exit: 1, exit: 2, exit: 3, exit: 4]
       end
 
       test "with outer halt and slowest first" do
@@ -639,7 +639,7 @@ defmodule TaskTest do
       test "with :on_timeout set to :kill_task" do
         opts = Keyword.merge(@opts, on_timeout: :kill_task, timeout: 50)
         assert [100, 1, 100, 1] |> Task.async_stream(&sleep/1, opts) |> Enum.to_list() ==
-               [{:exit, :timeout, 100}, {:ok, 1}, {:exit, :timeout, 100}, {:ok, 1}]
+               [exit: :timeout, ok: 1, exit: :timeout, ok: 1]
         refute_received _
       end
     end