defmodule Fishjam.WSNotifier do
@moduledoc """
Module defining a process responsible for establishing
WebSocket connection and receiving events from Fishjam server.
First, [configure the connection options](README.md#fishjam-connection-configuration).
```
# Start the Notifier
iex> {:ok, notifier} = Fishjam.WSNotifier.start()
{:ok, #PID<0.301.0>}
```
```
# Subscribe current process to server notifications.
iex> :ok = Fishjam.WSNotifier.subscribe_server_notifications(notifier)
# here add a room and a peer using functions from `Fishjam.Room` module
# you should receive a notification after the peer established connection
iex> flush()
{:fishjam, %Fishjam.Notification.PeerConnected{
room_id: "21604fbe-8ac8-44e6-8474-98b5f50f1863",
peer_id: "ae07f94e-0887-44c3-81d5-bfa9eac96252"
}}
:ok
```
When starting the Notifier, you can provide the name under which the process will be registered.
```
iex> {:ok, notifier} = Fishjam.WSNotifier.start_link(name: Fishjam.WSNotifier)
```
"""
use WebSockex
require Logger
alias Fishjam.Utils
alias Fishjam.{Notification, ServerMessage}
alias Fishjam.ServerMessage.{
Authenticated,
AuthRequest,
MetricsReport,
SubscribeRequest,
SubscribeResponse
}
@auth_timeout 2000
@subscribe_timeout 5000
@typedoc """
The reference to the `Notifier` process.
"""
@type notifier() :: GenServer.server()
@typedoc """
Connection options used to connect to Fishjam server.
"""
@type options() :: [
server_address: String.t(),
server_api_token: String.t(),
secure?: boolean(),
name: GenServer.name()
]
@doc """
Starts the Notifier process and connects to Fishjam.
Acts like `start/1` but links to the calling process.
See `start/1` for more information.
"""
@spec start_link(options()) :: {:ok, pid()} | {:error, term()}
def start_link(opts \\ []) do
connect(:start_link, opts)
end
@doc """
Starts the Notifier process and connects to Fishjam.
To learn how to receive notifications, see `subscribe/3`.
For information about options, see `t:Fishjam.Client.connection_options/0`.
"""
@spec start(options()) :: {:ok, pid()} | {:error, term()}
def start(opts \\ []) do
connect(:start, opts)
end
@doc """
Subscribes the process to receive server notifications from all the rooms.
Notifications are sent to the process in a form of `{:fishjam, msg}`,
where `msg` is one of structs defined under "Fishjam.Notification" section in the docs,
for example `{:fishjam, %Fishjam.Notification.RoomCrashed{room_id: "some_id"}}`.
"""
@spec subscribe_server_notifications(notifier()) :: :ok | {:error, atom()}
def subscribe_server_notifications(notifier) do
WebSockex.cast(notifier, {:subscribe_server_notifications, self()})
receive do
{:fishjam, {:subscribe_answer, :ok}} -> :ok
{:error, _reason} = error -> error
after
@subscribe_timeout -> {:error, :timeout}
end
end
@doc """
Subscribes the process to the WebRTC metrics from all the rooms.
Metrics are periodically sent to the process in a form of `{:fishjam, metrics_report}`,
where `metrics_report` is the `Fishjam.MetricsReport` struct.
"""
@spec subscribe_metrics(notifier()) :: :ok | {:error, :timeout}
def subscribe_metrics(notifier) do
WebSockex.cast(notifier, {:subscribe_metrics, self()})
receive do
{:fishjam, {:subscribe_answer, :ok}} -> :ok
after
@subscribe_timeout -> {:error, :timeout}
end
end
@impl true
def handle_cast({:subscribe_server_notifications, pid}, state) do
{request, state} = subscribe_request(:server_notification, pid, state)
{:reply, {:binary, ServerMessage.encode(request)}, state}
end
def handle_cast({:subscribe_metrics, pid}, state) do
{request, state} = subscribe_request(:metrics, pid, state)
{:reply, {:binary, ServerMessage.encode(request)}, state}
end
@impl true
def handle_frame({:binary, msg}, state) do
case ServerMessage.decode(msg) do
%ServerMessage{content: {_type, notification}} ->
handle_notification(notification, state)
%ServerMessage{content: nil, __unknown_fields__: _binary} ->
Logger.warning(
"Can't decode received notification. This probably means that fishjam is using a different version of protobuffs."
)
{:ok, state}
end
end
@impl true
def handle_info({:DOWN, _ref, :process, pid, _reason}, state) do
state =
[:server_notification, :metrics]
|> Enum.reduce(state, fn event_type, state ->
update_in(state.subscriptions[event_type], &MapSet.delete(&1, pid))
end)
{:ok, state}
end
@impl true
def terminate({:remote, 1000, "invalid token"}, state) do
send(state.caller_pid, {:fishjam, :invalid_token})
end
@impl true
def terminate(_reason, _state) do
:ok
end
defp connect(fun, opts) do
{address, api_token, secure?} = Utils.get_options_or_defaults(opts)
address = if secure?, do: "wss://#{address}", else: "ws://#{address}"
empty_subscriptions = %{server_notification: MapSet.new(), metrics: MapSet.new()}
state = %{
caller_pid: self(),
subscriptions: empty_subscriptions,
pending_subscriptions: empty_subscriptions
}
auth_msg =
%ServerMessage{content: {:auth_request, %AuthRequest{token: api_token}}}
|> ServerMessage.encode()
websockex_opts = Keyword.take(opts, [:name])
with {:ok, ws} <-
apply(WebSockex, fun, [
"#{address}/socket/server/websocket",
__MODULE__,
state,
websockex_opts
]),
:ok <- WebSockex.send_frame(ws, {:binary, auth_msg}) do
receive do
{:fishjam, :authenticated} ->
{:ok, ws}
{:fishjam, :invalid_token} ->
{:error, :invalid_token}
after
@auth_timeout ->
Process.exit(ws, :normal)
{:error, :authentication_timeout}
end
else
{:error, _reason} = error ->
error
end
end
defp handle_notification(%Authenticated{}, state) do
send(state.caller_pid, {:fishjam, :authenticated})
{:ok, state}
end
defp handle_notification(%SubscribeResponse{event_type: proto_event_type}, state) do
event_type = from_proto_event_type(proto_event_type)
{pending_subscriptions, state} = pop_in(state.pending_subscriptions[event_type])
state =
if Enum.empty?(pending_subscriptions) do
state
else
pending_subscriptions
|> Enum.reduce(state, fn
pid, state ->
send(pid, {:fishjam, {:subscribe_answer, :ok}})
update_in(state.subscriptions[event_type], &MapSet.put(&1, pid))
end)
end
{:ok, state}
end
defp handle_notification(%MetricsReport{metrics: metrics}, state) do
notification = %Fishjam.MetricsReport{metrics: Jason.decode!(metrics)}
state.subscriptions.metrics
|> Enum.each(fn pid ->
send(pid, {:fishjam, notification})
end)
{:ok, state}
end
defp handle_notification(%{room_id: _room_id} = message, state) do
state.subscriptions.server_notification
|> Enum.each(&send(&1, {:fishjam, Notification.to_notification(message)}))
{:ok, state}
end
defp subscribe_request(event_type, caller_pid, state) do
request = %ServerMessage{
content:
{:subscribe_request, %SubscribeRequest{event_type: to_proto_event_type(event_type)}}
}
state = update_in(state.pending_subscriptions[event_type], &MapSet.put(&1, caller_pid))
{request, state}
end
defp from_proto_event_type(:EVENT_TYPE_SERVER_NOTIFICATION), do: :server_notification
defp from_proto_event_type(:EVENT_TYPE_METRICS), do: :metrics
defp to_proto_event_type(:server_notification), do: :EVENT_TYPE_SERVER_NOTIFICATION
defp to_proto_event_type(:metrics), do: :EVENT_TYPE_METRICS
end
