Ensure async streams can be consumed from a separate process

Author: josevalim

lib/elixir/lib/task.ex

@@ -657,12 +657,14 @@ defmodule Task do
   end
 
   defp build_stream(enumerable, fun, options) do
-    owner = self()
+    fn acc, acc_fun ->
+      owner = get_owner(self())
 
-    &Task.Supervised.stream(enumerable, &1, &2, get_callers(owner), fun, options, fn ->
-      {:ok, pid} = Task.Supervised.start_link(get_owner(owner), :nomonitor)
-      {:ok, :link, pid}
-    end)
+      Task.Supervised.stream(enumerable, acc, acc_fun, get_callers(self()), fun, options, fn ->
+        {:ok, pid} = Task.Supervised.start_link(owner, :nomonitor)
+        {:ok, :link, pid}
+      end)
+    end
   end
 
   # Returns a tuple with the node where this is executed and either the

lib/elixir/lib/task/supervisor.ex

@@ -484,20 +484,22 @@ defmodule Task.Supervisor do
   end
 
   defp build_stream(supervisor, link_type, enumerable, fun, options) do
-    shutdown = options[:shutdown]
-    owner = self()
+    fn acc, acc_fun ->
+      shutdown = options[:shutdown]
+      owner = get_owner(self())
 
-    &Task.Supervised.stream(enumerable, &1, &2, get_callers(owner), fun, options, fn ->
-      args = [get_owner(owner), :monitor]
+      Task.Supervised.stream(enumerable, acc, acc_fun, get_callers(self()), fun, options, fn ->
+        args = [owner, :monitor]
 
-      case start_child_with_spec(supervisor, args, :temporary, shutdown) do
-        {:ok, pid} ->
-          if link_type == :link, do: Process.link(pid)
-          {:ok, link_type, pid}
+        case start_child_with_spec(supervisor, args, :temporary, shutdown) do
+          {:ok, pid} ->
+            if link_type == :link, do: Process.link(pid)
+            {:ok, link_type, pid}
 
-        {:error, :max_children} ->
-          {:error, :max_children}
-      end
-    end)
+          {:error, :max_children} ->
+            {:error, :max_children}
+        end
+      end)
+    end
   end
 end

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

@@ -450,6 +450,15 @@ defmodule Task.SupervisorTest do
 
       assert_receive :done
     end
+
+    test "consuming from another process", config do
+      parent = self()
+      stream = Task.Supervisor.async_stream(config[:supervisor], [1, 2, 3], &send(parent, &1))
+      Task.start(Stream, :run, [stream])
+      assert_receive 1
+      assert_receive 2
+      assert_receive 3
+    end
   end
 
   describe "async_stream_nolink" do

lib/elixir/test/elixir/task_test.exs

@@ -829,6 +829,15 @@ defmodule TaskTest do
 
       assert_receive :done
     end
+
+    test "consuming from another process" do
+      parent = self()
+      stream = Task.async_stream([1, 2, 3], &send(parent, &1))
+      Task.start(Stream, :run, [stream])
+      assert_receive 1
+      assert_receive 2
+      assert_receive 3
+    end
   end
 
   for {desc, concurrency} <- [==: 4, <: 2, >: 8] do