custom_kino_as_event_stream.livemd

Custom kino as event stream

Mix.install(
  [
    {:kino, "~> 0.9.4"}
  ],
  consolidate_protocols: false
)

Example

Kino.listen/2 works with any enumerable, so what we can:

1. Define custom struct wrapping Kino.JS.Live
2. Implement Kino.Render for our struct, simply by proxying to the underlying Kino.JS.Live
3. Add a function to build a stream of events
   • use Stream.resource/3
   • keep track of subscribers in Kino.JS.Live state
   • dispatch all relevant image to subscriber processes
4. Implement Enumerable for our struct, simply by proxying to the stream

defmodule KinoCounter do
  use Kino.JS
  use Kino.JS.Live

  defstruct [:js_live]

  def new(count) do
    js_live = Kino.JS.Live.new(__MODULE__, count)
    %__MODULE__{js_live: js_live}
  end

  def event_stream(kino) do
    Stream.resource(
      fn ->
        ref = make_ref()
        Kino.JS.Live.cast(kino.js_live, {:subscribe, self(), ref})
        monitor_ref = Kino.JS.Live.monitor(kino.js_live)
        {ref, monitor_ref}
      end,
      fn {ref, monitor_ref} ->
        receive do
          {^ref, event} ->
            {[event], {ref, monitor_ref}}

          {:DOWN, ^monitor_ref, :process, _pid, _reason} ->
            {:halt, {ref, monitor_ref}}
        end
      end,
      fn {_ref, _monitor_ref} ->
        Kino.JS.Live.cast(kino.js_live, {:unsubscribe, self()})
      end
    )
  end

  @impl true
  def init(count, ctx) do
    {:ok, assign(ctx, count: count, subscribers: %{})}
  end

  @impl true
  def handle_connect(ctx) do
    {:ok, ctx.assigns.count, ctx}
  end

  @impl true
  def handle_event("bump", _, ctx) do
    {:noreply, bump_count(ctx)}
  end

  defp bump_count(ctx) do
    ctx = update(ctx, :count, &(&1 + 1))
    broadcast_event(ctx, "update", ctx.assigns.count)
    notify_event(ctx, %{type: :bump, count: ctx.assigns.count})
    ctx
  end

  @impl true
  def handle_cast({:subscribe, pid, ref}, ctx) do
    Process.monitor(pid)
    {:noreply, update(ctx, :subscribers, &Map.put(&1, pid, ref))}
  end

  def handle_cast({:unsubscribe, pid}, ctx) do
    Process.monitor(pid)
    {:noreply, update(ctx, :subscribers, &Map.delete(&1, pid))}
  end

  @impl true
  def handle_info({:DOWN, _ref, :process, pid, _reason}, ctx) do
    {:noreply, update(ctx, :subscribers, &Map.delete(&1, pid))}
  end

  defp notify_event(ctx, event) do
    for {pid, ref} <- ctx.assigns.subscribers do
      send(pid, {ref, event})
    end
  end

  asset "main.js" do
    """
    export function init(ctx, count) {
      ctx.root.innerHTML = `
        <div id="count"></div>
        <button id="bump">Bump</button>
      `;

      const countEl = document.getElementById("count");
      const bumpEl = document.getElementById("bump");

      countEl.innerHTML = count;

      ctx.handleEvent("update", (count) => {
        countEl.innerHTML = count;
      });

      bumpEl.addEventListener("click", (event) => {
        ctx.pushEvent("bump");
      });
    }
    """
  end
end

defimpl Kino.Render, for: KinoCounter do
  def to_livebook(counter) do
    Kino.Render.to_livebook(counter.js_live)
  end
end

defimpl Enumerable, for: KinoCounter do
  def reduce(kino, acc, fun), do: Enumerable.reduce(KinoCounter.event_stream(kino), acc, fun)
  def member?(_kino, _value), do: {:error, __MODULE__}
  def count(_kino), do: {:error, __MODULE__}
  def slice(_kino), do: {:error, __MODULE__}
end

counter = KinoCounter.new(0)

Kino.listen(counter, fn event ->
  IO.inspect(event)
end)

Simpler alternative

The simples option would be to add a more specific KinoCounter.listen(kino, function). This way we don't even need the custom struct. The downside is that all functions run within the Kino.JS.Live process.