Feature: Add streams section

dev/components.ex

@@ -9,6 +9,7 @@ defmodule LiveDebuggerDev.Components do
     {"/embedded", "Embedded"},
     {"/embedded_in_controller", "EmbeddedInController"},
     {"/endless_crash_reload", "EndlessCrashReload"},
+    {"/stream", "Stream"},
     {"/async_demo", "AsyncDemo"}
   ]
 

dev/live_components/stream.ex

@@ -0,0 +1,24 @@
+defmodule LiveDebuggerDev.LiveComponents.StreamComponent do
+  use DevWeb, :live_component
+
+  def update(assigns, socket) do
+    items = [
+      %{id: "item-1", name: "first"},
+      %{id: "item-2", name: "second"},
+      %{id: "item-3", name: "third"}
+    ]
+
+    socket
+    |> assign(assigns)
+    |> stream(:component_items, items)
+    |> then(&{:ok, &1})
+  end
+
+  def render(assigns) do
+    ~H"""
+    <div>
+      <.box title="Component with stream" color="yellow"></.box>
+    </div>
+    """
+  end
+end

dev/live_views/stream.ex

@@ -0,0 +1,207 @@
+defmodule LiveDebuggerDev.LiveViews.Stream do
+  use DevWeb, :live_view
+
+  alias LiveDebuggerDev.LiveComponents
+
+  @impl true
+  def mount(_params, _session, socket) do
+    socket =
+      socket
+      |> stream_configure(:items, dom_id: &"item-#{&1.id}")
+      |> stream_configure(:another_items, dom_id: &"another-#{&1.id}")
+      |> assign(:current_id, 0)
+      |> assign(:another_items_id, 0)
+      |> stream(:items, [])
+      |> stream(:another_items, [])
+
+    {:ok, socket}
+  end
+
+  @impl true
+  def render(assigns) do
+    ~H"""
+    <.box title="Items Stream options">
+      <.button id="create-item" phx-click="create_item">Create Item</.button>
+      <.button id="update-item" phx-click="update_item">Update Last Item</.button>
+      <.button id="insert-many" phx-click="insert_many">Insert Many (Reverse Order)</.button>
+      <.button id="insert-at-index" phx-click="insert_at_index">Insert At Index 4</.button>
+      <.button id="delete-item" phx-click="delete_item">Delete Last</.button>
+      <.button id="reset-items" phx-click="reset_items">Reset Stream</.button>
+      <.button id="add-new-stream" phx-click="add_new_stream">Add new stream</.button>
+      <.button id="delete-both-last" phx-click="delete_both_last">
+        Delete Last From Both Streams
+      </.button>
+    </.box>
+
+    <hr />
+
+    <.box title="Items Stream">
+      <ul id="item-list" phx-update="stream">
+        <li :for={{id, item} <- @streams.items} id={id}>
+          <strong>ID:</strong> <%= item.id %>, <strong>Number:</strong> <%= item.number %>
+          <button phx-click="delete_by_dom_id" phx-value-id={id}>X</button>
+        </li>
+      </ul>
+    </.box>
+
+    <hr />
+
+    <.box title="Another Items stream options">
+      <.button id="create-another-item" phx-click="create_another_item">Create Another Item</.button>
+    </.box>
+
+    <.box title="Another Items Stream">
+      <ul id="another-items-list" phx-update="stream">
+        <li :for={{id, item} <- @streams.another_items} id={id}>
+          <strong><%= item.id %></strong> -><%= item.number %>
+        </li>
+      </ul>
+    </.box>
+
+    <hr />
+
+    <.live_component id="stream_component" module={LiveComponents.StreamComponent} />
+    """
+  end
+
+  @impl true
+  def handle_event("create_item", _params, socket) do
+    next_id = socket.assigns.current_id
+    item = %{id: next_id, number: Enum.random(1..1000)}
+
+    socket =
+      socket
+      |> stream_insert(:items, item, at: 0)
+      |> assign(:current_id, next_id + 1)
+
+    {:noreply, socket}
+  end
+
+  @impl true
+  def handle_event("update_item", _params, socket) do
+    id_to_update = socket.assigns.current_id - 1
+
+    if id_to_update >= 0 do
+      updated_item = %{id: id_to_update, number: Enum.random(1001..2000)}
+
+      socket =
+        socket
+        |> stream_insert(:items, updated_item, update_only: true)
+
+      {:noreply, socket}
+    else
+      {:noreply, socket}
+    end
+  end
+
+  @impl true
+  def handle_event("insert_many", _params, socket) do
+    start_id = socket.assigns.current_id
+
+    items =
+      Enum.map(0..2, fn i ->
+        %{id: start_id + i, number: Enum.random(1..99)}
+      end)
+
+    socket =
+      socket
+      |> stream(:items, Enum.reverse(items), at: -1)
+      |> assign(:current_id, start_id + length(items))
+
+    {:noreply, socket}
+  end
+
+  @impl true
+  def handle_event("insert_at_index", _params, socket) do
+    next_id = socket.assigns.current_id
+    item = %{id: next_id, number: 9999}
+
+    socket =
+      socket
+      |> stream_insert(:items, item, at: 4)
+      |> assign(:current_id, next_id + 1)
+
+    {:noreply, socket}
+  end
+
+  @impl true
+  def handle_event("delete_item", _params, socket) do
+    last_id = socket.assigns.current_id - 1
+
+    if last_id >= 0 do
+      socket =
+        socket
+        |> stream_delete(:items, %{id: last_id})
+        |> assign(:current_id, last_id)
+
+      {:noreply, socket}
+    else
+      {:noreply, socket}
+    end
+  end
+
+  @impl true
+  def handle_event("delete_by_dom_id", %{"id" => dom_id}, socket) do
+    socket = stream_delete_by_dom_id(socket, :items, dom_id)
+    {:noreply, socket}
+  end
+
+  @impl true
+  def handle_event("reset_items", _params, socket) do
+    socket =
+      socket
+      |> stream(:items, [], reset: true)
+      |> assign(:current_id, 0)
+
+    {:noreply, socket}
+  end
+
+  @impl true
+  def handle_event("create_another_item", _params, socket) do
+    next_id = socket.assigns.another_items_id
+    item = %{id: next_id, number: Enum.random(1..100)}
+
+    socket =
+      socket
+      |> stream_insert(:another_items, item, at: -1)
+      |> assign(:another_items_id, next_id + 1)
+
+    {:noreply, socket}
+  end
+
+  @impl true
+  def handle_event("delete_both_last", _params, socket) do
+    last_item_id = socket.assigns.current_id - 1
+    last_another_id = socket.assigns.another_items_id - 1
+
+    socket =
+      socket
+      |> then(fn s ->
+        if last_item_id >= 0 do
+          stream_delete(s, :items, %{id: last_item_id})
+        else
+          s
+        end
+      end)
+      |> then(fn s ->
+        if last_another_id >= 0 do
+          stream_delete(s, :another_items, %{id: last_another_id})
+        else
+          s
+        end
+      end)
+      |> assign(:current_id, max(last_item_id, 0))
+      |> assign(:another_items_id, max(last_another_id, 0))
+
+    {:noreply, socket}
+  end
+
+  @impl true
+  def handle_event("add_new_stream", _params, socket) do
+    socket =
+      socket
+      |> stream(:new_items, [])
+
+    {:noreply, socket}
+  end
+end

dev/router.ex

@@ -18,6 +18,7 @@ defmodule LiveDebuggerDev.Router do
     live("/messages", LiveDebuggerDev.LiveViews.Messages)
     live("/embedded", LiveDebuggerDev.LiveViews.Embedded)
     live("/endless_crash_reload", LiveDebuggerDev.LiveViews.EndlessCrashReload)
+    live("/stream", LiveDebuggerDev.LiveViews.Stream)
     live("/async_demo", LiveDebuggerDev.LiveViews.AsyncDemo)
     get("/embedded_in_controller", LiveDebuggerDev.EmbeddedLiveViewController, :embedded)
   end

lib/live_debugger/app/debugger/streams/actions.ex

@@ -0,0 +1,27 @@
+defmodule LiveDebugger.App.Debugger.Streams.Actions do
+  @moduledoc """
+  Actions for `LiveDebugger.App.Debugger.Streams` context.
+  """
+
+  alias LiveDebugger.App.Debugger.Streams.StreamUtils
+
+  @type stream_update_result :: %{
+          functions: [function()],
+          config: [function()],
+          name: atom()
+        }
+
+  @spec update_stream(
+          stream_updates :: StreamUtils.live_stream_item(),
+          dom_id_fun :: (any() -> String.t())
+        ) ::
+          {:ok, stream_update_result()} | {:error, String.t()}
+  def update_stream(stream_updates, dom_id_fun) do
+    with name <- stream_updates.name,
+         funs <- StreamUtils.stream_update_functions(stream_updates),
+         config <-
+           StreamUtils.stream_config(stream_updates, dom_id: dom_id_fun) do
+      {:ok, %{functions: funs, config: config, name: name}}
+    end
+  end
+end

lib/live_debugger/app/debugger/streams/queries.ex

@@ -0,0 +1,46 @@
+defmodule LiveDebugger.App.Debugger.Streams.Queries do
+  @moduledoc """
+  Queries for `LiveDebugger.App.Debugger.Streams` context.
+  """
+
+  alias LiveDebugger.API.TracesStorage
+
+  alias LiveDebugger.App.Debugger.Streams.StreamUtils
+
+  require Logger
+
+  @type streams_result :: %{
+          functions: [function()],
+          config: map(),
+          names: [atom()]
+        }
+
+  @spec fetch_streams_from_render_traces(pid :: pid(), node_id :: TreeNode.id()) ::
+          {:ok, streams_result()} | {:error, String.t()}
+  def fetch_streams_from_render_traces(pid, node_id) do
+    with {:ok, render_traces} <- fetch_render_traces(pid, node_id),
+         stream_traces <- StreamUtils.extract_stream_traces(render_traces),
+         names <- StreamUtils.streams_names(stream_traces),
+         funs <- StreamUtils.streams_functions(stream_traces, names),
+         config <- StreamUtils.streams_config(stream_traces, names) do
+      {:ok, %{functions: funs, config: config, names: names}}
+    else
+      :end_of_table ->
+        :end_of_table
+
+      error ->
+        Logger.error("Failed to fetch streams: #{inspect(error)}")
+        {:error, "Failed to fetch streams"}
+    end
+  end
+
+  defp fetch_render_traces(pid, node_id) do
+    case TracesStorage.get!(pid, functions: ["render/1"], node_id: node_id) do
+      :end_of_table ->
+        :end_of_table
+
+      {stream_updates, _trace} ->
+        {:ok, stream_updates}
+    end
+  end
+end