#
# Created by Boyd Multerer on 2021-04-12
# Copyright © 2021 Kry10 Limited. All rights reserved.
#
defmodule Scenic.Assets.Stream do
@moduledoc """
Manage streaming assets (for now only compressed images and bitmaps) that are available
to all Scenes and ViewPorts.
The `Scenic.Assets.Stream` API gives to access to a running GenServer that manages an
`:ets` table and subscriptions to changes to named streams. This means that streaming
assets are available globally to all Scenes and ViewPorts.
You should be aware that if you have a GenServer that is rapidly updating a stream,
but no scene's are listening, then you are doing unnecessary work. If you have only a
single Scene in a single ViewPort listening to that stream, then create the stream
in the scene.
If you have multiple Scenes listening to a stream, or the same stream in multiple
ViewPorts, then create and update the stream in an independent GenServer that you
manage outside of Scenic.
"""
# behaviour
# @callback valid?(asset :: any) :: boolean
use GenServer
alias Scenic.Assets.Stream.Image
alias Scenic.Assets.Stream.Bitmap
@type asset :: Image.t() | Bitmap.t()
@type id :: String.t()
# import IEx
# ===========================================================================
defmodule Error do
@moduledoc false
defexception message: nil, error: nil, id: nil
end
# ============================================================================
# Client API
@doc """
Check if a named stream has been published.
Returns a boolean indicating if the stream is published.
"""
@spec exists?(id :: String.t()) :: boolean
def exists?(id) when is_bitstring(id) do
case :ets.lookup(__MODULE__, id) do
[{^id, _, _}] -> true
[] -> false
end
end
@doc """
Check if a named stream has been published.
Returns a `:ok` if the stream is published. Raises an error if it is not.
"""
@spec exists!(id :: String.t()) :: :ok
def exists!(id) when is_bitstring(id) do
case :ets.lookup(__MODULE__, id) do
[{^id, _, _}] -> :ok
[] -> raise Error, message: "Not Found: #{id}", error: :not_found, id: id
end
end
@doc """
Fetch the currently published asset in a named stream.
Returns `{:ok, asset}` on success.
Returns `{:error, :not_found}` if the stream is not available.
"""
@spec fetch(id :: String.t()) :: {:ok, asset :: asset()} | {:error, :not_found}
def fetch(id) when is_bitstring(id) do
case :ets.lookup(__MODULE__, id) do
[{_, asset, _}] -> {:ok, asset}
[] -> {:error, :not_found}
end
end
@doc """
Put a streamable asset into a named stream.
If the named stream does not exist yet, it is created. If it already exists, then
it's content is updated with the asset.
Returns `:ok` on success.
Note: Once a stream is create, the asset being updated must be the same type as the
asset that was originally created. I.e. you can't replace a Bitmap with an Image. This
will make more sense in the future as other assets types (audio) become supported.
The contents of the asset is (lightly) validated as it is put into the :ets table.
If the content is invalid, or a different type than what is already in the stream,
then it returns `{:error, :invalid, asset_type}`.
"""
@spec put(id :: String.t(), asset :: asset()) ::
:ok | {:error, atom} | {:error, atom, any}
def put(id, {type, _meta, _bin} = asset) do
case type.valid?(asset) do
true ->
case :ets.lookup(__MODULE__, id) do
[{_, ^asset, _}] ->
# no change. no work to do
:ok
[] ->
# asset key does not yet exist
true = :ets.insert(__MODULE__, {id, asset, self()})
GenServer.cast(__MODULE__, {:put, id})
[{_, {^type, _, _}, _}] ->
# is the same type
true = :ets.insert(__MODULE__, {id, asset, self()})
GenServer.cast(__MODULE__, {:put, id})
[{_, {type, _, _}, _}] ->
# is a different type. reject the put
{:error, :invalid, type}
end
false ->
{:error, :invalid, type}
end
end
@doc """
Same as put, but reverses the params order (making it pipe-able) and raises on failures.
Returns the asset on success.
"""
@spec put!(asset :: asset(), id :: String.t()) :: asset()
def put!({_type, _meta, _bin} = asset, id) do
:ok = put(id, asset)
asset
end
@doc """
Fully delete a named stream.
If you recreate the stream after deleting it, you can place any asset type into
the new stream.
"""
@spec delete(id :: String.t()) :: :ok
def delete(id) do
GenServer.cast(__MODULE__, {:delete, id})
end
@doc """
Subscribe to changes in a named stream.
Call this from a GenServer, typically a `Driver` or something you manage yourself.
When an asset stream is updated you will receive the following message.
`{{Stream, :put}, stream_type, id}`
You can match against stream_type to select certain kids of assets. Use the id
to fetch the contents of the asset.
When an asset stream is deleted, you will receive the following message.
`{{Stream, :delete}, stream_type, id}`
You can subscribe to an stream before it has been published. You will then start
receiving put messages when it is created. Your subscription will not end if the
stream is deleted.
"""
# subscribing to a streaming asset that doesn't exist yet, still succeeds and
# will start sending updates if it is published in the future
@spec subscribe(id :: String.t()) :: :ok
def subscribe(id) do
GenServer.cast(__MODULE__, {:subscribe, self(), id})
end
@doc """
Unsubscribe to changes in a named stream.
Once your process unsubscribes to a named stream, it will stop receiving all
messages related to it
"""
@spec unsubscribe(id :: String.t() | :all) :: :ok
def unsubscribe(id) do
GenServer.cast(__MODULE__, {:unsubscribe, self(), id})
end
# ============================================================================
# Server API
# --------------------------------------------------------
@doc false
def start_link(_) do
GenServer.start_link(__MODULE__, nil, name: __MODULE__)
end
# --------------------------------------------------------
@doc false
def init(nil) do
# create the table. Optimized for reads
__MODULE__ = :ets.new(__MODULE__, [:named_table, :public, {:read_concurrency, true}])
# the state is an empty {by_key, by_pid}
{:ok, {%{}, %{}}}
end
# --------------------------------------------------------
# a subscriber we are monitoring went down. Clean up.
@doc false
def handle_info({:DOWN, _ref, :process, pid, _reason}, state) do
{:noreply, do_unsubscribe(pid, :all, state)}
end
# --------------------------------------------------------
@doc false
def handle_cast({:delete, id}, {by_key, _by_pid} = state) do
with {:ok, {type_id, _meta, _bin}} <- fetch(id),
true <- :ets.delete(__MODULE__, id),
{:ok, subs} <- Map.fetch(by_key, id) do
send_subs(subs, :delete, type_id, id)
end
{:noreply, state}
end
# --------------------------------------------------------
@doc false
def handle_cast({:put, id}, {by_key, _by_pid} = state) do
# send a message to this id's subscribers
with {:ok, subs} <- Map.fetch(by_key, id),
{:ok, {type_id, _meta, _bin}} <- fetch(id) do
send_subs(subs, :put, type_id, id)
end
{:noreply, state}
end
def handle_cast({:subscribe, pid, id}, {by_key, by_pid}) do
# add this id to the pid's subs list
# also monitor this pid if necessary
by_pid =
case Map.fetch(by_pid, pid) do
{:ok, {mon, subs}} ->
Map.put(by_pid, pid, {mon, [id | subs]})
:error ->
# This is a new subscriber
mon = Process.monitor(pid)
Map.put(by_pid, pid, {mon, [id]})
end
# add this pid to the id's sub list
by_key = Map.put(by_key, id, [pid | Map.get(by_key, id, [])])
{:noreply, {by_key, by_pid}}
end
def handle_cast({:unsubscribe, pid, id}, state) do
state = do_unsubscribe(pid, id, state)
{:noreply, state}
end
defp do_unsubscribe(pid, :all, {_, by_pid} = state) do
case Map.fetch(by_pid, pid) do
{:ok, {_, subs}} -> Enum.reduce(subs, state, &do_unsubscribe(pid, &1, &2))
_ -> state
end
end
defp do_unsubscribe(pid, id, {by_key, by_pid}) do
# remove the pid from the by_key map
pids = Map.get(by_key, id, []) |> List.delete(pid)
by_key = Map.put(by_key, id, pids)
# remove the id from the by_pid map
by_pid =
case Map.fetch(by_pid, pid) do
{:ok, {mon, subs}} ->
subs
|> List.delete(id)
|> case do
[] ->
Process.demonitor(mon)
Map.delete(by_pid, pid)
subs ->
Map.put(by_pid, pid, {mon, subs})
end
_ ->
by_pid
end
{by_key, by_pid}
end
defp send_subs(subs, verb, type_id, id) do
Enum.each(subs, &Process.send(&1, {{__MODULE__, verb}, type_id, id}, []))
end
end
