Gather and handle relay candidates

.credo.exs

@@ -122,14 +122,14 @@
           #
           {Credo.Check.Refactor.Apply, []},
           {Credo.Check.Refactor.CondStatements, []},
-          {Credo.Check.Refactor.CyclomaticComplexity, []},
+          {Credo.Check.Refactor.CyclomaticComplexity, [max_complexity: 10]},
           {Credo.Check.Refactor.FunctionArity, []},
           {Credo.Check.Refactor.LongQuoteBlocks, []},
           {Credo.Check.Refactor.MatchInCondition, []},
           {Credo.Check.Refactor.MapJoin, []},
           {Credo.Check.Refactor.NegatedConditionsInUnless, []},
           {Credo.Check.Refactor.NegatedConditionsWithElse, []},
-          {Credo.Check.Refactor.Nesting, []},
+          {Credo.Check.Refactor.Nesting, [max_nesting: 3]},
           {Credo.Check.Refactor.UnlessWithElse, []},
           {Credo.Check.Refactor.WithClauses, []},
           {Credo.Check.Refactor.FilterFilter, []},
@@ -199,7 +199,7 @@
           {Credo.Check.Warning.MapGetUnsafePass, []},
           {Credo.Check.Warning.MixEnv, []},
           {Credo.Check.Warning.UnsafeToAtom, []},
-          
+
 
           # {Credo.Check.Refactor.MapInto, []},
 

example/peer.exs

@@ -100,7 +100,7 @@ defmodule Peer do
           {_, _, _, _} -> true
           {_, _, _, _, _, _, _, _} -> false
         end,
-        stun_servers: ["stun:stun.l.google.com:19302"]
+        ice_servers: [%ICEServer{url: "stun:stun.l.google.com:19302"}]
       )
 
     {:ok, ufrag, passwd} = ICEAgent.get_local_credentials(pid)

lib/ex_ice/candidate_pair.ex

@@ -3,21 +3,19 @@ defmodule ExICE.CandidatePair do
   ICE candidate pair representation.
   """
 
-  alias ExICE.Candidate
-
   @type state() :: :waiting | :in_progress | :succeeded | :failed | :frozen
 
   @type t() :: %__MODULE__{
           id: integer(),
-          local_cand: Candidate.t(),
+          local_cand_id: integer(),
           nominated?: boolean(),
           priority: non_neg_integer(),
-          remote_cand: Candidate.t(),
+          remote_cand_id: integer(),
           state: state(),
           valid?: boolean()
         }
 
-  @enforce_keys [:id, :local_cand, :remote_cand, :priority]
+  @enforce_keys [:id, :local_cand_id, :remote_cand_id, :priority]
   defstruct @enforce_keys ++
               [
                 nominated?: false,

lib/ex_ice/ice_agent.ex

@@ -48,7 +48,10 @@ defmodule ExICE.ICEAgent do
   This behavior can be overwritten using the following options.
 
   * `ip_filter` - filter applied when gathering local candidates
-  * `stun_servers` - list of STUN servers
+  * `ice_servers` - list of STUN/TURN servers
+  * `ice_transport_policy` - candidate types to be used.
+    * `all` - all ICE candidates will be considered (default).
+    * `relay` - only relay candidates will be considered.
   * `on_gathering_state_change` - where to send gathering state change notifications. Defaults to a process that spawns `ExICE`.
   * `on_connection_state_change` - where to send connection state change notifications. Defaults to a process that spawns `ExICE`.
   * `on_data` - where to send data. Defaults to a process that spawns `ExICE`.
@@ -59,7 +62,14 @@ defmodule ExICE.ICEAgent do
   """
   @type opts() :: [
           ip_filter: (:inet.ip_address() -> boolean),
-          stun_servers: [String.t()],
+          ice_servers: [
+            %{
+              :url => String.t(),
+              optional(:username) => String.t(),
+              optional(:credential) => String.t()
+            }
+          ],
+          ice_transport_policy: :all | :relay,
           on_gathering_state_change: pid() | nil,
           on_connection_state_change: pid() | nil,
           on_data: pid() | nil,
@@ -347,6 +357,12 @@ defmodule ExICE.ICEAgent do
     {:noreply, %{state | ice_agent: ice_agent}}
   end
 
+  @impl true
+  def handle_info({:ex_turn, ref, msg}, state) do
+    ice_agent = ExICE.Priv.ICEAgent.handle_ex_turn_msg(state.ice_agent, ref, msg)
+    {:noreply, %{state | ice_agent: ice_agent}}
+  end
+
   @impl true
   def handle_info(msg, state) do
     Logger.warning("Got unexpected msg: #{inspect(msg)}")

lib/ex_ice/priv/candidate.ex

@@ -27,9 +27,6 @@ defmodule ExICE.Priv.Candidate do
   @callback send_data(t(), :inet.ip_address(), :inet.port_number(), binary()) ::
               {:ok, t()} | {:error, term(), t()}
 
-  @callback receive_data(t(), :inet.ip_address(), :inet.port_number(), binary()) ::
-              {:ok, t()} | {:ok, binary(), t()} | {:error, term(), t()}
-
   @spec priority(type()) :: integer()
   def priority(type) do
     type_preference =

lib/ex_ice/priv/candidate/host.ex

@@ -30,9 +30,4 @@ defmodule ExICE.Priv.Candidate.Host do
       {:error, reason} -> {:error, reason, cand}
     end
   end
-
-  @impl true
-  def receive_data(cand, _src_ip, _src_port, data) do
-    {:ok, data, cand}
-  end
 end

lib/ex_ice/priv/candidate/prflx.ex

@@ -30,9 +30,4 @@ defmodule ExICE.Priv.Candidate.Prflx do
       {:error, reason} -> {:error, reason, cand}
     end
   end
-
-  @impl true
-  def receive_data(cand, _src_ip, _src_port, data) do
-    {:ok, data, cand}
-  end
 end

lib/ex_ice/priv/candidate/relay.ex

@@ -6,12 +6,12 @@ defmodule ExICE.Priv.Candidate.Relay do
 
   @type t() :: %__MODULE__{base: CandidateBase.t()}
 
-  @enforce_keys [:base]
-  defstruct @enforce_keys
+  @enforce_keys [:base, :client]
+  defstruct @enforce_keys ++ [buffered_packets: []]
 
   @impl true
   def new(config) do
-    %__MODULE__{base: CandidateBase.new(:relay, config)}
+    %__MODULE__{base: CandidateBase.new(:relay, config), client: Keyword.fetch!(config, :client)}
   end
 
   @impl true
@@ -25,14 +25,74 @@ defmodule ExICE.Priv.Candidate.Relay do
 
   @impl true
   def send_data(cand, dst_ip, dst_port, data) do
-    case cand.base.transport_module.send(cand.base.socket, {dst_ip, dst_port}, data) do
-      :ok -> {:ok, cand}
-      {:error, reason} -> {:error, reason, cand}
+    if MapSet.member?(cand.client.permissions, dst_ip) do
+      {:send, turn_addr, data, client} = ExTURN.Client.send(cand.client, {dst_ip, dst_port}, data)
+      cand = %{cand | client: client}
+      do_send(cand, turn_addr, data)
+    else
+      {:send, turn_addr, turn_data, client} = ExTURN.Client.create_permission(cand.client, dst_ip)
+
+      cand = %{
+        cand
+        | client: client,
+          buffered_packets: [{dst_ip, dst_port, data} | cand.buffered_packets]
+      }
+
+      do_send(cand, turn_addr, turn_data)
     end
   end
 
-  @impl true
-  def receive_data(cand, _src_ip, _src_port, data) do
-    {:ok, data, cand}
+  @spec receive_data(t(), :inet.ip_address(), :inet.port_number(), binary()) ::
+          {:ok, t()}
+          | {:ok, :inet.ip_address(), :inet.port_number(), t()}
+          | {:error, term(), t()}
+  def receive_data(cand, src_ip, src_port, data) do
+    case ExTURN.Client.handle_message(cand.client, {:socket_data, src_ip, src_port, data}) do
+      {:permission_created, permission_ip, client} ->
+        cand = %{cand | client: client}
+        send_buffered_packets(cand, permission_ip)
+
+      {:channel_created, _addr, client} ->
+        cand = %{cand | client: client}
+        {:ok, cand}
+
+      {:data, {src_ip, src_port}, data, client} ->
+        cand = %{cand | client: client}
+        {:ok, src_ip, src_port, data, cand}
+
+      {:error, reason, client} ->
+        cand = %{cand | client: client}
+        {:error, reason, cand}
+    end
+  end
+
+  defp send_buffered_packets(cand, permission_ip) do
+    {packets_to_send, rest} =
+      Enum.split_with(cand.buffered_packets, fn {dst_ip, _dst_port, _data} ->
+        dst_ip == permission_ip
+      end)
+
+    cand = %{cand | buffered_packets: rest}
+    do_send_buffered_packets(cand, Enum.reverse(packets_to_send))
+  end
+
+  defp do_send_buffered_packets(cand, []), do: {:ok, cand}
+
+  defp do_send_buffered_packets(cand, [{dst_ip, dst_port, packet} | packets]) do
+    {:send, turn_addr, data, client} = ExTURN.Client.send(cand.client, {dst_ip, dst_port}, packet)
+
+    cand = %{cand | client: client}
+
+    case do_send(cand, turn_addr, data) do
+      {:ok, cand} -> do_send_buffered_packets(cand, packets)
+      {:error, _reason, _cand} = error -> error
+    end
+  end
+
+  defp do_send(cand, dst_addr, data) do
+    case cand.base.transport_module.send(cand.base.socket, dst_addr, data) do
+      :ok -> {:ok, cand}
+      {:error, reason} -> {:error, reason, cand}
+    end
   end
 end

lib/ex_ice/priv/candidate/srflx.ex

@@ -30,9 +30,4 @@ defmodule ExICE.Priv.Candidate.Srflx do
       {:error, reason} -> {:error, reason, cand}
     end
   end
-
-  @impl true
-  def receive_data(cand, _src_ip, _src_port, data) do
-    {:ok, data, cand}
-  end
 end

lib/ex_ice/priv/candidate_pair.ex

@@ -11,19 +11,19 @@ defmodule ExICE.Priv.CandidatePair do
 
   @type t() :: %__MODULE__{
           id: integer(),
-          local_cand: Candidate.t(),
+          local_cand_id: integer(),
           nominate?: boolean(),
           nominated?: boolean(),
           priority: non_neg_integer(),
-          remote_cand: Candidate.t(),
+          remote_cand_id: integer(),
           state: state(),
           valid?: boolean,
           succeeded_pair_id: integer() | nil,
           discovered_pair_id: integer() | nil,
           keepalive_timer: reference() | nil
         }
 
-  @enforce_keys [:id, :local_cand, :remote_cand, :priority]
+  @enforce_keys [:id, :local_cand_id, :remote_cand_id, :priority]
   defstruct @enforce_keys ++
               [
                 nominate?: false,
@@ -39,12 +39,12 @@ defmodule ExICE.Priv.CandidatePair do
   @spec new(Candidate.t(), Candidate.t(), ExICE.ICEAgent.role(), state(), valid?: boolean()) ::
           t()
   def new(local_cand, remote_cand, agent_role, state, opts \\ []) do
-    priority = priority(agent_role, local_cand, remote_cand)
+    priority = priority(agent_role, local_cand.base.priority, remote_cand.priority)
 
     %__MODULE__{
       id: Utils.id(),
-      local_cand: local_cand,
-      remote_cand: remote_cand,
+      local_cand_id: local_cand.base.id,
+      remote_cand_id: remote_cand.id,
       priority: priority,
       state: state,
       valid?: opts[:valid?] || false
@@ -66,19 +66,19 @@ defmodule ExICE.Priv.CandidatePair do
   end
 
   @doc false
-  @spec recompute_priority(t(), ExICE.ICEAgent.role()) :: t()
-  def recompute_priority(pair, role) do
-    %__MODULE__{pair | priority: priority(role, pair.local_cand, pair.remote_cand)}
+  @spec recompute_priority(t(), integer(), integer(), ExICE.ICEAgent.role()) :: t()
+  def recompute_priority(pair, local_cand_prio, remote_cand_prio, role) do
+    %__MODULE__{pair | priority: priority(role, local_cand_prio, remote_cand_prio)}
   end
 
   @doc false
-  @spec priority(ExICE.ICEAgent.role(), Candidate.t(), ExICE.Candidate.t()) :: non_neg_integer()
-  def priority(:controlling, local_cand, remote_cand) do
-    do_priority(local_cand.base.priority, remote_cand.priority)
+  @spec priority(ExICE.ICEAgent.role(), integer(), integer()) :: non_neg_integer()
+  def priority(:controlling, local_cand_prio, remote_cand_prio) do
+    do_priority(local_cand_prio, remote_cand_prio)
   end
 
-  def priority(:controlled, local_cand, remote_cand) do
-    do_priority(remote_cand.priority, local_cand.base.priority)
+  def priority(:controlled, local_cand_prio, remote_cand_prio) do
+    do_priority(remote_cand_prio, local_cand_prio)
   end
 
   defp do_priority(g, d) do
@@ -89,15 +89,12 @@ defmodule ExICE.Priv.CandidatePair do
   @doc false
   @spec to_candidate_pair(t()) :: ExICE.CandidatePair.t()
   def to_candidate_pair(pair) do
-    %cand_mod{} = cand = pair.local_cand
-    local_cand = cand_mod.to_candidate(cand)
-
     %ExICE.CandidatePair{
       id: pair.id,
-      local_cand: local_cand,
+      local_cand_id: pair.local_cand_id,
       nominated?: pair.nominated?,
       priority: pair.priority,
-      remote_cand: pair.remote_cand,
+      remote_cand_id: pair.remote_cand_id,
       state: pair.state,
       valid?: pair.valid?
     }

lib/ex_ice/priv/checklist.ex

@@ -31,20 +31,14 @@ defmodule ExICE.Priv.Checklist do
 
   @spec find_pair(t(), CandidatePair.t()) :: CandidatePair.t() | nil
   def find_pair(checklist, pair) do
-    find_pair(checklist, pair.local_cand, pair.remote_cand)
+    find_pair(checklist, pair.local_cand_id, pair.remote_cand_id)
   end
 
-  @spec find_pair(t(), Candidate.t(), Candidate.t()) :: CandidatePair.t() | nil
-  def find_pair(checklist, local_cand, remote_cand) do
-    # TODO which pairs are actually the same?
+  @spec find_pair(t(), integer(), integer()) :: CandidatePair.t() | nil
+  def find_pair(checklist, local_cand_id, remote_cand_id) do
     checklist
     |> Enum.find({nil, nil}, fn {_id, p} ->
-      p.local_cand.base.base_address == local_cand.base.base_address and
-        p.local_cand.base.base_port == local_cand.base.base_port and
-        p.local_cand.base.address == local_cand.base.address and
-        p.local_cand.base.port == local_cand.base.port and
-        p.remote_cand.address == remote_cand.address and
-        p.remote_cand.port == remote_cand.port
+      p.local_cand_id == local_cand_id and p.remote_cand_id == remote_cand_id
     end)
     |> elem(1)
   end
@@ -64,13 +58,6 @@ defmodule ExICE.Priv.Checklist do
     not (waiting?(checklist) or in_progress?(checklist))
   end
 
-  @spec get_foundations(t()) :: [{integer(), integer()}]
-  def get_foundations(checklist) do
-    for {_id, pair} <- checklist do
-      {pair.local_cand.base.foundation, pair.remote_cand.foundation}
-    end
-  end
-
   @spec prune(t()) :: t()
   def prune(checklist) do
     # This is done according to RFC 8838 sec. 10
@@ -80,17 +67,20 @@ defmodule ExICE.Priv.Checklist do
     waiting =
       waiting
       |> Enum.sort_by(fn {_id, p} -> p.priority end, :desc)
-      |> Enum.uniq_by(fn {_id, p} ->
-        {p.local_cand.base.base_address, p.local_cand.base.base_port, p.remote_cand}
-      end)
+      # RFC 8445, sec. 6.1.2.4. states that two candidate pairs
+      # are redundant if their local candidates have the same base
+      # and their remote candidates are identical.
+      # But, because we replace reflexive candidates with their bases,
+      # checking againts local_cand_id should work fine.
+      |> Enum.uniq_by(fn {_id, p} -> {p.local_cand_id, p.remote_cand_id} end)
 
     Map.new(waiting ++ in_flight_or_done)
   end
 
   @spec prune(t(), Candidate.t()) :: t()
   def prune(checklist, local_cand) do
     checklist
-    |> Enum.reject(fn {_pair_id, pair} -> pair.local_cand.base.id == local_cand.base.id end)
+    |> Enum.reject(fn {_pair_id, pair} -> pair.local_cand_id == local_cand.base.id end)
     |> Map.new()
   end
 
@@ -104,11 +94,4 @@ defmodule ExICE.Priv.Checklist do
       end
     end
   end
-
-  @spec recompute_pair_prios(t(), ExICE.ICEAgent.role()) :: t()
-  def recompute_pair_prios(checklist, role) do
-    Map.new(checklist, fn {pair_id, pair} ->
-      {pair_id, CandidatePair.recompute_priority(pair, role)}
-    end)
-  end
 end