PubSub convenience functions

Author: tfwright

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.PubSub.broadcast(session.id, {:job, %{pid: self(), progress: i/count, label: "Regenerating passwords"}})
+      LiveAdmin.PubSub.update_job(session.id, self(), progress: i/count, label: "Regenerating passwords")
     end)
 
     {:ok, "updated"}

lib/live_admin/components/container.ex

@@ -114,45 +114,38 @@ defmodule LiveAdmin.Components.Container do
         try do
           case apply(m, f, [Resource.query(resource, search, config) | args]) do
             {:ok, message} ->
-              LiveAdmin.PubSub.broadcast(
+              LiveAdmin.PubSub.announce(
                 session.id,
-                {:announce,
-                 %{
-                   message:
-                     trans("Task %{name} succeeded: '%{message}'",
-                       inter: [name: name, message: message]
-                     )
-                 }, type: :success}
+                :success,
+                trans("Task %{name} succeeded: '%{message}'",
+                  inter: [name: name, message: message]
+                )
               )
 
             {:error, message} ->
-              LiveAdmin.PubSub.broadcast(
+              LiveAdmin.PubSub.announce(
                 session.id,
-                {:announce,
-                 %{
-                   message:
-                     trans("Task %{name} failed: '%{message}'",
-                       inter: [name: name, message: message]
-                     ),
-                   type: :error
-                 }}
+                :error,
+                trans("Task %{name} failed: '%{message}'",
+                  inter: [name: name, message: message]
+                )
               )
           end
         rescue
           error ->
             Logger.error(inspect(error))
 
-            LiveAdmin.PubSub.broadcast(
+            LiveAdmin.PubSub.announce(
               session.id,
-              {:announce,
-               %{message: trans("Task %{name} failed", inter: [name: name]), type: :error}}
+              :error,
+              trans("Task %{name} failed", inter: [name: name])
             )
         after
-          LiveAdmin.PubSub.broadcast(session.id, {:job, %{pid: self(), progress: 1}})
+          LiveAdmin.PubSub.update_job(session.id, self(), progress: 1)
         end
       end)
 
-    LiveAdmin.PubSub.broadcast(session.id, {:job, %{pid: task.pid, progress: 0, label: name}})
+    LiveAdmin.PubSub.update_job(session.id, task.pid, progress: 0, label: name)
 
     {:noreply, push_navigate(socket, to: route_with_params(socket.assigns))}
   end

lib/live_admin/components/nav/jobs.ex

@@ -78,7 +78,11 @@ defmodule LiveAdmin.Components.Nav.Jobs do
   end
 
   @impl true
-  def handle_info(_, socket), do: {:noreply, socket}
+  def handle_info(data, socket) do
+    Logger.warning("Unhandled broadcast: #{inspect(data)}")
+
+    {:noreply, socket}
+  end
 
   @impl true
   def render(assigns) do

lib/live_admin/components/resource/index.ex

@@ -586,49 +586,50 @@ defmodule LiveAdmin.Components.Container.Index do
       fn ->
         pid = self()
 
-        LiveAdmin.PubSub.broadcast(session.id, {:job, %{pid: pid, progress: 0, label: label}})
+        LiveAdmin.PubSub.update_job(session.id, pid, progress: 0, label: label)
 
         records = Resource.all(ids, resource, prefix, repo)
 
-        {type, message} =
-          records
-          |> Enum.with_index()
-          |> Enum.reduce(0, fn {record, i}, failed_count ->
-            try do
-              case apply(mod, func, [record | args]) do
-                {:ok, _} -> failed_count
-                {:error, _} -> failed_count + 1
-              end
-            rescue
-              _ -> failed_count + 1
-            after
-              LiveAdmin.PubSub.broadcast(
-                session.id,
-                {:job, %{pid: pid, progress: (i + 1) / length(records)}}
-              )
+        records
+        |> Enum.with_index()
+        |> Enum.reduce(0, fn {record, i}, failed_count ->
+          try do
+            case apply(mod, func, [record | args]) do
+              {:ok, _} -> failed_count
+              {:error, _} -> failed_count + 1
             end
-          end)
-          |> case do
-            0 ->
-              {:success,
-               trans("%{name} action run successfully on %{count} records",
-                 inter: [name: name, count: length(records)]
-               )}
-
-            error_count ->
-              {:error,
-               trans(
-                 "%{name} action failed on %{error_count} records (%{success_count} succeeeded)",
-                 inter: [
-                   name: name,
-                   error_count: error_count,
-                   success_count: length(records) - error_count
-                 ]
-               )}
+          rescue
+            _ -> failed_count + 1
+          after
+            LiveAdmin.PubSub.update_job(session.id, pid, progress: (i + 1) / length(records))
           end
+        end)
+        |> case do
+          0 ->
+            LiveAdmin.PubSub.announce(
+              session.id,
+              :success,
+              trans("%{name} action run successfully on %{count} records",
+                inter: [name: name, count: length(records)]
+              )
+            )
+
+          error_count ->
+            LiveAdmin.PubSub.announce(
+              session.id,
+              :error,
+              trans(
+                "%{name} action failed on %{error_count} records (%{success_count} succeeeded)",
+                inter: [
+                  name: name,
+                  error_count: error_count,
+                  success_count: length(records) - error_count
+                ]
+              )
+            )
+        end
 
-        LiveAdmin.PubSub.broadcast(session.id, {:job, %{pid: pid, progress: 1}})
-        LiveAdmin.PubSub.broadcast(session.id, {:announce, %{message: message, type: type}})
+        LiveAdmin.PubSub.update_job(session.id, pid, progress: 1)
       end,
       timeout: :infinity
     )

lib/live_admin/pub_sub.ex

@@ -10,6 +10,9 @@ defmodule LiveAdmin.PubSub do
   """
 
   @type session_id :: String.t()
+  @type status :: :error | :success | :info
+  @type message :: String.t()
+  @type data :: Keyword.t()
 
   @spec broadcast(session_id, {atom(), map()}) :: :ok
   @spec broadcast({atom(), map()}) :: :ok
@@ -22,8 +25,24 @@ defmodule LiveAdmin.PubSub do
       if(session_id, do: "session:#{session_id}", else: "all"),
       event
     )
+  end
 
-    :ok
+  @spec announce(session_id, status, message) :: :ok
+  @spec announce(status, message) :: :ok
+  @doc """
+    Add a message with status to alerts
+  """
+  def announce(session_id \\ nil, status, message) do
+    broadcast(session_id, {:announce, %{message: message, type: status}})
+  end
+
+  @spec update_job(session_id, pid, data) :: :ok
+  @spec update_job(pid, data) :: :ok
+  @doc """
+    Update job progress
+  """
+  def update_job(session_id \\ nil, pid, data) do
+    broadcast(session_id, {:job, Enum.into(data, %{pid: pid})})
   end
 
   @spec subscribe(session_id) :: :ok