Add async supervised (#2818)

* test: adds test for async-supervised

* test: completes tests on async-supervised functionality

* doc: adds documentation for the assign_async/4 supervisor option

* Update lib/phoenix_live_view.ex

Co-authored-by: José Valim <jose.valim@gmail.com>

* Update lib/phoenix_live_view/async.ex

Co-authored-by: José Valim <jose.valim@gmail.com>

---------

Co-authored-by: José Valim <jose.valim@gmail.com>
Co-authored-by: Chris McCord <chris@chrismccord.com>

lib/phoenix_live_view.ex

@@ -1833,6 +1833,11 @@ defmodule Phoenix.LiveView do
   and the result when the function completes.
 
   The task is only started when the socket is connected.
+   
+  ## Options
+    
+    * `:supervisor` - allows you to specify a `Task.Supervisor` to supervise the task.
+
 
   ## Examples
 
@@ -1858,12 +1863,11 @@ defmodule Phoenix.LiveView do
         # ...
         send_update(parent, Component, data)
       end)
-
   """
-  def assign_async(%Socket{} = socket, key_or_keys, func)
+  def assign_async(%Socket{} = socket, key_or_keys, func, opts \\ [])
       when (is_atom(key_or_keys) or is_list(key_or_keys)) and
              is_function(func, 0) do
-    Async.assign_async(socket, key_or_keys, func)
+    Async.assign_async(socket, key_or_keys, func, opts)
   end
 
   @doc """

lib/phoenix_live_view/async.ex

@@ -3,11 +3,11 @@ defmodule Phoenix.LiveView.Async do
 
   alias Phoenix.LiveView.{AsyncResult, Socket, Channel}
 
-  def start_async(%Socket{} = socket, key, func) when is_function(func, 0) do
-    run_async_task(socket, key, func, :start)
+  def start_async(%Socket{} = socket, key, func, opts \\ []) when is_function(func, 0) do
+    run_async_task(socket, key, func, :start, opts)
   end
 
-  def assign_async(%Socket{} = socket, key_or_keys, func)
+  def assign_async(%Socket{} = socket, key_or_keys, func, opts \\ [])
       when (is_atom(key_or_keys) or is_list(key_or_keys)) and
              is_function(func, 0) do
     keys = List.wrap(key_or_keys)
@@ -49,14 +49,21 @@ defmodule Phoenix.LiveView.Async do
 
     socket
     |> Phoenix.Component.assign(new_assigns)
-    |> run_async_task(keys, wrapped_func, :assign)
+    |> run_async_task(keys, wrapped_func, :assign, opts)
   end
 
-  defp run_async_task(%Socket{} = socket, key, func, kind) do
+  defp run_async_task(%Socket{} = socket, key, func, kind, opts) do
     if Phoenix.LiveView.connected?(socket) do
       lv_pid = self()
       cid = cid(socket)
-      {:ok, pid} = Task.start_link(fn -> do_async(lv_pid, cid, key, func, kind) end)
+      {:ok, pid} = if supervisor = Keyword.get(opts, :supervisor) do
+        Task.Supervisor.start_child(supervisor, fn ->
+          Process.link(lv_pid)
+          do_async(lv_pid, cid, key, func, kind)
+        end)
+      else
+        Task.start_link(fn -> do_async(lv_pid, cid, key, func, kind) end)
+      end
 
       ref =
         :erlang.monitor(:process, pid, alias: :reply_demonitor, tag: {__MODULE__, key, cid, kind})

test/phoenix_live_view/integrations/assign_async_test.exs

@@ -158,4 +158,30 @@ defmodule Phoenix.LiveView.AssignAsyncTest do
       assert render_async(lv, 200) =~ "lc_data: 123"
     end
   end
+
+  describe "LiveView assign_async, supervised" do
+    setup do
+      start_supervised!({Task.Supervisor, name: TestAsyncSupervisor})
+      :ok
+    end
+
+    test "valid return", %{conn: conn} do
+      {:ok, lv, _html} = live(conn, "/assign_async?test=sup_ok")
+      assert render_async(lv) =~ "data: 123"
+    end
+
+    test "raise during execution", %{conn: conn} do
+      {:ok, lv, _html} = live(conn, "/assign_async?test=sup_raise")
+
+      assert render_async(lv) =~ "{:exit, {%RuntimeError{message: "boom"}"
+      assert render(lv)
+    end
+
+    test "exit during execution", %{conn: conn} do
+      {:ok, lv, _html} = live(conn, "/assign_async?test=sup_exit")
+
+      assert render_async(lv) =~ "{:exit, :boom}"
+      assert render(lv)
+    end
+  end
 end

test/support/live_views/general.ex

@@ -376,14 +376,26 @@ defmodule Phoenix.LiveViewTest.AssignAsyncLive do
     {:ok, assign_async(socket, :data, fn -> {:ok, %{data: 123}} end)}
   end
 
+  def mount(%{"test" => "sup_ok"}, _session, socket) do
+    {:ok, assign_async(socket, :data, fn -> {:ok, %{data: 123}} end, supervisor: TestAsyncSupervisor)}
+  end
+
   def mount(%{"test" => "raise"}, _session, socket) do
     {:ok, assign_async(socket, :data, fn -> raise("boom") end)}
   end
 
+  def mount(%{"test" => "sup_raise"}, _session, socket) do
+    {:ok, assign_async(socket, :data, fn -> raise("boom") end, supervisor: TestAsyncSupervisor)}
+  end
+
   def mount(%{"test" => "exit"}, _session, socket) do
     {:ok, assign_async(socket, :data, fn -> exit(:boom) end)}
   end
 
+  def mount(%{"test" => "sup_exit"}, _session, socket) do
+    {:ok, assign_async(socket, :data, fn -> exit(:boom) end, supervisor: TestAsyncSupervisor)}
+  end
+
   def mount(%{"test" => "lv_exit"}, _session, socket) do
     {:ok,
      assign_async(socket, :data, fn ->