phoenix_pubsub

defmodule Phoenix.PubSub.PG2 do
  @moduledoc """
  Phoenix PubSub adapter based on `:pg`/`:pg2`.

  It runs on Distributed Erlang and is the default adapter.
  """

  use GenServer
  @behaviour Phoenix.PubSub.Adapter

  ## Adapter callbacks

  @impl true
  def node_name(_), do: node()

  @impl true
  def broadcast(adapter_name, topic, message, dispatcher) do
    case pg_members(adapter_name) do
      {:error, {:no_such_group, _}} ->
        {:error, :no_such_group}

      pids ->
        message = forward_to_local(topic, message, dispatcher)

        for pid <- pids, node(pid) != node() do
          send(pid, message)
        end

        :ok
    end
  end

  @impl true
  def direct_broadcast(adapter_name, node_name, topic, message, dispatcher) do
    send({adapter_name, node_name}, {:forward_to_local, topic, message, dispatcher})
    :ok
  end

  defp forward_to_local(topic, message, dispatcher) do
    {:forward_to_local, topic, message, dispatcher}
  end

  ## GenServer callbacks

  @doc false
  def start_link(opts) do
    name = Keyword.fetch!(opts, :name)
    adapter_name = Keyword.fetch!(opts, :adapter_name)
    GenServer.start_link(__MODULE__, {name, adapter_name}, name: adapter_name)
  end

  @impl true
  def init({name, adapter_name}) do
    :ok = pg_join(adapter_name)
    {:ok, name}
  end

  @impl true
  def handle_info({:forward_to_local, topic, message, dispatcher}, pubsub) do
    Phoenix.PubSub.local_broadcast(pubsub, topic, message, dispatcher)
    {:noreply, pubsub}
  end

  @impl true
  def handle_info(_, pubsub) do
    {:noreply, pubsub}
  end

  if Code.ensure_loaded?(:pg) do
    defp pg_members(adapter_name) do
      :pg.get_members(Phoenix.PubSub, adapter_name)
    end

    defp pg_join(adapter_name) do
      :ok = :pg.join(Phoenix.PubSub, adapter_name, self())
    end
  else
    defp pg_members(adapter_name) do
      :pg2.get_members(pg2_namespace(adapter_name))
    end

    defp pg_join(adapter_name) do
      namespace = pg2_namespace(adapter_name)
      :ok = :pg2.create(namespace)
      :ok = :pg2.join(namespace, self())
      :ok
    end

    defp pg2_namespace(adapter_name), do: {:phx, adapter_name}
  end
end