Close TD-14 replace polling with messaging

lib/bds/cli_sync.ex

@@ -54,6 +54,11 @@ defmodule BDS.CliSync do
      end)}
   end
 
+  def data_version do
+    %{rows: [[version]]} = Repo.query!("PRAGMA data_version", [])
+    version
+  end
+
   def prune_notifications(now \\ Persistence.now_ms()) when is_integer(now) do
     {processed_count, _} =
       Repo.delete_all(

lib/bds/cli_sync/watcher.ex

@@ -29,7 +29,12 @@ defmodule BDS.CliSync.Watcher do
           Keyword.get(opts, :poll_interval_ms),
           @default_poll_interval_ms
         ),
-      pubsub: Keyword.get(opts, :pubsub, BDS.PubSub)
+      pubsub: Keyword.get(opts, :pubsub, BDS.PubSub),
+      data_version_reader: Keyword.get(opts, :data_version_reader, &CliSync.data_version/0),
+      notification_fetcher:
+        Keyword.get(opts, :notification_fetcher, &CliSync.db_file_change_detected/0),
+      pruner: Keyword.get(opts, :pruner, &CliSync.prune_notifications/0),
+      last_data_version: nil
     }
 
     {:ok, schedule_poll(state)}
@@ -49,18 +54,24 @@ defmodule BDS.CliSync.Watcher do
   end
 
   defp process_notifications(state) do
-    {:ok, notifications} = CliSync.db_file_change_detected()
-    {:ok, _pruned} = CliSync.prune_notifications()
+    current_data_version = state.data_version_reader.()
 
-    Enum.each(notifications, fn notification ->
-      Phoenix.PubSub.broadcast(
-        state.pubsub,
-        topic(),
-        {:entity_changed, notification_payload(notification)}
-      )
-    end)
+    if state.last_data_version == current_data_version do
+      %{state | last_data_version: current_data_version}
+    else
+      {:ok, notifications} = state.notification_fetcher.()
+      {:ok, _pruned} = state.pruner.()
 
-    state
+      Enum.each(notifications, fn notification ->
+        Phoenix.PubSub.broadcast(
+          state.pubsub,
+          topic(),
+          {:entity_changed, notification_payload(notification)}
+        )
+      end)
+
+      %{state | last_data_version: current_data_version}
+    end
   end
 
   defp notification_payload(notification) do

lib/bds/desktop/shell_commands.ex

@@ -559,27 +559,62 @@ defmodule BDS.Desktop.ShellCommands do
     end
   end
 
-  defp wait_for_group_phase(_group_id, _names, timeout) when timeout <= 0, do: :timeout
-
   defp wait_for_group_phase(group_id, names, timeout) do
+    if timeout <= 0 do
+      :timeout
+    else
+      Phoenix.PubSub.subscribe(BDS.PubSub, Tasks.topic())
+
+      try do
+        case group_phase_status(group_id, names) do
+          :waiting -> wait_for_group_phase_message(group_id, names, timeout)
+          status -> status
+        end
+      after
+        Phoenix.PubSub.unsubscribe(BDS.PubSub, Tasks.topic())
+      end
+    end
+  end
+
+  defp wait_for_group_phase_message(group_id, names, timeout) do
+    started_at = System.monotonic_time(:millisecond)
+
+    receive do
+      {:task_terminal, task} ->
+        elapsed = System.monotonic_time(:millisecond) - started_at
+
+        cond do
+          task.group_id == group_id and task.name in names and task.status == :failed ->
+            :failed
+
+          task.group_id == group_id and task.name in names ->
+            case group_phase_status(group_id, names) do
+              :waiting ->
+                wait_for_group_phase_message(group_id, names, timeout - elapsed)
+
+              status ->
+                status
+            end
+
+          true ->
+            wait_for_group_phase_message(group_id, names, timeout - elapsed)
+        end
+    after
+      timeout ->
+        :timeout
+    end
+  end
+
+  defp group_phase_status(group_id, names) do
     tasks =
       BDS.Tasks.list_tasks()
       |> Enum.filter(&(&1.group_id == group_id and &1.name in names))
 
     cond do
-      length(tasks) < length(names) ->
-        Process.sleep(50)
-        wait_for_group_phase(group_id, names, timeout - 50)
-
-      Enum.any?(tasks, &(&1.status == :failed)) ->
-        :failed
-
-      Enum.all?(tasks, &(&1.status == :completed)) ->
-        :ok
-
-      true ->
-        Process.sleep(50)
-        wait_for_group_phase(group_id, names, timeout - 50)
+      length(tasks) < length(names) -> :waiting
+      Enum.any?(tasks, &(&1.status == :failed)) -> :failed
+      Enum.all?(tasks, &(&1.status == :completed)) -> :ok
+      true -> :waiting
     end
   end
 

lib/bds/tasks.ex

@@ -7,11 +7,14 @@ defmodule BDS.Tasks do
   @default_progress_throttle_ms 250
   @default_recent_finished_limit 10
   @default_finished_task_ttl_ms :timer.hours(1)
+  @topic "tasks"
 
   def start_link(_opts) do
     GenServer.start_link(__MODULE__, %{}, name: __MODULE__)
   end
 
+  def topic, do: @topic
+
   def submit_task(name, work, attrs \\ %{})
       when is_binary(name) and is_function(work, 1) and is_map(attrs) do
     GenServer.call(__MODULE__, {:submit_task, name, work, attrs})
@@ -136,6 +139,8 @@ defmodule BDS.Tasks do
           |> start_queued_tasks()
           |> schedule_finished_task_eviction()
 
+        broadcast_terminal_task(next_state.tasks[task_id])
+
         {:reply, :ok, next_state}
 
       Enum.any?(state.queue, fn {queued_id, _work} -> queued_id == task_id end) ->
@@ -148,6 +153,8 @@ defmodule BDS.Tasks do
           |> start_queued_tasks()
           |> schedule_finished_task_eviction()
 
+        broadcast_terminal_task(next_state.tasks[task_id])
+
         {:reply, :ok, next_state}
 
       state.tasks[task_id] == nil ->
@@ -179,6 +186,8 @@ defmodule BDS.Tasks do
       |> start_queued_tasks()
       |> schedule_finished_task_eviction()
 
+    broadcast_terminal_task(next_state.tasks[task_id])
+
     {:reply, :ok, next_state}
   end
 
@@ -193,6 +202,8 @@ defmodule BDS.Tasks do
       |> start_queued_tasks()
       |> schedule_finished_task_eviction()
 
+    broadcast_terminal_task(next_state.tasks[task_id])
+
     {:reply, :ok, next_state}
   end
 
@@ -240,6 +251,10 @@ defmodule BDS.Tasks do
           |> start_queued_tasks()
           |> schedule_finished_task_eviction()
 
+        if task.status != :cancelled do
+          broadcast_terminal_task(next_state.tasks[task_id])
+        end
+
         {:noreply, next_state}
     end
   end
@@ -271,6 +286,10 @@ defmodule BDS.Tasks do
           |> start_queued_tasks()
           |> schedule_finished_task_eviction()
 
+        if task.status != :cancelled and next_state.tasks[task_id].status == :failed do
+          broadcast_terminal_task(next_state.tasks[task_id])
+        end
+
         {:noreply, next_state}
     end
   end
@@ -362,6 +381,14 @@ defmodule BDS.Tasks do
     end)
   end
 
+  defp broadcast_terminal_task(nil), do: :ok
+
+  defp broadcast_terminal_task(task) when task.status in [:completed, :failed, :cancelled] do
+    Phoenix.PubSub.broadcast(BDS.PubSub, topic(), {:task_terminal, public_task(task)})
+  end
+
+  defp broadcast_terminal_task(_task), do: :ok
+
   defp schedule_finished_task_eviction(state) do
     Process.send_after(self(), :evict_finished_tasks, finished_task_ttl_ms())
     state