Alternative implementation of LiveData basing on ideas borrowed from Phoenix.Channel and LiveView

live_data.ex

defmodule LiveData.Server do
  use GenServer

  def init({_endpoint, {pid, _}}) do
    {:ok, Process.monitor(pid)}
  end

  def handle_info({Phoenix.Channel, auth_payload, {pid, _} = from, socket}, ref) do
    Process.demonitor(ref)
    fastlane = {:fastlane, socket.transport_pid, socket.serializer}
    Phoenix.PubSub.subscribe(socket.pubsub_server, socket.topic, metadata: fastlane)
    GenServer.reply(from, {:ok, %{}})
    init_join(socket)
  end

  def handle_info(
        %Phoenix.Socket.Message{topic: topic, event: event, payload: payload, ref: ref},
        %{transport_pid: transport_pid, topic: topic, join_ref: join_ref, serializer: serializer} =
          socket
      ) do
    reply = %Phoenix.Socket.Reply{
      topic: topic,
      join_ref: join_ref,
      ref: ref,
      status: :ok,
      payload: %{}
    }

    send(transport_pid, serializer.encode!(reply))
    send(self(), :inc)
    {:noreply, socket}
  end

  def handle_info({:DOWN, ref, _, _, reason}, ref) do
    {:stop, reason, ref}
  end

  def handle_info({:DOWN, _, _, transport_pid, reason}, %{transport_pid: transport_pid} = socket) do
    reason = if reason == :normal, do: {:shutdown, :closed}, else: reason
    {:stop, reason, socket}
  end

  def handle_info(msg, socket) do
    {:noreply, new_socket} = socket.channel.handle_info(msg, socket)

    push_diff(new_socket, socket)
    {:noreply, new_socket}
  end

  def handle_cast(:close, socket) do
    {:stop, {:shutdown, :closed}, socket}
  end

  defp push_diff(new_socket, previous_socket) do
    diff = JSONDiff.diff(previous_socket.assigns, new_socket.assigns)

    if diff != [] do
      push(new_socket, "diff", %{diff: diff})
    end
  end

  defp push(socket, event, payload) do
    message = %Phoenix.Socket.Message{topic: socket.topic, event: event, payload: payload}
    send(socket.transport_pid, socket.serializer.encode!(message))
    socket
  end

  defp init_join(socket) do
    %{transport_pid: transport_pid, serializer: serializer, pubsub_server: pubsub_server} = socket

    Process.monitor(transport_pid)

    {:noreply, %{socket | joined: true}}
  end
end

defmodule LiveData do
  defmacro __using__(_) do
    quote do
      def start_link(triplet) do
        GenServer.start_link(LiveData.Server, triplet)
      end

      def child_spec(init_arg) do
        %{
          id: __MODULE__,
          start: {__MODULE__, :start_link, [init_arg]},
          shutdown: 5000,
          restart: :temporary
        }
      end
    end
  end
end

defmodule ChannelsWeb.IncData do
  use LiveData

  def handle_info(:inc, socket) do
    socket = %{
      socket
      | assigns: socket.assigns |> Map.put(:count, Map.get(socket.assigns, :count, 0) + 1)
    }

    Process.send_after(self(), :inc, 1000)

    {:noreply, socket}
  end
end