Add assigns functions and assign_async

lib/live_data.ex

@@ -86,7 +86,7 @@ defmodule LiveData do
   the list change order, any individual properties change, or other changes.
   """
 
-  alias LiveData.{Socket, Tracked}
+  alias LiveData.{Socket, Tracked, Async, AsyncResult}
 
   @type rendered :: any()
 
@@ -109,9 +109,83 @@ defmodule LiveData do
   end
 
   def assign(%Socket{assigns: assigns} = socket, key, value) do
+    validate_assign_key!(key)
     assigns = Map.put(assigns, key, value)
     %{socket | assigns: assigns}
   end
 
+  def assign(%Socket{} = socket, keyword_or_map)
+      when is_map(keyword_or_map) or is_list(keyword_or_map) do
+    Enum.reduce(keyword_or_map, socket, fn {key, value}, acc ->
+      assign(acc, key, value)
+    end)
+  end
+
+  def assign_new(%Socket{} = socket, key, fun) when is_function(fun, 1) do
+    validate_assign_key!(key)
+    assigns = assign_new(socket.assigns, key, fun)
+    %Socket{socket | assigns: assigns}
+  end
+
+  def assign_new(%{} = assigns, key, fun) when is_function(fun, 1) do
+    validate_assign_key!(key)
+
+    case assigns do
+      %{^key => _} -> assigns
+      _ -> Map.put_new(assigns, key, fun.(assigns))
+    end
+  end
+
+  @doc """
+  Assigns keys asynchronously.
+
+  The task is linked to the caller and errors are wrapped.
+  Each key passed to `assign_async/3` will be assigned to
+  an `%AsyncResult{}` struct holding the status of the operation
+  and the result when completed.
+
+  ## Examples
+
+      def mount(%{"slug" => slug}, socket) do
+        {:ok,
+         socket
+         |> assign(:foo, "bar")
+         |> assign_async(:org, fn -> {:ok, %{org: fetch_org!(slug)}} end)
+         |> assign_async([:profile, :rank], fn -> {:ok, %{profile: ..., rank: ...}} end)}
+      end
+
+  See the moduledoc for more information.
+  """
+  def assign_async(%Socket{} = socket, key_or_keys, func)
+      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)
+  end
+
+  defp validate_assign_key!(key) when is_atom(key), do: :ok
+
+  defp validate_assign_key!(key) do
+    raise ArgumentError, "assigns in LiveData must be atoms, got: #{inspect(key)}"
+  end
+
+  def async_result(%AsyncResult{} = async_assign, clauses) do
+    if !Keyword.keyword?(clauses) or
+         !Enum.all?(clauses, fn {key, _} -> key in [:ok, :loading, :failed] end) do
+      raise ArgumentError,
+            "invalid clauses, expected :ok, :loading, or :failed: #{inspect(clauses)}"
+    end
+
+    cond do
+      async_assign.ok? ->
+        clauses[:ok].(async_assign.result)
+
+      async_assign.loading ->
+        clauses[:loading].()
+
+      !is_nil(async_assign.failed) ->
+        clauses[:failed].(async_assign.result)
+    end
+  end
+
   def debug_prints?, do: Application.get_env(:live_data, :deft_compiler_debug_prints, false)
 end

lib/live_data/async.ex

@@ -0,0 +1,203 @@
+defmodule LiveData.Async do
+  @moduledoc false
+
+  alias LiveData.{AsyncResult, Socket, Channel}
+
+  def start_async(%Socket{} = socket, key, func)
+      when is_atom(key) and is_function(func, 0) do
+    run_async_task(socket, key, func, :start)
+  end
+
+  def assign_async(%Socket{} = socket, key_or_keys, func)
+      when (is_atom(key_or_keys) or is_list(key_or_keys)) and
+             is_function(func, 0) do
+    keys = List.wrap(key_or_keys)
+
+    # verifies result inside task
+    wrapped_func = fn ->
+      case func.() do
+        {:ok, %{} = assigns} ->
+          if Enum.find(keys, &(not is_map_key(assigns, &1))) do
+            raise ArgumentError, """
+            expected assign_async to return map of assigns for all keys
+            in #{inspect(keys)}, but got: #{inspect(assigns)}
+            """
+          else
+            {:ok, assigns}
+          end
+
+        {:error, reason} ->
+          {:error, reason}
+
+        other ->
+          raise ArgumentError, """
+          expected assign_async to return {:ok, map} of
+          assigns for #{inspect(keys)} or {:error, reason}, got: #{inspect(other)}
+          """
+      end
+    end
+
+    new_assigns =
+      Enum.map(keys, fn key ->
+        case socket.assigns do
+          %{^key => %AsyncResult{ok?: true} = existing} ->
+            {key, AsyncResult.loading(existing, keys)}
+
+          %{} ->
+            {key, AsyncResult.loading(keys)}
+        end
+      end)
+
+    socket
+    |> LiveData.assign(new_assigns)
+    |> run_async_task(keys, wrapped_func, :assign)
+  end
+
+  defp run_async_task(%Socket{} = socket, key, func, kind) do
+    lv_pid = self()
+    cid = cid(socket)
+    {:ok, pid} = Task.start_link(fn -> do_async(lv_pid, cid, key, func, kind) end)
+
+    ref =
+      :erlang.monitor(:process, pid, alias: :reply_demonitor, tag: {__MODULE__, key, cid, kind})
+
+    send(pid, {:context, ref})
+
+    update_private_async(socket, &Map.put(&1, key, {ref, pid, kind}))
+  end
+
+  defp do_async(lv_pid, cid, key, func, async_kind) do
+    receive do
+      {:context, ref} ->
+        try do
+          result = func.()
+          Channel.report_async_result(ref, async_kind, ref, cid, key, {:ok, result})
+        catch
+          catch_kind, reason ->
+            Process.unlink(lv_pid)
+            caught_result = to_exit(catch_kind, reason, __STACKTRACE__)
+            Channel.report_async_result(ref, async_kind, ref, cid, key, caught_result)
+            :erlang.raise(catch_kind, reason, __STACKTRACE__)
+        end
+    end
+  end
+
+  def cancel_async(%Socket{} = socket, %AsyncResult{} = result, reason) do
+    case result do
+      %AsyncResult{loading: keys} when is_list(keys) ->
+        new_assigns = for key <- keys, do: {key, AsyncResult.failed(result, {:exit, reason})}
+
+        socket
+        |> LiveData.assign(new_assigns)
+        |> cancel_async(keys, reason)
+
+      %AsyncResult{} ->
+        socket
+    end
+  end
+
+  def cancel_async(%Socket{} = socket, key, reason) do
+    case get_private_async(socket, key) do
+      {_ref, pid, _kind} when is_pid(pid) ->
+        Process.unlink(pid)
+        Process.exit(pid, reason)
+        update_private_async(socket, &Map.delete(&1, key))
+
+      nil ->
+        socket
+    end
+  end
+
+  def handle_async(socket, maybe_component, kind, key, ref, result) do
+    case prune_current_async(socket, key, ref) do
+      {:ok, pruned_socket} ->
+        handle_kind(pruned_socket, maybe_component, kind, key, result)
+
+      :error ->
+        socket
+    end
+  end
+
+  def handle_trap_exit(socket, maybe_component, kind, key, ref, reason) do
+    handle_async(socket, maybe_component, kind, key, ref, {:exit, reason})
+  end
+
+  defp handle_kind(socket, maybe_component, :start, key, result) do
+    callback_mod = maybe_component || socket.view
+
+    case callback_mod.handle_async(key, result, socket) do
+      {:noreply, %Socket{} = new_socket} ->
+        new_socket
+
+      other ->
+        raise ArgumentError, """
+        expected #{inspect(callback_mod)}.handle_async/3 to return {:noreply, socket}, got:
+
+            #{inspect(other)}
+        """
+    end
+  end
+
+  defp handle_kind(socket, _maybe_component, :assign, keys, result) do
+    case result do
+      {:ok, {:ok, %{} = assigns}} ->
+        new_assigns =
+          for {key, val} <- assigns do
+            {key, AsyncResult.ok(get_current_async!(socket, key), val)}
+          end
+
+        LiveData.assign(socket, new_assigns)
+
+      {:ok, {:error, reason}} ->
+        new_assigns =
+          for key <- keys do
+            {key, AsyncResult.failed(get_current_async!(socket, key), {:error, reason})}
+          end
+
+        LiveData.assign(socket, new_assigns)
+
+      {:exit, _reason} = normalized_exit ->
+        new_assigns =
+          for key <- keys do
+            {key, AsyncResult.failed(get_current_async!(socket, key), normalized_exit)}
+          end
+
+        LiveData.assign(socket, new_assigns)
+    end
+  end
+
+  # handle race of async being canceled and then reassigned
+  defp prune_current_async(socket, key, ref) do
+    case get_private_async(socket, key) do
+      {^ref, _pid, _kind} -> {:ok, update_private_async(socket, &Map.delete(&1, key))}
+      {_ref, _pid, _kind} -> :error
+      nil -> :error
+    end
+  end
+
+  defp update_private_async(%{private: private} = socket, func) do
+    existing = Map.get(private, :live_async, %{})
+    %{socket | private: Map.put(private, :live_async, func.(existing))}
+  end
+
+  defp get_private_async(%Socket{} = socket, key) do
+    socket.private[:live_async][key]
+  end
+
+  defp get_current_async!(socket, key) do
+    # handle case where assign is temporary and needs to be rebuilt
+    case socket.assigns do
+      %{^key => %AsyncResult{} = current_async} -> current_async
+      %{^key => _other} -> AsyncResult.loading(key)
+      %{} -> raise ArgumentError, "missing async assign #{inspect(key)}"
+    end
+  end
+
+  defp to_exit(:throw, reason, stack), do: {:exit, {{:nocatch, reason}, stack}}
+  defp to_exit(:error, reason, stack), do: {:exit, {reason, stack}}
+  defp to_exit(:exit, reason, _stack), do: {:exit, reason}
+
+  defp cid(%Socket{} = socket) do
+    if myself = socket.assigns[:myself], do: myself.cid
+  end
+end

lib/live_data/async_result.ex

@@ -0,0 +1,78 @@
+defmodule LiveData.AsyncResult do
+  @moduledoc ~S'''
+  Provides a data structure for tracking the state of an async assign.
+
+  See the `Async Operations` section of the `LiveData` docs for more information.
+
+  ## Fields
+
+    * `:ok?` - When true, indicates the `:result` has been set successfully at least once.
+    * `:loading` - The current loading state
+    * `:failed` - The current failed state
+    * `:result` - The successful result of the async task
+  '''
+
+  defstruct ok?: false,
+            loading: nil,
+            failed: nil,
+            result: nil
+
+  alias LiveData.AsyncResult
+
+  @doc """
+  Updates the loading state.
+
+  When loading, the failed state will be reset to `nil`.
+
+  ## Examples
+
+      AsyncResult.loading()
+      AsyncResult.loading(my_async)
+      AsyncResult.loading(my_async, %{my: :loading_state})
+  """
+  def loading do
+    %AsyncResult{loading: true}
+  end
+
+  def loading(%AsyncResult{} = result) do
+    %AsyncResult{result | loading: true, failed: nil}
+  end
+
+  def loading(loading_state) do
+    %AsyncResult{loading: loading_state, failed: nil}
+  end
+
+  def loading(%AsyncResult{} = result, loading_state) do
+    %AsyncResult{result | loading: loading_state, failed: nil}
+  end
+
+  @doc """
+  Updates the failed state.
+
+  When failed, the loading state will be reset to `nil`.
+
+  ## Examples
+
+      AsyncResult.failed(my_async, {:exit, :boom})
+      AsyncResult.failed(my_async, {:error, reason})
+  """
+  def failed(%AsyncResult{} = result, reason) do
+    %AsyncResult{result | failed: reason, loading: nil}
+  end
+
+  @doc """
+  Updates the successful result.
+
+  The `:ok?` field will also be set to `true` to indicate this result has
+  completed successfully at least once, regardless of future state changes.
+
+  When ok'd, the loading and failed state will be reset to `nil`.
+
+  ## Examples
+
+      AsyncResult.ok(my_async, my_result)
+  """
+  def ok(%AsyncResult{} = result, value) do
+    %AsyncResult{result | failed: nil, loading: nil, ok?: true, result: value}
+  end
+end