phoenix_pubsub v1.1.0 Phoenix.PubSub View Source

Front-end to Phoenix pubsub layer.

Used internally by Channels for pubsub broadcast but also provides an API for direct usage.

Adapters

Phoenix pubsub was designed to be flexible and support multiple backends. We currently ship with two backends:

  • Phoenix.PubSub.PG2 - uses Distributed Elixir, directly exchanging notifications between servers

  • Phoenix.PubSub.Redis - uses Redis to exchange data between servers

Pubsub adapters are often configured in your endpoint:

config :my_app, MyApp.Endpoint,
  pubsub: [adapter: Phoenix.PubSub.PG2,
           pool_size: 1,
           name: MyApp.PubSub]

The configuration above takes care of starting the pubsub backend and exposing its functions via the endpoint module. If no adapter but a name is given, nothing will be started, but the pubsub system will work by sending events and subscribing to the given name.

Direct usage

It is also possible to use Phoenix.PubSub directly or even run your own pubsub backends outside of an Endpoint.

The first step is to start the adapter of choice in your supervision tree:

supervisor(Phoenix.PubSub.Redis, [:my_pubsub, host: "192.168.100.1"])

The configuration above will start a Redis pubsub and register it with name :my_pubsub.

You can now use the functions in this module to subscribe and broadcast messages:

iex> PubSub.subscribe :my_pubsub, "user:123"
:ok
iex> Process.info(self())[:messages]
[]
iex> PubSub.broadcast :my_pubsub, "user:123", {:user_update, %{id: 123, name: "Shane"}}
:ok
iex> Process.info(self())[:messages]
{:user_update, %{id: 123, name: "Shane"}}

Implementing your own adapter

PubSub adapters run inside their own supervision tree. If you are interested in providing your own adapter, let’s call it Phoenix.PubSub.MyQueue, the first step is to provide a supervisor module that receives the server name and a bunch of options on start_link/2:

defmodule Phoenix.PubSub.MyQueue do
  def start_link(name, options) do
    Supervisor.start_link(__MODULE__, {name, options},
                          name: Module.concat(name, Supervisor))
  end

  def init({name, options}) do
    ...
  end
end

On init/1, you will define the supervision tree and use the given name to register the main pubsub process locally. This process must be able to handle the following GenServer calls:

  • subscribe - subscribes the given pid to the given topic sends: {:subscribe, pid, topic, opts} respond with: :ok | {:error, reason} | {:perform, {m, f, a}}

  • unsubscribe - unsubscribes the given pid from the given topic sends: {:unsubscribe, pid, topic} respond with: :ok | {:error, reason} | {:perform, {m, f, a}}

  • broadcast - broadcasts a message on the given topic sends: {:broadcast, :none | pid, topic, message} respond with: :ok | {:error, reason} | {:perform, {m, f, a}}

Offloading work to clients via MFA response

The Phoenix.PubSub API allows any of its functions to handle a response from the adapter matching {:perform, {m, f, a}}. The PubSub client will recursively invoke all MFA responses until a result is returned. This is useful for offloading work to clients without blocking your PubSub adapter. See Phoenix.PubSub.PG2 implementation for examples.

Link to this section Summary

Functions

Broadcasts message on given topic

Broadcasts message on given topic

Broadcasts message to all but from_pid on given topic

Broadcasts message to all but from_pid on given topic. See Phoenix.PubSub.broadcast/3 for usage details

Broadcasts message on given topic, to a single node

Broadcasts message on given topic, to a single node

Broadcasts message to all but from_pid on given topic, to a single node

Broadcasts message to all but from_pid on given topic, to a single node. See Phoenix.PubSub.broadcast/3 for usage details

Returns the node name of the PubSub server

Subscribes the caller to the PubSub adapter’s topic

Unsubscribes the caller from the PubSub adapter’s topic

Link to this section Types

Link to this type node_name() View Source
node_name() :: atom :: binary()

Link to this section Functions

Link to this function broadcast!(server, topic, message) View Source
broadcast!(atom(), binary(), term()) :: :ok | no_return()

Broadcasts message on given topic.

Raises Phoenix.PubSub.BroadcastError if broadcast fails. See Phoenix.PubSub.broadcast/3 for usage details.

Link to this function broadcast(server, topic, message) View Source
broadcast(atom(), binary(), term()) :: :ok | {:error, term()}

Broadcasts message on given topic.

  • server - The Pid or registered server name and optional node to scope the broadcast, for example: MyApp.PubSub, {MyApp.PubSub, :a@node}
  • topic - The topic to broadcast to, ie: "users:123"
  • message - The payload of the broadcast
Link to this function broadcast_from!(server, from_pid, topic, message) View Source
broadcast_from!(atom() | {atom(), atom()}, pid(), binary(), term()) ::
  :ok | no_return()

Broadcasts message to all but from_pid on given topic.

Raises Phoenix.PubSub.BroadcastError if broadcast fails. See Phoenix.PubSub.broadcast/3 for usage details.

Link to this function broadcast_from(server, from_pid, topic, message) View Source
broadcast_from(atom(), pid(), binary(), term()) :: :ok | {:error, term()}

Broadcasts message to all but from_pid on given topic. See Phoenix.PubSub.broadcast/3 for usage details.

Link to this function direct_broadcast!(node_name, server, topic, message) View Source
direct_broadcast!(node_name(), atom(), binary(), term()) :: :ok | no_return()

Broadcasts message on given topic, to a single node.

Raises Phoenix.PubSub.BroadcastError if broadcast fails. See Phoenix.PubSub.broadcast/3 for usage details.

Link to this function direct_broadcast(node_name, server, topic, message) View Source
direct_broadcast(node_name(), atom(), binary(), term()) ::
  :ok | {:error, term()}

Broadcasts message on given topic, to a single node.

  • node - The name of the node to broadcast the message on
  • server - The Pid or registered server name and optional node to scope the broadcast, for example: MyApp.PubSub, {MyApp.PubSub, :a@node}
  • topic - The topic to broadcast to, ie: "users:123"
  • message - The payload of the broadcast
Link to this function direct_broadcast_from!(node_name, server, from_pid, topic, message) View Source
direct_broadcast_from!(node_name(), atom(), pid(), binary(), term()) ::
  :ok | no_return()

Broadcasts message to all but from_pid on given topic, to a single node.

Raises Phoenix.PubSub.BroadcastError if broadcast fails. See Phoenix.PubSub.broadcast/3 for usage details.

Link to this function direct_broadcast_from(node_name, server, from_pid, topic, message) View Source
direct_broadcast_from(node_name(), atom(), pid(), binary(), term()) ::
  :ok | {:error, term()}

Broadcasts message to all but from_pid on given topic, to a single node. See Phoenix.PubSub.broadcast/3 for usage details.

Link to this function node_name(server) View Source
node_name(atom()) :: atom :: binary()

Returns the node name of the PubSub server.

Link to this function subscribe(server, topic) View Source
subscribe(atom(), binary()) :: :ok | {:error, term()}
Link to this function subscribe(server, pid, topic) View Source
subscribe(atom(), pid(), binary()) :: :ok | {:error, term()}
subscribe(atom(), binary(), Keyword.t()) :: :ok | {:error, term()}

Subscribes the caller to the PubSub adapter’s topic.

  • server - The Pid registered name of the server
  • topic - The topic to subscribe to, for example: "users:123"
  • opts - The optional list of options. See below.

Duplicate Subscriptions

Callers should only subscribe to a given topic a single time. Duplicate subscriptions for a Pid/topic pair are allowed and will cause duplicate events to be sent; however, when using Phoenix.PubSub.unsubscribe/3, all duplicate subscriptions will be dropped.

Options

  • :link - links the subscriber to the pubsub adapter
  • :fastlane - Provides a fastlane path for the broadcasts for %Phoenix.Socket.Broadcast{} events. The fastlane process is notified of a cached message instead of the normal subscriber. Fastlane handlers must implement fastlane/1 callbacks which accepts a Phoenix.Socket.Broadcast struct and returns a fastlaned format for the handler. For example:

    PubSub.subscribe(MyApp.PubSub, "topic1",
      fastlane: {fast_pid, Phoenix.Transports.WebSocketSerializer, ["event1"]})
Link to this function subscribe(server, pid, topic, opts) View Source
subscribe(atom(), pid(), binary(), Keyword.t()) :: :ok | {:error, term()}
Link to this function unsubscribe(server, topic) View Source
unsubscribe(atom(), binary()) :: :ok | {:error, term()}
Link to this function unsubscribe(server, pid, topic) View Source
unsubscribe(atom(), pid(), binary()) :: :ok | {:error, term()}

Unsubscribes the caller from the PubSub adapter’s topic.