Live_view stream support

[vitocal]

Sharing some improvements i've made to live_view stream usage with live_svelte. I've used @woutdp code as a starting point. Ty very much!

It's still kinda rough, but i've added insert, update, delete and order support! It's still missing the length feature.

Anyway, here's the example:

defmodule ExampleWeb.LiveStream do
  use ExampleWeb, :live_view

  require Protocol

  defimpl Jason.Encoder, for: MapSet do
    def encode(struct, opts) do
      Jason.Encode.list(Enum.to_list(struct), opts)
    end
  end

  defimpl Jason.Encoder, for: Tuple do
    def encode(data, options) when is_tuple(data) do
      data
      |> Tuple.to_list()
      |> Jason.Encode.list(options)
    end
  end

  Protocol.derive(Jason.Encoder, Phoenix.LiveView.LiveStream, only: [:name, :inserts, :deletes])

  def render(assigns) do
    ~V"""
    <script>
      import { slide, fade } from "svelte/transition";
      import {from_stream} from "js/live_stream.ts";

      export let streams;
      let logs = from_stream(streams.logs)

      $: {
        logs = from_stream(streams.logs)
      }
    </script>

    <ul>
      {#each logs as log (log.dom_id)}
        <li id={log.dom_id} in:fade>
          {log.id} - {log.body}
        </li>
      {/each}
    </ul>
    """
  end

  def mount(_params, _session, socket) do
    if connected?(socket), do: :timer.send_interval(1000, self(), :tick)
    {:ok, stream(socket, :logs, [add_log("initial")])}
  end

  def handle_info(:tick, socket) do
    datetime =
      DateTime.utc_now()
      |> DateTime.to_string()

    {:noreply, stream_insert(socket, :logs, add_log(datetime))}
  end

  defp add_log(body) do
    %{id: System.unique_integer([:positive]), body: body}
  end
end

And add the live_stream.ts file

/*
 * Svelte synchronization with Phoenix LiveStreams
 */
export interface ILiveStream {
    name: string,
    inserts: ILiveStreamDeltaInsert[],
    deletes: ILiveStreamDeltaDelete[]
}
export type ILiveStreamDeltaInsert = [string, number, any, number | null]
export type ILiveStreamDeltaDelete = string
export type ILiveStreamItem = Record<any, any> & { dom_id: string }

export function LiveStreamItem(delta: ILiveStreamDeltaInsert): ILiveStreamItem {
    const [dom_id, _at, item, _length] = delta;
    return { ...item, dom_id };
}

/**
 * Holds the streamed collections
 */
let liveStreamItems: Record<string, ILiveStreamItem[]> = {};

/**
  Reads the `stream` updates and returns a array for svelte.

  @usage: In your svelte file:
  <script>
   export let items = from_stream(stream);
   export let stream;

   // Setup updates from stream
   $: { items = from_stream(stream) }
  </script>

   {#each items as item (item.dom_id)}
     <Post post={item} {live} />
   {/each}

 */
export function from_stream(stream: ILiveStream): ILiveStreamItem[] {
    let items = liveStreamItems[stream.name] || [];
    if (stream.inserts.length >= 1) {
        insert(stream.inserts);
    } else if (stream.deletes.length >= 1) {
        deletes(stream.deletes);
    }

    commit();
    return items;

    /** Find item id corresponding to dom_id */
    function findItemIdx(dom_id: string): number {
        return items.findIndex((item) => item.dom_id === dom_id)
    }

    /** Remove items */
    function deletes(streamed_deletes: ILiveStreamDeltaDelete[]) {
        items = items.filter(item => !streamed_deletes.includes(item.dom_id));
    }

    /**
     * Insert many or update a single item
     */
    function insert(streamed_inserts: ILiveStreamDeltaInsert[]) {
        if (streamed_inserts.length > 1) {
            items = [...items, ...streamed_inserts.map(LiveStreamItem).filter((item) => findItemIdx(item.dom_id) == -1)];
        } else {
            const [dom_id, at, _item, _length] = streamed_inserts[0];
            const item = LiveStreamItem(streamed_inserts[0]);
            const is_edit = findItemIdx(dom_id);

            if (is_edit >= 0) {
                items[findItemIdx(dom_id)] = item;
            } else {
                const end = (at >= 0 ? at : items.length);
                items = [
                    ...items.slice(0, end),
                    item,
                    ...items.slice(end, items.length)
                ];
            }
        }
    };

    function commit() {
        liveStreamItems[stream.name] = items;
    }
}

/**
 * Reset the stream items
 */
export function reset_stream(stream: ILiveStream): ILiveStreamItem[] {
    liveStreamItems[stream.name] = []
    return liveStreamItems[stream.name];
}

We could improve it further, but i'm a elixir newbie, i need some guidance.
Anyone knows how can we remove the inlined Jason implementation from our LiveView file?