#
# Created by Boyd Multerer on August 20, 2018.
# Changed to Scenic.PubSub on 2021-06-08
# Copyright © 2018-2021 Kry10 Limited. All rights reserved.
#
# Centralized channel data pub-sub with cache
# Was originally Scenic.Sensor.
defmodule Scenic.PubSub do
use GenServer
@moduledoc """
`Scenic.PubSub` is a combination pub/sub server and data cache.
It is intended to be the interface between sensors (or other data sources) and Scenic scenes.
## Why Scenic.PubSub
Sensors (or other data sources) and scenes often need to communicate, but tend to operate on different timelines.
Some sensors update fairly slowly or don't behave well when asked to get data at random times by multiple clients.
`Scenic.PubSub` is backed by a `GenServer` that collects data from a data source in a well-behaved manner,
yet is able to serve that data on demand or by subscription to many clients.
## Global Scope
It is important to note that `Scenic.PubSub` is global in scope. In other words, anything published
into `Scenic.PubSub` is visible to all `ViewPorts` and `Scenes`.
## Registering Data Sources
Before a process can start publishing data from a source, it must register a source id with `Scenic.PubSub`.
This source id should be an atom. This prevents other processes from stepping on that data and alerts any
subscribing processes that the data is coming online.
Scenic.PubSub.register( source_id )
The `source_id` parameter must be an atom that names the sensor. Subscribers will use this id to request
data or subscriptions to the source.
You can can also unregister data sources if they are no longer available.
Scenic.PubSub.unregister( source_id )
Simply exiting the data source process does also cleans up its registration.
## Publishing Data
When a sensor process publishes data, two things happen. First, that data is cached in an `:ets` table so
that future requests for that data from scenes happen quickly and don't need to bother the data
source process. Second, any processes that have subscribed to that source are sent a message containing the new data.
Scenic.PubSub.publish( source_id, value )
The `source_id` parameter must be the atom that was previously registered.
The `value` parameter can be anything that makes sense for the data source.
## Subscribing to a Data Source
Scenes (or any other process) can subscribe to a data source. They will receive messages when the source updates its data, comes online, or goes away.
Scenic.PubSub.subscribe( source_id )
The `source_id` parameter is the atom registered for the data source. Note that the name source does NOT
need to be registered when a listening process subscribes to it. When the source process eventually registers and
starts publishing data, the listening subscribers will be notified.
The subscribing process will then start receiving messages that can be handled with `handle_info/2`
event | message sent to subscribers
--- | ---
data published | `{{Scenic.PubSub, :data}, {source_id, value, timestamp}}`
source registered | `{{Scenic.PubSub, :registered}, {source_id, opts}}`
source unregistered | `{{Scenic.PubSub, :unregistered}, source_id}`
Scenes can also unsubscribe if they are no longer interested in updates.
Scenic.PubSub.unsubscribe( source_id )
## Other functions
Any process can get data from a source on demand, whether or not it is a subscriber.
Scenic.PubSub.get( source_id )
>> {:ok, data}
Any process can list the currently registered data sources.
Scenic.PubSub.list()
>> [{source_id, opts, pid}]
"""
# ets table names
@table __MODULE__
@name __MODULE__
@data {__MODULE__, :data}
@registered {__MODULE__, :registered}
@unregistered {__MODULE__, :unregistered}
defmodule Error do
@moduledoc false
defexception message: nil, source_id: nil
end
# ============================================================================
# client api
# --------------------------------------------------------
@doc """
Retrieve the cached data value for a named data source.
This data is pulled from an `:ets` table and does not put load on the data source itself.
## Parameters
* `source_id` an atom that is registered to a data source.
## Return Value
data
If the data source is either not registered, or has not yet published any data, get returns
nil
"""
@spec get(source_id :: atom) :: any | nil
def get(source_id) when is_atom(source_id) do
case :ets.lookup(@table, source_id) do
[{_key, data, _timestamp}] -> data
_ -> nil
end
end
# --------------------------------------------------------
@doc """
Retrieve the cached data value for a named data source.
Raises an error if the value is not registered
This data is pulled from an `:ets` table and does not put load on the data source itself.
## Parameters
* `source_id` an atom that is registered to a data source.
## Return Value
data
If the data source is either not registered, or has not yet published any data, get returns
nil
"""
@spec get!(source_id :: atom) :: any
def get!(source_id) when is_atom(source_id) do
case fetch(source_id) do
{:ok, data} -> data
_ -> raise Error, message: "#{inspect(source_id)} is not registered", source_id: source_id
end
end
# --------------------------------------------------------
@doc """
Retrieve the cached data for a named data source.
This data is pulled from an `:ets` table and does not put load on the data source itself.
## Parameters
* `source_id` an atom that is registered to a data source.
## Return Value
{:ok, {source_id, data, timestamp}}
* `source_id` is the atom representing the data source.
* `data` source_id whatever data the data source last published.
* `timestamp` is the time - from `:os.system_time(:micro_seconds)` - the last data was published.
If the data source is either not registered, or has not yet published any data, get returns
{:error, :no_data}
"""
@spec fetch(source_id :: atom) :: {:ok, any} | {:error, :not_found}
def fetch(source_id) when is_atom(source_id) do
case :ets.lookup(@table, source_id) do
[{_key, data, _timestamp}] ->
{:ok, data}
# no data
_ ->
{:error, :not_found}
end
end
# --------------------------------------------------------
@doc """
Retrieve the full cached data for a named data source.
This data is pulled from an `:ets` table and does not put load on the data source itself.
## Parameters
* `source_id` an atom that is registered to a data source.
## Return Value
{:ok, {source_id, data, timestamp}}
* `source_id` is the atom representing the data source.
* `data` source_id whatever data the data source last published.
* `timestamp` is the time - from `:os.system_time(:micro_seconds)` - the last data was published.
If the data source is either not registered, or has not yet published any data, get returns
{:error, :not_found}
"""
@spec query(source_id :: atom) :: {:ok, any} | {:error, :not_found}
def query(source_id) when is_atom(source_id) do
case :ets.lookup(@table, source_id) do
[data] ->
{:ok, data}
# no data
_ ->
{:error, :not_found}
end
end
# --------------------------------------------------------
@doc """
List the registered data sources.
## Return Value
`list/0` returns a list of registered data sources
[{source_id, version, description, pid}]
* `source_id` is the atom representing the data source.
* `opts` options list of metadata about the data source.
* `:version` is the version string supplied by the data source during registration.
* `:description` is the description string supplied by the data source during registration.
* `:registered_at` The system time the data source was registered at.
* `pid` is the pid of the data source's process.
"""
@spec list() :: [{atom, Keyword.t(), pid}]
def list() do
:ets.match(@table, {{:registration, :"$1"}, :"$2", :"$3"})
|> Enum.map(fn [key, opts, pid] -> {key, opts, pid} end)
end
# --------------------------------------------------------
@doc """
Publish a data point from a data source.
When a data source uses `publish/2` to publish data, that data is recorded in the
cache and a
{{Scenic.PubSub, :data}, {source_id, my_value, timestamp}}
message is sent to each subscriber. The timestamp is the current time in microseconds as returned
from `:os.system_time(:micro_seconds)`.
## Parameters
* `source_id` an atom that is registered to a data source.
* `data` the data to publish.
## Return Value
On success, returns `:ok`
It returns `{:error, :not_registered}` if the caller is not the
registered process for the data source.
"""
@spec publish(source_id :: atom, data :: any) :: :ok
def publish(source_id, data) when is_atom(source_id) do
timestamp = :os.system_time(:micro_seconds)
pid = self()
# enforce that this is coming from the registered data source pid
case :ets.lookup(@table, {:registration, source_id}) do
[{_, _, ^pid}] ->
send(@name, {:put_data, source_id, data, timestamp})
:ok
# no data
_ ->
{:error, :not_registered}
end
end
# --------------------------------------------------------
@doc """
Subscribe the calling process to receive events about a data source.
The messages the subscriber will start receiving about a data source are:
event | message sent to subscribers
--- | ---
data published | `{{Scenic.PubSub, :data}, {source_id, value, timestamp}}`
source registered | `{{Scenic.PubSub, :registered}, {source_id, opts}}`
source unregistered | `{{Scenic.PubSub, :unregistered}, source_id}`
## Parameters
* `source_id` an atom that is registered to a data source.
## Return Value
On success, returns `:ok`
"""
@spec subscribe(source_id :: atom) :: :ok
def subscribe(source_id) when is_atom(source_id) do
GenServer.call(@name, {:subscribe, source_id, self()})
end
# --------------------------------------------------------
@doc """
Unsubscribe the calling process from receive events about a data source.
The caller will stop receiving events about a data source
## Parameters
* `source_id` an atom that is registered to a data source.
## Return Value
Returns `:ok`
"""
@spec unsubscribe(source_id :: atom) :: :ok
def unsubscribe(source_id) when is_atom(source_id) do
send(@name, {:unsubscribe, source_id, self()})
:ok
end
# --------------------------------------------------------
@register_opts_schema [
version: [type: :string, doc: "Data format version"],
description: [type: :string, doc: "Your appropriate description"]
]
@doc """
Register the calling process as a data source for the named id.
## Parameters
* `source_id` the data source being registered.
* `opts` optional information about the data source.
Supported options:\n#{NimbleOptions.docs(@register_opts_schema)}
## Return Value
On success, returns `{:ok, source_id}`
If `source_id` is already registered to another process, it returns
{:error, :already_registered}
"""
@spec register(source_id :: atom, opts :: Keyword.t()) ::
{:ok, atom} | {:error, :already_registered}
def register(source_id, opts \\ []) when is_atom(source_id) do
opts = Enum.into(opts, [])
case NimbleOptions.validate(opts, @register_opts_schema) do
{:ok, opts} -> opts
{:error, error} -> raise Error, message: error, source_id: source_id
end
opts = Keyword.put(opts, :registered_at, :os.system_time(:micro_seconds))
GenServer.call(
@name,
{:register, source_id, opts, self()}
)
end
# --------------------------------------------------------
@doc """
Unregister the calling process as a data source for a data source.
## Parameters
* `source_id` the data source being registered.
## Return Value
Returns `:ok`
"""
@spec unregister(source_id :: atom) :: :ok
def unregister(source_id) when is_atom(source_id) do
send(@name, {:unregister, source_id, self()})
:ok
end
# ============================================================================
# internal api
# --------------------------------------------------------
@doc false
def start_link(_) do
GenServer.start_link(__MODULE__, :ok, name: @name)
end
# --------------------------------------------------------
@doc false
def init(:ok) do
# set up the initial state
state = %{
data_table_id: :ets.new(@table, [:named_table]),
subs_id: %{},
subs_pid: %{}
}
# trap exits so we don't just crash when a subscriber goes away
Process.flag(:trap_exit, true)
{:ok, state}
end
# ============================================================================
# --------------------------------------------------------
# a data source (or whatever) is putting data
@doc false
# the client api enforced the pid check
# yes, you could get around that by sending this message directly
# not best-practice, but is an escape valve.
# timestamp should be from :os.system_time(:micro_seconds)
def handle_info({:put_data, source_id, data, timestamp}, state) do
:ets.insert(@table, {source_id, data, timestamp})
send_subs(source_id, @data, {source_id, data, timestamp}, state)
{:noreply, state}
end
# --------------------------------------------------------
@doc false
def handle_info({:unsubscribe, source_id, pid}, state) do
{:noreply, unsubscribe(pid, source_id, state)}
end
# ============================================================================
# handle linked processes going down
# --------------------------------------------------------
def handle_info({:EXIT, pid, _reason}, state) do
# unsubscribe everything this pid was listening to
state = do_unsubscribe(pid, :all, state)
# if this pid was registered as a data source, unregister it
:ets.match(@table, {{:registration, :"$1"}, :_, :_, pid})
|> Enum.each(fn [id] -> do_unregister(id, pid, state) end)
{:noreply, state}
end
# --------------------------------------------------------
@doc false
def handle_info({:unregister, source_id, pid}, state) do
do_unregister(source_id, pid, state)
{:noreply, state}
end
# ============================================================================
# CALLs - mostly for postive confirmation of sign-up style things
# --------------------------------------------------------
@doc false
def handle_call({:subscribe, source_id, pid}, _from, state) do
{reply, state} = do_subscribe(pid, source_id, state)
# send the already-set value if one is set
case query(source_id) do
{:ok, data} -> send(pid, {@data, data})
_ -> :ok
end
{:reply, reply, state}
end
# --------------------------------------------------------
@doc false
# handle data source registration
def handle_call({:register, source_id, opts, pid}, _from, state) do
key = {:registration, source_id}
{reply, state} =
case :ets.lookup(@table, key) do
# registered to pid - ok to change
[{_, _, ^pid}] ->
do_register(pid, source_id, opts, state)
# previously crashed
[{_, _, nil}] ->
do_register(pid, source_id, opts, state)
# registered to other. fail
[_] ->
{{:error, :already_registered}, state}
[] ->
do_register(pid, source_id, opts, state)
end
{:reply, reply, state}
end
# ============================================================================
# handle data source registrations
# --------------------------------------------------------
defp do_register(pid, source_id, opts, state) do
key = {:registration, source_id}
:ets.insert(@table, {key, opts, pid})
# link the data source
Process.link(pid)
# alert the subscribers
send_subs(source_id, @registered, {source_id, opts}, state)
# reply is sent back to the data source
{{:ok, source_id}, state}
end
# --------------------------------------------------------
defp do_unregister(source_id, pid, state) do
reg_key = {:registration, source_id}
# first, get the registration and confirm this pid is registered
case :ets.lookup(@table, reg_key) do
[{_, _, ^pid}] ->
# alert the subscribers
send_subs(source_id, @unregistered, source_id, state)
# delete the table entries
:ets.delete(@table, reg_key)
:ets.delete(@table, source_id)
unlink_pid(pid, state)
:ok
# no registered. do nothing
_ ->
:ok
end
end
# ============================================================================
# handle client subscriptions
# --------------------------------------------------------
@spec do_subscribe(pid :: GenServer.server(), source_id :: atom, state :: map) :: any
defp do_subscribe(pid, source_id, %{subs_id: subs_id, subs_pid: subs_pid} = state) do
# record the subscription
subs_id =
Map.put(
subs_id,
source_id,
[pid | Map.get(subs_id, source_id, [])] |> Enum.uniq()
)
subs_pid =
Map.put(
subs_pid,
pid,
[source_id | Map.get(subs_pid, pid, [])] |> Enum.uniq()
)
# make sure the subscriber is linked
Process.link(pid)
{:ok, %{state | subs_id: subs_id, subs_pid: subs_pid}}
end
# --------------------------------------------------------
@spec do_unsubscribe(pid :: GenServer.server(), source_id :: atom, state :: map) :: any
defp do_unsubscribe(pid, :all, %{subs_pid: subs_pid} = state) do
Map.get(subs_pid, pid, [])
|> Enum.reduce(state, &unsubscribe(pid, &1, &2))
end
# --------------------------------------------------------
defp unsubscribe(pid, source_id, %{subs_id: subs_id, subs_pid: subs_pid} = state) do
# clean up the subs for a given source_id
subs_by_id =
Map.get(subs_id, source_id, [])
|> Enum.reject(fn sub_pid -> sub_pid == pid end)
subs_id = Map.put(subs_id, source_id, subs_by_id)
# part two
subs_by_pid =
Map.get(subs_pid, pid, [])
|> Enum.reject(fn sub_id -> sub_id == source_id end)
subs_pid = Map.put(subs_pid, pid, subs_by_pid)
state = %{state | subs_id: subs_id, subs_pid: subs_pid}
# if pid no longer subscribed to anything, then some further cleanup
state =
case subs_by_pid do
[] ->
{_, state} = pop_in(state, [:subs_pid, pid])
state
_ ->
state
end
# if channel has no subscribers, then some further cleanup
state =
case subs_by_id do
[] ->
{_, state} = pop_in(state, [:subs_id, source_id])
state
_ ->
state
end
# does the right thing. only unlinks if no longer subscribing
# to anything and is not a channel
unlink_pid(pid, state)
state
end
# --------------------------------------------------------
defp send_subs(source_id, verb, msg, %{subs_id: subs}) do
msg = {verb, msg}
subs
|> Map.get(source_id, [])
|> Enum.each(&send(&1, msg))
end
# --------------------------------------------------------
# only unlink a pid if it is not a registered channel AND it
# has no subscriptions. return the state
defp unlink_pid(pid, %{subs_pid: subs_pid}) do
no_subs =
case subs_pid[pid] do
nil -> true
[] -> true
_ -> false
end
not_sensor =
case :ets.match(@table, {{:registration, :"$1"}, :_, pid}) do
[] -> true
_ -> false
end
if no_subs && not_sensor do
Process.unlink(pid)
end
end
end
