I’ve reworked my version a bit. Thanks again to rhcarvalho. I couldn’t have done this without your version.

I realized I needed

1. Throttle handle_info calls only - I don’t see handle_event calls at a sufficient rate to be concerned.
   Since the msg param of handle_info can be data of any shape, I added a callback to know when to throttle. The callback can use pattern matching or any other means.
2. The first message should be handled immediately.
   Actually the first message and any other message when a batch hasn’t been processed since the timeout.
   Let’s say the timeout is 500ms and 1000ms goes by since the last batch. The next message should be released immediately.
3. Messages should be batched based on timeout not count. I was shooting for some level of realtime-ness without overloading the socket. I want events to be published as available at a slower rate. I didn’t want to hold onto messages indefinitely.

  @behaviour HandleInfoThrottler

  on_mount({HandleInfoThrottler, {:vote_throttler, __MODULE__, 500}})

Here’s my updated version.

  defmodule WeddingGamesWeb.HandleInfoThrottler do
    @moduledoc """
    Throttles handle_info messages based on timeout.

    The first handle_info received is dispatched immediately.

    Subsequent handle_info messages are batched and dispatched after the timeout has elapsed.

    ## Usage
    You add one or more handle_info throttlers to you LiveView by adding on_mount hooks.
    - `throttler_name` is a unique name for the throttler.
    - `callback_module` is the module that implements the HandleInfoThrottler behaviour.
    - `timeout` is the timeout in milliseconds.

    ### on_mount

    ```elixir
    on_mount({HandleInfoThrottler, {:throttler_name, callback_module, timeout}})
    ```

    """

    @doc """
      Return true if the throttler should throttle the message.  Return false when it does not apply.
    """
    @callback throttle_message?(throttler_name :: atom(), msg :: any()) :: boolean()

    @doc """
    Called when one or more messages are ready to be processed.  This is either the first message
    or the collection of batched messages

    """
    @callback handle_batched_info(
                throttler_name :: atom(),
                batch :: list(),
                socket :: Phoenix.LiveView.Socket.t()
              ) :: {:noreply, Phoenix.LiveView.Socket.t()}

    @doc """
    Establishes a handle_info throttler
    """
    def on_mount({throttler_name, handler, timeout}, _params, _session, socket) do
      {:cont,
      socket
      |> put_handler(throttler_name, handler)
      |> put_timeout(throttler_name, timeout)
      |> put_last_batch_at(throttler_name, ~U[2021-01-01T00:00:00Z])
      |> Phoenix.LiveView.attach_hook(
        {"handle_info_throttler_#{throttler_name}"},
        :handle_info,
        fn
          {__MODULE__, throttler_name, :flush_batch}, socket ->
            {:halt, flush_batch(socket, throttler_name)}

          msg, socket ->
            handler = get_handler(socket, throttler_name)

            if handler.batch_message?(throttler_name, msg) do
              {:halt,
                socket
                |> add_to_batch(throttler_name, msg)
                |> maybe_flush_or_schedule(throttler_name)}
            else
              {:cont, socket}
            end
        end
      )}
    end

    defp maybe_flush_or_schedule(socket, throttler_name) do
      last_batch_at = get_last_batch_at(socket, throttler_name)
      timeout = get_timeout(socket, throttler_name)

      ms_since_last_batch = DateTime.diff(DateTime.utc_now(), last_batch_at, :millisecond)

      if ms_since_last_batch > timeout do
        socket
        |> flush_batch(throttler_name)
      else
        socket
        |> schedule_flush_batch_if_needed(throttler_name, timeout - ms_since_last_batch)
      end
    end

    defp flush_batch(socket, throttler_name) do
      handler = get_handler(socket, throttler_name)
      batch = get_batch(socket, throttler_name) |> Enum.reverse()

      {:noreply, socket} = handler.handle_batched_info(throttler_name, batch, socket)

      socket
      |> put_batch(throttler_name, [])
      |> cancel_timeout(throttler_name)
      |> put_last_batch_at(throttler_name, DateTime.utc_now())
    end

    defp schedule_flush_batch_if_needed(socket, throttler_name, send_after_ms) do
      get_timeout_ref(socket, throttler_name)
      |> case do
        nil ->
          schedule_flush_batch(socket, throttler_name, send_after_ms)

        _ ->
          socket
      end
    end

    defp schedule_flush_batch(socket, throttler_name, send_after_ms) do
      socket
      |> put_timeout_ref(
        throttler_name,
        Process.send_after(self(), {__MODULE__, throttler_name, :flush_batch}, send_after_ms)
      )
    end

    defp cancel_timeout(socket, throttler_name) do
      if timeout_ref = get_timeout_ref(socket, throttler_name) do
        Process.cancel_timer(timeout_ref)
      end

      socket |> put_timeout_ref(throttler_name, nil)
    end

    defp add_to_batch(socket, throttler_name, msg) do
      # note: batch throttler_names are in reverse order
      batch = [msg | get_batch(socket, throttler_name)]
      put_batch(socket, throttler_name, batch)
    end

    defp get_batch(socket, throttler_name) do
      socket.private[{__MODULE__, throttler_name, :batch}] || []
    end

    defp get_last_batch_at(socket, throttler_name) do
      socket.private[{__MODULE__, throttler_name, :last_batch_at}]
    end

    defp get_handler(socket, throttler_name) do
      socket.private[{__MODULE__, throttler_name, :handler}]
    end

    defp get_timeout(socket, throttler_name) do
      socket.private[{__MODULE__, throttler_name, :timeout}]
    end

    def get_timeout_ref(socket, throttler_name) do
      socket.private[{__MODULE__, throttler_name, :timeout_ref}]
    end

    defp put_batch(socket, throttler_name, batch) do
      socket |> Phoenix.LiveView.put_private({__MODULE__, throttler_name, :batch}, batch)
    end

    defp put_last_batch_at(socket, throttler_name, last_batch_at) do
      socket
      |> Phoenix.LiveView.put_private({__MODULE__, throttler_name, :last_batch_at}, last_batch_at)
    end

    defp put_handler(socket, throttler_name, handler) do
      socket |> Phoenix.LiveView.put_private({__MODULE__, throttler_name, :handler}, handler)
    end

    defp put_timeout(socket, throttler_name, timeout) do
      socket |> Phoenix.LiveView.put_private({__MODULE__, throttler_name, :timeout}, timeout)
    end

    defp put_timeout_ref(socket, throttler_name, timeout_ref) do
      socket
      |> Phoenix.LiveView.put_private({__MODULE__, throttler_name, :timeout_ref}, timeout_ref)
    end
  end

First off - very nice and clean!! Thank you!

Your solution exposed me to a couple of facilities I hadn’t encountered before like LiveView.attach_hook and LiveView.put_private.

I liked your idea that the Batcher could also process handle_info.

I experimented with reschedule_flush_batch to not cancel the existing timer - only set one to get handle_batch after either the count or the timeout from the first message of a batch. Testament to your design that it was easy to try. Either timeout interpretation could be correct.

This might help you in the future
https://github.com/phoenix-playground/phoenix_playground