View Source Phoenix.Tracker behaviour (Phoenix.PubSub v2.1.1)
Provides distributed presence tracking to processes.
Tracker shards use a heartbeat protocol and CRDT to replicate presence information across a cluster in an eventually consistent, conflict-free manner. Under this design, there is no single source of truth or global process. Each node runs a pool of trackers and node-local changes are replicated across the cluster and handled locally as a diff of changes.
implementing-a-tracker
Implementing a Tracker
To start a tracker, first add the tracker to your supervision tree:
children = [
# ...
{MyTracker, [name: MyTracker, pubsub_server: MyApp.PubSub]}
]
Next, implement MyTracker
with support for the Phoenix.Tracker
behaviour callbacks. An example of a minimal tracker could include:
defmodule MyTracker do
use Phoenix.Tracker
def start_link(opts) do
opts = Keyword.merge([name: __MODULE__], opts)
Phoenix.Tracker.start_link(__MODULE__, opts, opts)
end
def init(opts) do
server = Keyword.fetch!(opts, :pubsub_server)
{:ok, %{pubsub_server: server, node_name: Phoenix.PubSub.node_name(server)}}
end
def handle_diff(diff, state) do
for {topic, {joins, leaves}} <- diff do
for {key, meta} <- joins do
IO.puts "presence join: key \"#{key}\" with meta #{inspect meta}"
msg = {:join, key, meta}
Phoenix.PubSub.direct_broadcast!(state.node_name, state.pubsub_server, topic, msg)
end
for {key, meta} <- leaves do
IO.puts "presence leave: key \"#{key}\" with meta #{inspect meta}"
msg = {:leave, key, meta}
Phoenix.PubSub.direct_broadcast!(state.node_name, state.pubsub_server, topic, msg)
end
end
{:ok, state}
end
end
Trackers must implement start_link/1
, init/1
, and handle_diff/2
.
The init/1
callback allows the tracker to manage its own state when
running within the Phoenix.Tracker
server. The handle_diff
callback
is invoked with a diff of presence join and leave events, grouped by
topic. As replicas heartbeat and replicate data, the local tracker state is
merged with the remote data, and the diff is sent to the callback. The
handler can use this information to notify subscribers of events, as
done above.
An optional handle_info/2
callback may also be invoked to handle
application specific messages within your tracker.
special-considerations
Special Considerations
Operations within handle_diff/2
happen in the tracker server's context.
Therefore, blocking operations should be avoided when possible, and offloaded
to a supervised task when required. Also, a crash in the handle_diff/2
will
crash the tracker server, so operations that may crash the server should be
offloaded with a Task.Supervisor
spawned process.
Link to this section Summary
Functions
Returns a specification to start this module under a supervisor.
Gets presences tracked under a given topic and key pair.
Gracefully shuts down by broadcasting permdown to all replicas.
Lists all presences tracked under a given topic.
Starts a tracker pool.
Tracks a presence.
Untracks a presence.
Updates a presence's metadata.
Link to this section Types
@type topic() :: String.t()
Link to this section Callbacks
Link to this section Functions
Returns a specification to start this module under a supervisor.
See Supervisor
.
Gets presences tracked under a given topic and key pair.
server_name
- The registered name of the tracker servertopic
- ThePhoenix.PubSub
topickey
- The key of the presence
Returns a lists of presence metadata.
examples
Examples
iex> Phoenix.Tracker.get_by_key(MyTracker, "lobby", "user1")
[{#PID<0.88.0>, %{name: "User 1"}, {#PID<0.89.0>, %{name: "User 1"}]
@spec graceful_permdown(atom()) :: :ok
Gracefully shuts down by broadcasting permdown to all replicas.
examples
Examples
iex> Phoenix.Tracker.graceful_permdown(MyTracker)
:ok
Lists all presences tracked under a given topic.
server_name
- The registered name of the tracker servertopic
- ThePhoenix.PubSub
topic
Returns a lists of presences in key/metadata tuple pairs.
examples
Examples
iex> Phoenix.Tracker.list(MyTracker, "lobby")
[{123, %{name: "user 123"}}, {456, %{name: "user 456"}}]
Starts a tracker pool.
tracker
- The tracker module implementing thePhoenix.Tracker
behaviourtracker_arg
- The argument to pass to the tracker handlerinit/1
pool_opts
- The list of options used to construct the shard pool
required-pool_opts
Required pool_opts
:
:name
- The name of the server, such as:MyApp.Tracker
This will also form the common prefix for all shard names:pubsub_server
- The name of the PubSub server, such as:MyApp.PubSub
optional-pool_opts
Optional pool_opts
:
:broadcast_period
- The interval in milliseconds to send delta broadcasts across the cluster. Default1500
:max_silent_periods
- The max integer of broadcast periods for which no delta broadcasts have been sent. Default10
(15s heartbeat):down_period
- The interval in milliseconds to flag a replica as temporarily down. Defaultbroadcast_period * max_silent_periods * 2
(30s down detection). Note: This must be at least 2x thebroadcast_period
.:permdown_period
- The interval in milliseconds to flag a replica as permanently down, and discard its state. Note: This must be at least greater than thedown_period
. Default1_200_000
(20 minutes):clock_sample_periods
- The numbers of heartbeat windows to sample remote clocks before collapsing and requesting transfer. Default2
:max_delta_sizes
- The list of delta generation sizes to keep before falling back to sending entire state. Defaults[100, 1000, 10_000]
.:log_level
- The log level to log events, defaults:debug
and can be disabled withfalse
:pool_size
- The number of tracker shards to launch. Default1
@spec track(atom(), pid(), topic(), term(), map()) :: {:ok, ref :: binary()} | {:error, reason :: term()}
Tracks a presence.
server_name
- The registered name of the tracker serverpid
- The Pid to tracktopic
- ThePhoenix.PubSub
topic for this presencekey
- The key identifying this presencemeta
- The map of metadata to attach to this presence
A process may be tracked multiple times, provided the topic and key pair are unique for any prior calls for the given process.
examples
Examples
iex> Phoenix.Tracker.track(MyTracker, self(), "lobby", u.id, %{stat: "away"})
{:ok, "1WpAofWYIAA="}
iex> Phoenix.Tracker.track(MyTracker, self(), "lobby", u.id, %{stat: "away"})
{:error, {:already_tracked, #PID<0.56.0>, "lobby", "123"}}
Untracks a presence.
server_name
- The registered name of the tracker serverpid
- The Pid to untracktopic
- ThePhoenix.PubSub
topic to untrack for this presencekey
- The key identifying this presence
All presences for a given Pid can be untracked by calling the
Phoenix.Tracker.untrack/2
signature of this function.
examples
Examples
iex> Phoenix.Tracker.untrack(MyTracker, self(), "lobby", u.id)
:ok
iex> Phoenix.Tracker.untrack(MyTracker, self())
:ok
@spec update(atom(), pid(), topic(), term(), map() | (map() -> map())) :: {:ok, ref :: binary()} | {:error, reason :: term()}
Updates a presence's metadata.
server_name
- The registered name of the tracker serverpid
- The Pid being trackedtopic
- ThePhoenix.PubSub
topic to update for this presencekey
- The key identifying this presencemeta
- Either a new map of metadata to attach to this presence, or a function. The function will receive the current metadata as input and the return value will be used as the new metadata
examples
Examples
iex> Phoenix.Tracker.update(MyTracker, self(), "lobby", u.id, %{stat: "zzz"})
{:ok, "1WpAofWYIAA="}
iex> Phoenix.Tracker.update(MyTracker, self(), "lobby", u.id, fn meta -> Map.put(meta, :away, true) end)
{:ok, "1WpAofWYIAA="}