Refactor notifications API

README.md

@@ -14,7 +14,7 @@ Significant features:
 * Overridable views, styles, and API
 * Custom actions at the resource and record level, with support for dynamic inputs
 * Edit (nested) embedded schemas
-* Notifications
+* Notifications via [Phoenix.PubSub](https://github.com/phoenixframework/phoenix_pubsub)
 * i18n via [Gettext](https://github.com/elixir-gettext/gettext)
 
 See for yourself, try out the [demo app](#development)
@@ -154,7 +154,7 @@ A dropdown will be added to the top nav bar that will allow you to switch betwee
 
 ### Notifications
 
-`LiveAdmin.Notifier` provides an API for triggering events from LiveAdmin.Resource modules, or other application code.
+LiveAdmin broadcasts notifications notifications about certain events and responds with UI events.
 
 Currently supported:
 

README.md.eex

@@ -14,7 +14,7 @@ Significant features:
 * Overridable views, styles, and API
 * Custom actions at the resource and record level, with support for dynamic inputs
 * Edit (nested) embedded schemas
-* Notifications
+* Notifications via [Phoenix.PubSub](https://github.com/phoenixframework/phoenix_pubsub)
 * i18n via [Gettext](https://github.com/elixir-gettext/gettext)
 
 See for yourself, try out the [demo app](#development)
@@ -154,7 +154,7 @@ A dropdown will be added to the top nav bar that will allow you to switch betwee
 
 ### Notifications
 
-`LiveAdmin.Notifier` provides an API for triggering events from LiveAdmin.Resource modules, or other application code.
+LiveAdmin broadcasts notifications notifications about certain events and responds with UI events.
 
 Currently supported:
 

dev/user_admin.ex

@@ -72,7 +72,7 @@ defmodule DemoWeb.UserAdmin do
       |> Ecto.Changeset.change(encrypted_password: :crypto.strong_rand_bytes(16) |> Base.encode16())
       |> Demo.Repo.update()
 
-      LiveAdmin.Notifier.job(session, self(), i/count)
+      LiveAdmin.PubSub.broadcast(session.id, {:job, %{pid: self(), progress: i/count, label: "Regenerating passwords"}})
     end)
 
     {:ok, "updated"}

lib/live_admin/components/container.ex

@@ -2,6 +2,8 @@ defmodule LiveAdmin.Components.Container do
   use Phoenix.LiveView
   use PhoenixHTMLHelpers
 
+  require Logger
+
   import LiveAdmin,
     only: [
       resource_title: 2,
@@ -18,15 +20,12 @@ defmodule LiveAdmin.Components.Container do
   alias Phoenix.LiveView.JS
 
   @impl true
-  def mount(_params, %{"session_id" => session_id}, socket) do
+  def mount(_params, _session, socket) do
     socket =
       assign(socket, loading: !connected?(socket), jobs: [])
 
     if connected?(socket) do
       Process.send_after(self(), :clear_flash, 1000)
-
-      :ok = Phoenix.PubSub.subscribe(LiveAdmin.PubSub, "session:#{session_id}")
-      :ok = Phoenix.PubSub.subscribe(LiveAdmin.PubSub, "all")
     end
 
     {:ok, socket}
@@ -115,34 +114,45 @@ defmodule LiveAdmin.Components.Container do
         try do
           case apply(m, f, [Resource.query(resource, search, config) | args]) do
             {:ok, message} ->
-              LiveAdmin.Notifier.announce(
-                session,
-                trans("Task %{name} succeeded: '%{message}'",
-                  inter: [name: name, message: message]
-                ),
-                type: :success
+              LiveAdmin.PubSub.broadcast(
+                session.id,
+                {:announce,
+                 %{
+                   message:
+                     trans("Task %{name} succeeded: '%{message}'",
+                       inter: [name: name, message: message]
+                     )
+                 }, type: :success}
               )
 
             {:error, message} ->
-              LiveAdmin.Notifier.announce(
-                session,
-                trans("Task %{name} failed: '%{message}'", inter: [name: name, message: message]),
-                type: :error
+              LiveAdmin.PubSub.broadcast(
+                session.id,
+                {:announce,
+                 %{
+                   message:
+                     trans("Task %{name} failed: '%{message}'",
+                       inter: [name: name, message: message]
+                     ),
+                   type: :error
+                 }}
               )
           end
         rescue
-          _ ->
-            LiveAdmin.Notifier.announce(
-              session,
-              trans("Task %{name} failed", inter: [name: name]),
-              type: :error
+          error ->
+            Logger.error(inspect(error))
+
+            LiveAdmin.PubSub.broadcast(
+              session.id,
+              {:announce,
+               %{message: trans("Task %{name} failed", inter: [name: name]), type: :error}}
             )
         after
-          LiveAdmin.Notifier.job(session, self(), 1)
+          LiveAdmin.PubSub.broadcast(session.id, {:job, %{pid: self(), progress: 1}})
         end
       end)
 
-    LiveAdmin.Notifier.job(session, task.pid, 0, label: name)
+    LiveAdmin.PubSub.broadcast(session.id, {:job, %{pid: task.pid, progress: 0, label: "name"}})
 
     {:noreply, push_redirect(socket, to: route_with_params(socket.assigns))}
   end

lib/live_admin/components/nav/jobs.ex

@@ -6,8 +6,8 @@ defmodule LiveAdmin.Components.Nav.Jobs do
   @impl true
   def mount(_, %{"session_id" => session_id}, socket) do
     if connected?(socket) do
-      :ok = Phoenix.PubSub.subscribe(LiveAdmin.PubSub, "session:#{session_id}")
-      :ok = Phoenix.PubSub.subscribe(LiveAdmin.PubSub, "all")
+      :ok = LiveAdmin.PubSub.subscribe(session_id)
+      :ok = LiveAdmin.PubSub.subscribe()
     end
 
     {:ok, assign(socket, jobs: []), layout: false}
@@ -26,34 +26,40 @@ defmodule LiveAdmin.Components.Nav.Jobs do
   end
 
   @impl true
-  def handle_info(info = {:announce, message, type}, socket) do
+  def handle_info(info = {:announce, %{message: message, type: type}}, socket) do
     Logger.debug("ANNOUNCE: #{inspect(info)}")
     {:noreply, push_event(socket, type, %{msg: message})}
   end
 
   @impl true
-  def handle_info({:job, pid, :start, label}, socket) do
-    Process.monitor(pid)
-
-    socket = update(socket, :jobs, fn jobs -> [{pid, label, 0.0} | jobs] end)
+  def handle_info({:job, %{pid: pid, progress: progress}}, socket)
+      when progress >= 1 do
+    Process.send_after(self(), {:remove_job, pid}, 1500)
 
-    {:noreply, socket}
+    {:noreply, set_progress(socket, pid, 1.0)}
   end
 
   @impl true
-  def handle_info({:job, pid, :progress, progress}, socket) do
-    socket = set_progress(socket, pid, progress)
+  def handle_info({:job, job = %{pid: pid, progress: progress}}, socket) do
+    socket =
+      update(socket, :jobs, fn jobs ->
+        jobs
+        |> Enum.find_index(fn {job_pid, _, _} -> job_pid == pid end)
+        |> case do
+          nil ->
+            Process.monitor(pid)
+            [{pid, Map.fetch!(job, :label), 0.0} | jobs]
+
+          i ->
+            List.update_at(jobs, i, fn {pid, job_label, _} ->
+              {pid, Map.get(job, :label, job_label), progress}
+            end)
+        end
+      end)
 
     {:noreply, socket}
   end
 
-  @impl true
-  def handle_info({:job, pid, :complete}, socket) do
-    Process.send_after(self(), {:remove_job, pid}, 1500)
-
-    {:noreply, set_progress(socket, pid, 1.0)}
-  end
-
   @impl true
   def handle_info({:remove_job, pid}, socket) do
     socket =

lib/live_admin/components/resource/index.ex

@@ -426,7 +426,7 @@ defmodule LiveAdmin.Components.Container.Index do
       fn ->
         pid = self()
 
-        LiveAdmin.Notifier.job(session, pid, 0, label: label)
+        LiveAdmin.PubSub.broadcast(session.id, {:job, %{pid: pid, progress: 0, label: label}})
 
         records = Resource.all(ids, resource, prefix, repo)
 
@@ -442,7 +442,10 @@ defmodule LiveAdmin.Components.Container.Index do
             rescue
               _ -> failed_count + 1
             after
-              LiveAdmin.Notifier.job(session, pid, (i + 1) / length(records))
+              LiveAdmin.PubSub.broadcast(
+                session.id,
+                {:job, %{pid: pid, progress: (i + 1) / length(records)}}
+              )
             end
           end)
           |> case do
@@ -464,8 +467,8 @@ defmodule LiveAdmin.Components.Container.Index do
                )}
           end
 
-        LiveAdmin.Notifier.job(session, pid, 1)
-        LiveAdmin.Notifier.announce(session, message, type: type)
+        LiveAdmin.PubSub.broadcast(session.id, {:job, %{pid: pid, progress: 1}})
+        LiveAdmin.PubSub.broadcast(session.id, {:announce, %{message: message, type: type}})
       end,
       timeout: :infinity
     )

lib/live_admin/pub_sub.ex

@@ -0,0 +1,38 @@
+defmodule LiveAdmin.PubSub do
+  @moduledoc """
+    PubSub system for LiveAdmin events.
+
+    Currently LiveAdmin has built in support for the following events:
+      * :job - Shows progress bar indicating status of any ongoing process. LiveAdmin uses this internally to indicate progress of actions on multiple records and tasks. Metadata consists of `pid`, `progress` float/int, and `label` string.
+      * :announce - Show temporary message with severity level. Meta consists of `message` string, and `type` (`:error`, `:success`, or `:info`)
+  """
+
+  @type session_id :: String.t()
+
+  @spec broadcast(session_id, {atom(), map()}) :: :ok
+  @spec broadcast({atom(), map()}) :: :ok
+  @doc """
+    Notify LiveAdmin of event, consisting of a unique name (either global or scoped to the session, if that is passed) and metadata.
+  """
+  def broadcast(session_id \\ nil, event) do
+    Phoenix.PubSub.broadcast(
+      __MODULE__,
+      if(session_id, do: "session:#{session_id}", else: "all"),
+      event
+    )
+
+    :ok
+  end
+
+  @spec subscribe(session_id) :: :ok
+  @spec subscribe() :: :ok
+  @doc """
+    Subscribe to LiveAdmin events for a specific session.
+  """
+  def subscribe(session_id), do: Phoenix.PubSub.subscribe(__MODULE__, "session:#{session_id}")
+
+  @doc """
+    Subscribe to *all* LiveAdmin events.
+  """
+  def subscribe(), do: Phoenix.PubSub.subscribe(__MODULE__, "all")
+end