[RTC-220] Server notifications

README.md

@@ -9,6 +9,7 @@ Elixir server SDK for [Jellyfish](https://github.com/jellyfish-dev/jellyfish).
 Currently it allows for:
 
 - making API calls to Jellyfish server (QoL wrapper for HTTP requests)
+- listening to Jellyfish server notifications via WebSocket
 
 ## Installation
 
@@ -24,10 +25,15 @@ end
 
 ## Usage
 
-Make API calls to Jellyfish (authentication required, for more information see [Jellyfish docs](https://jellyfish-dev.github.io/jellyfish-docs/getting_started/authentication)):
+Make API calls to Jellyfish (authentication required, for more information see [Jellyfish docs](https://jellyfish-dev.github.io/jellyfish-docs/getting_started/authentication))
+and receive server notifications:
 
 ```elixir
-client = Jellyfish.Client.new("http://address-of-your-server.com", "your-jellyfish-token")
+# start process responsible for receiving notifications
+{:ok, _pid} = Jellyfish.Notifier.start(server_address: "ws://address-of-your-server.com", server_api_key: "your-jellyfish-token")
+
+# create HTTP client instance
+client = Jellyfish.Client.new(server_address: "http://address-of-your-server.com", server_api_key: "your-jellyfish-token")
 
 # Create room
 {:ok, %Jellyfish.Room{id: room_id}} = Jellyfish.Room.create(client, max_peers: 10)
@@ -38,6 +44,10 @@ room_id
 # Add peer
 {:ok, %Jellyfish.Peer{id: peer_id}, peer_token} = Jellyfish.Room.add_peer(client, room_id, "webrtc")
 
+receive do
+  {:jellyfish, {:peer_connected, ^room_id, ^peer_id}} -> # handle the notification
+end
+
 # Delete peer
 :ok = Jellyfish.Room.delete_peer(client, room_id, peer_id)
 ```

lib/jellyfish/client.ex

@@ -17,6 +17,8 @@ defmodule Jellyfish.Client do
   For the list of supported Tesla adapters refer to [Tesla docs](https://hexdocs.pm/tesla/readme.html#adapters).
   """
 
+  alias Jellyfish.Utils
+
   @enforce_keys [
     :http_client
   ]
@@ -26,45 +28,46 @@ defmodule Jellyfish.Client do
           http_client: Tesla.Client.t()
         }
 
-  @doc """
-  Creates a new instance of `t:Jellyfish.Client.t/0`.
+  @typedoc """
+  Options needed to open connection to Jellyfish server.
 
-  ## Parameters
+  * `:server_address` - url or IP address of the Jellyfish server instance.
+  * `:server_api_token` - token used for authorizing HTTP requests and WebSocket connection.
+  It's the same token as the one configured in Jellyfish.
+  * `:secure?` - if `true`, use HTTPS and WSS instead of HTTP and WS, `false` by default.
 
-    * `address` - url or IP address of the Jellyfish server instance
-    * `server_api_token` - token used for authorizing HTTP requests. It's the same
-    token as the one configured in Jellyfish.
+  When an option is not explicily passed, value set in `config.exs` is used:
+  ```
+  # in config.exs
+  config :jellyfish_server_sdk, 
+    server_address: "you-jellyfish-server-address.com",
+    server_api_token: "your-jellyfish-token",
+    secure?: true
+  ```
   """
-  @spec new(String.t(), String.t()) :: t()
-  def new(address, server_api_token), do: build_client(address, server_api_token)
+  @type connection_options :: [
+          server_address: String.t(),
+          server_api_token: String.t(),
+          secure?: boolean()
+        ]
 
   @doc """
   Creates a new instance of `t:Jellyfish.Client.t/0`.
 
-  Uses token set in `config.exs`. To explicitly pass the token, see `new/2`.
-  ```
-  # in config.exs
-  config :jellyfish_server_sdk, server_api_token: "your-jellyfish-token"
-
-  client = Jellyfish.Client.new("http://address-of-your-server.com")
-  ```
-
-  See `new/2` for description of parameters.
+  For information about options, see `t:connection_options/0`.
   """
-  @spec new(String.t()) :: t()
-  def new(address) do
-    server_api_token = Application.fetch_env!(:jellyfish_server_sdk, :server_api_token)
-    build_client(address, server_api_token)
-  end
+  @spec new(connection_options()) :: t()
+  def new(opts \\ []) do
+    {address, api_token, secure?} = Utils.get_options_or_defaults(opts)
+    address = if secure?, do: "https://#{address}", else: "http://#{address}"
+    adapter = Application.get_env(:jellyfish_server_sdk, :tesla_adapter, Tesla.Adapter.Mint)
 
-  defp build_client(address, server_api_token) do
     middleware = [
       {Tesla.Middleware.BaseUrl, address},
-      {Tesla.Middleware.BearerAuth, token: server_api_token},
+      {Tesla.Middleware.BearerAuth, token: api_token},
       Tesla.Middleware.JSON
     ]
 
-    adapter = Application.get_env(:jellyfish_server_sdk, :tesla_adapter, Tesla.Adapter.Mint)
     http_client = Tesla.client(middleware, adapter)
 
     %__MODULE__{http_client: http_client}

lib/jellyfish/component.ex

@@ -6,7 +6,7 @@ defmodule Jellyfish.Component do
   For more information refer to [Jellyfish documentation](https://www.membrane.stream)
   """
 
-  alias Jellyfish.Exception.ResponseStructureError
+  alias Jellyfish.Exception.StructureError
 
   @enforce_keys [
     :id,
@@ -54,7 +54,7 @@ defmodule Jellyfish.Component do
         }
 
       _other ->
-        raise ResponseStructureError
+        raise StructureError
     end
   end
 end

lib/jellyfish/exception.ex

@@ -1,17 +1,31 @@
 defmodule Jellyfish.Exception do
   @moduledoc false
 
-  defmodule ResponseStructureError do
+  defmodule StructureError do
     defexception [:message]
 
     @impl true
     def exception(_opts) do
       msg = """
-      Received server response with unexpected structure.
+      Received server response or notification with unexpected structure.
       Make sure you are using correct combination of Jellyfish and SDK versions.
       """
 
       %__MODULE__{message: msg}
     end
   end
+
+  defmodule ProtocolPrefixError do
+    defexception [:message]
+
+    @impl true
+    def exception(_opts) do
+      msg = """
+      Passed address starts with protocol prefix, like "http://" or "https://", which is undesired.
+      To use SSL, set `secure?: true` option in `config.exs` or pass this option to called function.
+      """
+
+      %__MODULE__{message: msg}
+    end
+  end
 end

lib/jellyfish/notifier.ex

@@ -0,0 +1,154 @@
+defmodule Jellyfish.Notifier do
+  @moduledoc """
+  Module defining a process responsible for establishing
+  WebSocket connection and receiving notifications form Jellyfish server.
+
+  ```
+  iex> {:ok, pid} = Jellyfish.Notifier.start(server_address: "address-of-jellyfish-server.com", server_api_token: "your-jellyfish-token")
+  {:ok, #PID<0.301.0>}
+
+  # here add a room and a peer using functions from `Jellyfish.Room` module
+  # you should receive a notification after the peer established connection
+
+  iex> flush()
+  {:jellyfish,
+   {:peer_connected, "21604fbe-8ac8-44e6-8474-98b5f50f1863",
+    "ae07f94e-0887-44c3-81d5-bfa9eac96252"}}
+  :ok
+  ```
+  """
+
+  use WebSockex
+
+  alias Jellyfish.{Client, Utils}
+  alias Jellyfish.Exception.StructureError
+
+  @auth_timeout 2000
+
+  @doc """
+  Starts the Notifier process and connects to Jellyfish. 
+
+  Acts like `start/1` but links to the calling process.
+
+  See `start/1` for more information.
+  """
+  @spec start_link(Client.connection_options()) :: {:ok, pid()} | {:error, term()}
+  def start_link(opts \\ []) do
+    connect(:start_link, opts)
+  end
+
+  @doc """
+  Starts the Notifier process and connects to Jellyfish. 
+
+  Received notifications are send to the calling process in
+  a form of `{:jellyfish, msg}`, where `msg` is 
+  `type` or `{type, room_id}` or `{type, room_id, (peer/component)_id}`.
+  Refer to [Jellyfish docs](https://jellyfish-dev.github.io/jellyfish-docs/) to learn more about server notifications.
+
+  For information about options, see `t:Jellyfish.Client.connection_options/0`.
+  """
+  @spec start(Client.connection_options()) :: {:ok, pid()} | {:error, term()}
+  def start(opts \\ []) do
+    connect(:start, opts)
+  end
+
+  @impl true
+  def handle_frame({:text, msg}, state) do
+    with {:ok, decoded_msg} <- Jason.decode(msg),
+         %{"type" => "controlMessage", "data" => data} <- decoded_msg,
+         {:ok, notification} <- decode_notification(data) do
+      # notification will be either {type, room_id, peer_id/component_id},
+      # {type, room_id} or just type
+      send(state.receiver_pid, {:jellyfish, notification})
+    else
+      _other -> raise StructureError
+    end
+
+    {:ok, state}
+  end
+
+  @impl true
+  def handle_cast(_msg, state) do
+    # ignore incoming messages
+    {:ok, state}
+  end
+
+  @impl true
+  def terminate({:remote, 1000, "invalid token"}, state) do
+    send(state.receiver_pid, {:jellyfish, :invalid_token})
+  end
+
+  @impl true
+  def terminate(_reason, state) do
+    send(state.receiver_pid, {:jellyfish, :disconnected})
+  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}"
+    state = %{receiver_pid: self()}
+
+    auth_msg =
+      %{type: "controlMessage", data: %{type: "authRequest", token: api_token}}
+      |> Jason.encode!()
+
+    with {:ok, pid} <-
+           apply(WebSockex, fun, ["#{address}/socket/server/websocket", __MODULE__, state]),
+         :ok <- WebSockex.send_frame(pid, {:text, auth_msg}) do
+      receive do
+        {:jellyfish, :authenticated} ->
+          {:ok, pid}
+
+        {:jellyfish, :invalid_token} ->
+          {:error, :invalid_token}
+      after
+        @auth_timeout ->
+          Process.exit(pid, :normal)
+          {:error, :authentication_timeout}
+      end
+    else
+      {:error, _reason} = error -> error
+    end
+  end
+
+  defp decode_notification(%{"type" => type, "roomId" => room_id, "id" => id}) do
+    decoded_type =
+      case type do
+        "authenticated" -> :authenticated
+        "peerConnected" -> :peer_connected
+        "peerDisconnected" -> :peer_disconected
+        "componentCrashed" -> :component_crashed
+        _other -> nil
+      end
+
+    if is_nil(decoded_type) do
+      {:error, :invalid_type}
+    else
+      {:ok, {decoded_type, room_id, id}}
+    end
+  end
+
+  defp decode_notification(%{"type" => type, "room_id" => id}) do
+    decoded_type =
+      case type do
+        "roomCrashed" -> :room_crashed
+        _other -> nil
+      end
+
+    if is_nil(decoded_type),
+      do: {:error, :invalid_type},
+      else: {:ok, {decoded_type, id}}
+  end
+
+  defp decode_notification(%{"type" => type}) do
+    decoded_type =
+      case type do
+        "authenticated" -> :authenticated
+        _other -> nil
+      end
+
+    if is_nil(decoded_type), do: {:error, :invalid_type}, else: {:ok, decoded_type}
+  end
+
+  defp decode_notification(_other), do: {:error, :invalid_type}
+end

lib/jellyfish/peer.ex

@@ -7,7 +7,7 @@ defmodule Jellyfish.Peer do
   For more information refer to [Jellyfish documentation](https://www.membrane.stream)
   """
 
-  alias Jellyfish.Exception.ResponseStructureError
+  alias Jellyfish.Exception.StructureError
 
   @enforce_keys [
     :id,
@@ -49,7 +49,7 @@ defmodule Jellyfish.Peer do
         }
 
       _other ->
-        raise ResponseStructureError
+        raise StructureError
     end
   end
 end