WIP

lib/phoenix_live_view/application.ex

@@ -6,6 +6,11 @@ defmodule Phoenix.LiveView.Application do
   @impl true
   def start(_type, _args) do
     Phoenix.LiveView.Logger.install()
-    Supervisor.start_link([], strategy: :one_for_one, name: Phoenix.LiveView.Supervisor)
+
+    children = [
+      {Task.Supervisor, name: Phoenix.LiveView.TaskSup}
+    ]
+
+    Supervisor.start_link(children, strategy: :one_for_one, name: Phoenix.LiveView.Supervisor)
   end
 end

lib/phoenix_live_view/async_assign.ex

@@ -0,0 +1,185 @@
+defmodule Phoenix.LiveView.AsyncAssign do
+  @moduledoc ~S'''
+  Adds async_assign functionality to LiveViews.
+
+  ## Examples
+
+  defmodule MyLive do
+
+    def render(assigns) do
+
+      ~H"""
+      <div :if={@async.org.loading?}>Loading organization...</div>
+      <div :if={@async.org.result == nil}}>You don't have an org yet</div>
+      <div :if={@async.org.error}>there was an error loading the organization</div>
+
+      <.async_result let={org} item={@async.org}>
+        <:loading>Loading organization...</:loading>
+        <:empty>You don't have an organization yet</:error>
+        <:error>there was an error loading the organization</:error>
+        <%= org.name %>
+      <.async_result>
+
+      <div :for={orgs <- @async.orgs}>...</div>
+      """
+    end
+
+    def mount(%{"slug" => slug}, _, socket) do
+      {:ok,
+       socket
+       |> assign(:foo, "bar)
+       |> assign_async(:org, fn -> {:ok, %{org: fetch_org!(slug)} end)
+       |> assign_async(:profile, [:profile, :rank], fn -> {:ok, %{profile: ..., rank: ...}} end)}
+    end
+  end
+  '''
+
+  defstruct name: nil,
+            ref: nil,
+            pid: nil,
+            keys: [],
+            loading?: false,
+            error: nil,
+            result: nil,
+            canceled?: false
+
+  alias Phoenix.LiveView.AsyncAssign
+
+  @doc """
+  TODO
+  """
+  def assign_async(%Phoenix.LiveView.Socket{} = socket, name, func) do
+    assign_async(socket, name, [name], func)
+  end
+
+  def assign_async(%Phoenix.LiveView.Socket{} = socket, name, keys, func)
+      when is_atom(name) and is_list(keys) and is_function(func, 0) do
+    socket = cancel_existing(socket, name)
+    base_async = %AsyncAssign{name: name, keys: keys, loading?: true}
+
+    async_assign =
+      if Phoenix.LiveView.connected?(socket) do
+        lv_pid = self()
+        ref = make_ref()
+        cid = if myself = socket.assigns[:myself], do: myself.cid
+
+        {:ok, pid} =
+          Task.start_link(fn ->
+            do_async(lv_pid, cid, %AsyncAssign{base_async | pid: self(), ref: ref}, func)
+          end)
+
+        %AsyncAssign{base_async | pid: pid, ref: ref}
+      else
+        base_async
+      end
+
+    Enum.reduce(keys, socket, fn key, acc -> update_async(acc, key, async_assign) end)
+  end
+
+  @doc """
+  TODO
+  """
+  def cancel_async(%Phoenix.LiveView.Socket{} = socket, name) do
+    case get(socket, name) do
+      %AsyncAssign{loading?: false} ->
+        socket
+
+      %AsyncAssign{canceled?: false, pid: pid} = existing ->
+        Process.unlink(pid)
+        Process.exit(pid, :kill)
+        async = %AsyncAssign{existing | loading?: false, error: nil, canceled?: true}
+        update_async(socket, name, async)
+
+      nil ->
+        raise ArgumentError,
+              "no async assign #{inspect(name)} previously assigned with assign_async"
+    end
+  end
+
+  defp do_async(lv_pid, cid, %AsyncAssign{} = async_assign, func) do
+    %AsyncAssign{keys: known_keys, name: name, ref: ref} = async_assign
+
+    try do
+      case func.() do
+        {:ok, %{} = results} ->
+          if Map.keys(results) -- known_keys == [] do
+            Phoenix.LiveView.Channel.write_socket(lv_pid, cid, fn socket ->
+              write_current_async(socket, name, ref, fn %AsyncAssign{} = current ->
+                Enum.reduce(results, socket, fn {key, result}, acc ->
+                  async = %AsyncAssign{current | result: result, loading?: false}
+                  update_async(acc, key, async)
+                end)
+              end)
+            end)
+          else
+            raise ArgumentError, """
+            expected assign_async to return map of
+            assigns for all keys in #{inspect(known_keys)}, but got: #{inspect(results)}
+            """
+          end
+
+        {:error, reason} ->
+          Phoenix.LiveView.Channel.write_socket(lv_pid, cid, fn socket ->
+            write_current_async(socket, name, ref, fn %AsyncAssign{} = current ->
+              Enum.reduce(known_keys, socket, fn key, acc ->
+                async = %AsyncAssign{current | result: nil, loading?: false, error: reason}
+                update_async(acc, key, async)
+              end)
+            end)
+          end)
+
+        other ->
+          raise ArgumentError, """
+          expected assign_async to return {:ok, map} of
+          assigns for #{inspect(known_keys)} or {:error, reason}, got: #{inspect(other)}
+          """
+      end
+    catch
+      kind, reason ->
+        Process.unlink(lv_pid)
+
+        Phoenix.LiveView.Channel.write_socket(lv_pid, cid, fn socket ->
+          write_current_async(socket, name, ref, fn %AsyncAssign{} = current ->
+            Enum.reduce(known_keys, socket, fn key, acc ->
+              async = %AsyncAssign{current | loading?: false, error: {kind, reason}}
+              update_async(acc, key, async)
+            end)
+          end)
+        end)
+
+        :erlang.raise(kind, reason, __STACKTRACE__)
+    end
+  end
+
+  defp update_async(socket, key, %AsyncAssign{} = new_async) do
+    socket
+    |> ensure_async()
+    |> Phoenix.Component.update(:async, fn async_map ->
+      Map.put(async_map, key, new_async)
+    end)
+  end
+
+  defp ensure_async(socket) do
+    Phoenix.Component.assign_new(socket, :async, fn -> %{} end)
+  end
+
+  defp get(%Phoenix.LiveView.Socket{} = socket, name) do
+    socket.assigns[:async] && Map.get(socket.assigns.async, name)
+  end
+
+  # handle race of async being canceled and then reassigned
+  defp write_current_async(socket, name, ref, func) do
+    case get(socket, name) do
+      %AsyncAssign{ref: ^ref} = async_assign -> func.(async_assign)
+      %AsyncAssign{ref: _ref} -> socket
+    end
+  end
+
+  defp cancel_existing(socket, name) do
+    if get(socket, name) do
+      cancel_async(socket, name)
+    else
+      socket
+    end
+  end
+end

lib/phoenix_live_view/channel.ex

@@ -30,6 +30,14 @@ defmodule Phoenix.LiveView.Channel do
     )
   end
 
+  def write_socket(lv_pid, cid, func) when is_function(func, 1) do
+    GenServer.call(lv_pid, {@prefix, :write_socket, cid, func})
+  end
+
+  def async_pids(lv_pid) do
+    GenServer.call(lv_pid, {@prefix, :async_pids})
+  end
+
   def ping(pid) do
     GenServer.call(pid, {@prefix, :ping}, :infinity)
   end
@@ -282,6 +290,31 @@ defmodule Phoenix.LiveView.Channel do
     {:reply, :ok, state}
   end
 
+  def handle_call({@prefix, :async_pids}, _from, state) do
+    %{socket: socket} = state
+    lv_pids = get_async_pids(socket.assigns)
+
+    component_pids =
+      state
+      |> component_assigns()
+      |> Enum.flat_map(fn
+        {_cid, %{async: %{}} = assigns} -> get_async_pids(assigns)
+        {_cid, %{}} -> []
+      end)
+
+    {:reply, {:ok, lv_pids ++ component_pids}, state}
+  end
+
+  def handle_call({@prefix, :write_socket, cid, func}, _from, state) do
+    new_state =
+      write_socket(state, cid, nil, fn socket, _ ->
+        %Phoenix.LiveView.Socket{} = new_socket = func.(socket)
+        {new_socket, {:ok, nil, state}}
+      end)
+
+    {:reply, :ok, new_state}
+  end
+
   def handle_call({@prefix, :fetch_upload_config, name, cid}, _from, state) do
     read_socket(state, cid, fn socket, _ ->
       result =
@@ -1391,4 +1424,16 @@ defmodule Phoenix.LiveView.Channel do
   end
 
   defp maybe_subscribe_to_live_reload(response), do: response
+
+  defp component_assigns(state) do
+    %{components: {components, _ids, _}} = state
+
+    Enum.into(components, %{}, fn {cid, {_mod, _id, assigns, _private, _prints}} ->
+      {cid, assigns}
+    end)
+  end
+
+  defp get_async_pids(assigns) do
+    Enum.map(assigns[:async] || %{}, fn {_, %Phoenix.LiveView.AsyncAssign{pid: pid}} -> pid end)
+  end
 end

lib/phoenix_live_view/test/client_proxy.ex

@@ -526,6 +526,12 @@ defmodule Phoenix.LiveViewTest.ClientProxy do
     {:noreply, state}
   end
 
+  def handle_call({:async_pids, topic_or_element}, _from, state) do
+    topic = proxy_topic(topic_or_element)
+    %{pid: pid} = fetch_view_by_topic!(state, topic)
+    {:reply, Phoenix.LiveView.Channel.async_pids(pid), state}
+  end
+
   def handle_call({:render_event, topic_or_element, type, value}, from, state) do
     topic = proxy_topic(topic_or_element)
     %{pid: pid} = fetch_view_by_topic!(state, topic)

lib/phoenix_live_view/test/live_view_test.ex

@@ -468,8 +468,7 @@ defmodule Phoenix.LiveViewTest do
   def __render_component__(endpoint, %{module: component}, assigns, opts) do
     socket = %Socket{endpoint: endpoint, router: opts[:router]}
 
-    assigns =
-      Map.new(assigns)
+    assigns = Map.new(assigns)
 
     # TODO: Make the ID required once we support only stateful module components as live_component
     mount_assigns = if assigns[:id], do: %{myself: %Phoenix.LiveComponent.CID{cid: -1}}, else: %{}
@@ -922,6 +921,33 @@ defmodule Phoenix.LiveViewTest do
     call(view, {:render_event, {proxy_topic(view), to_string(event), view.target}, type, value})
   end
 
+  @doc """
+  TODO
+  """
+  def await_async(view_or_element, timeout \\ 100) do
+    pids =
+      case view_or_element do
+        %View{} = view -> call(view, {:async_pids, {proxy_topic(view), nil, nil}})
+        %Element{} = element -> call(element, {:async_pids, element})
+      end
+
+    task =
+      Task.async(fn ->
+        pids
+        |> Enum.map(&Process.monitor(&1))
+        |> Enum.each(fn ref ->
+          receive do
+            {:DOWN, ^ref, :process, _pid, _reason} -> :ok
+          end
+        end)
+      end)
+
+    case Task.yield(task, timeout) || Task.ignore(task) do
+      {:ok, _} -> :ok
+      nil -> raise RuntimeError, "expected async processes to finish within #{timeout}ms"
+    end
+  end
+
   @doc """
   Simulates a `live_patch` to the given `path` and returns the rendered result.
   """

test/phoenix_live_view/integrations/assign_async_test.exs

@@ -0,0 +1,48 @@
+defmodule Phoenix.LiveView.AssignAsyncTest do
+  use ExUnit.Case, async: true
+  import Phoenix.ConnTest
+
+  import Phoenix.LiveViewTest
+  alias Phoenix.LiveViewTest.Endpoint
+
+  @endpoint Endpoint
+
+  setup do
+    {:ok, conn: Plug.Test.init_test_session(Phoenix.ConnTest.build_conn(), %{})}
+  end
+
+  describe "assign_async" do
+    test "bad return", %{conn: conn} do
+      {:ok, lv, _html} = live(conn, "/async?test=bad_return")
+
+      await_async(lv)
+      assert render(lv) =~
+               "error: {:error, %ArgumentError{message: "expected assign_async to return {:ok, map} of\\nassigns for [:data] or {:error, reason}, got: 123\\n"}}"
+
+      assert render(lv)
+    end
+
+    test "missing known key", %{conn: conn} do
+      {:ok, lv, _html} = live(conn, "/async?test=bad_ok")
+
+      await_async(lv)
+      assert render(lv) =~
+               "expected assign_async to return map of\\nassigns for all keys in [:data]"
+
+      assert render(lv)
+    end
+
+    test "valid return", %{conn: conn} do
+      {:ok, lv, _html} = live(conn, "/async?test=ok")
+      await_async(lv)
+      assert render(lv) =~ "data: 123"
+    end
+
+    test "raise during execution", %{conn: conn} do
+      {:ok, lv, _html} = live(conn, "/async?test=raise")
+
+      await_async(lv)
+      assert render(lv) =~ "error: {:error, %RuntimeError{message: "boom"}}"
+    end
+  end
+end