Channels, PubSub and Presence

Real-time messaging built in

Phoenix.PubSub is a publish/subscribe system: processes subscribe to topics ("stock:pen") and any process can broadcast a message to all subscribers, across all nodes of a cluster. It is how LiveViews update when data changes elsewhere: a context function updates the database and broadcasts, and every subscribed LiveView receives the message in handle_info. Channels provide bidirectional WebSocket communication for any client (JavaScript, mobile apps, IoT devices): clients join topics, push events to the server, and the server replies or broadcasts; join/3, handle_in/3 and broadcast!/3 are the core callbacks, and sockets authenticate in connect/3. Phoenix.Presence tracks who is online in a topic using a conflict-free replicated data type, so presence lists stay consistent across a cluster without a central store, and it powers "who is viewing" or "users online" features. Because each connection is a lightweight process, a single Phoenix server can handle a very large number of simultaneous WebSocket connections.

Broadcasting from a context and a chat channel

Context code broadcasts; subscribers update in real time.

defmodule Shop.Inventory do
  def reserve(sku, qty) do
    with {:ok, remaining} <- decrement_stock(sku, qty) do
      Phoenix.PubSub.broadcast(Shop.PubSub, "stock:#{sku}", {:stock_changed, remaining})
      {:ok, remaining}
    end
  end

  def stock(_sku), do: 10
  defp decrement_stock(_sku, qty) when qty <= 10, do: {:ok, 10 - qty}
  defp decrement_stock(_sku, _qty), do: {:error, :insufficient}
end

defmodule ShopWeb.SupportChannel do
  use ShopWeb, :channel
  alias ShopWeb.Presence

  @impl true
  def join("support:" <> ticket_id, _payload, socket) do
    send(self(), :after_join)
    {:ok, assign(socket, :ticket_id, ticket_id)}
  end

  @impl true
  def handle_info(:after_join, socket) do
    {:ok, _} = Presence.track(socket, socket.assigns.user_id, %{online_at: System.system_time(:second)})
    push(socket, "presence_state", Presence.list(socket))
    {:noreply, socket}
  end

  @impl true
  def handle_in("message", %{"body" => body}, socket) when byte_size(body) in 1..2_000 do
    broadcast!(socket, "message", %{user: socket.assigns.user_id, body: body})
    {:noreply, socket}
  end

  def handle_in("message", _payload, socket), do: {:reply, {:error, %{reason: "invalid message"}}, socket}
end

A radio station

PubSub is a radio station: broadcasters transmit on a frequency (topic) and anyone tuned in hears it, without knowing who else is listening. Presence is the live counter of how many radios are tuned to each station.

Quick check: What does Phoenix.PubSub let processes do?

• Share memory
• Run SQL queries
• Compile templates
• Subscribe to topics and receive messages broadcast to them, even across cluster nodes