allow multiple tcp clients

Author: sgfn

lib/ex_ice/ice_agent.ex

@@ -322,13 +322,13 @@ defmodule ExICE.ICEAgent do
 
   @impl true
   def init(opts) do
-    # TODO: this is ugly, and will not allow us to run more than two TCP ICE agents at the same time
+    # TODO: this is ugly, and potentially a bottleneck
     opts =
       if opts[:transport] == :tcp do
-        {:ok, _pid} = ExICE.Priv.Transport.TCP.Client.start_link()
-        opts ++ [transport_module: ExICE.Priv.Transport.TCP.Client]
+        {:ok, pid} = ExICE.Priv.Transport.TCP.Client.start_link()
+        opts ++ [transport_module: %{mod: ExICE.Priv.Transport.TCP.Client, ref: pid}]
       else
-        opts
+        opts ++ [transport_module: %{mod: ExIce.Priv.Transport.UDP, ref: nil}]
       end
 
     ice_agent = ExICE.Priv.ICEAgent.new(opts)

lib/ex_ice/priv/candidate.ex

@@ -118,7 +118,7 @@ defmodule ExICE.Priv.Candidate do
 
   defp generate_other_preference(local_preferences, attempts) do
     # 0..8191
-    <<pref::13, _::bitstring>> = :crypto.strong_rand_bytes(2)
+    <<pref::13, _::3>> = :crypto.strong_rand_bytes(2)
 
     if local_preferences |> Map.values() |> Enum.member?(pref) do
       generate_other_preference(local_preferences, attempts - 1)

lib/ex_ice/priv/candidate/host.ex

@@ -2,7 +2,7 @@ defmodule ExICE.Priv.Candidate.Host do
   @moduledoc false
   @behaviour ExICE.Priv.Candidate
 
-  alias ExICE.Priv.CandidateBase
+  alias ExICE.Priv.{CandidateBase, Transport}
 
   @type t() :: %__MODULE__{base: CandidateBase.t()}
 
@@ -28,7 +28,7 @@ defmodule ExICE.Priv.Candidate.Host 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
+    case Transport.send(cand.base.transport_module, cand.base.socket, {dst_ip, dst_port}, data) do
       :ok -> {:ok, cand}
       {:error, reason} -> {:error, reason, cand}
     end

lib/ex_ice/priv/candidate/prflx.ex

@@ -2,7 +2,7 @@ defmodule ExICE.Priv.Candidate.Prflx do
   @moduledoc false
   @behaviour ExICE.Priv.Candidate
 
-  alias ExICE.Priv.CandidateBase
+  alias ExICE.Priv.{CandidateBase, Transport}
 
   @type t() :: %__MODULE__{base: CandidateBase.t()}
 
@@ -28,7 +28,7 @@ defmodule ExICE.Priv.Candidate.Prflx 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
+    case Transport.send(cand.base.transport_module, cand.base.socket, {dst_ip, dst_port}, data) do
       :ok -> {:ok, cand}
       {:error, reason} -> {:error, reason, cand}
     end

lib/ex_ice/priv/candidate/relay.ex

@@ -2,7 +2,7 @@ defmodule ExICE.Priv.Candidate.Relay do
   @moduledoc false
   @behaviour ExICE.Priv.Candidate
 
-  alias ExICE.Priv.CandidateBase
+  alias ExICE.Priv.{CandidateBase, Transport}
 
   @type t() :: %__MODULE__{base: CandidateBase.t()}
 
@@ -125,7 +125,7 @@ defmodule ExICE.Priv.Candidate.Relay do
   end
 
   defp do_send(cand, dst_addr, data) do
-    case cand.base.transport_module.send(cand.base.socket, dst_addr, data) do
+    case Transport.send(cand.base.transport_module, cand.base.socket, dst_addr, data) do
       :ok -> {:ok, cand}
       {:error, reason} -> {:error, reason, cand}
     end

lib/ex_ice/priv/candidate/srflx.ex

@@ -2,7 +2,7 @@ defmodule ExICE.Priv.Candidate.Srflx do
   @moduledoc false
   @behaviour ExICE.Priv.Candidate
 
-  alias ExICE.Priv.CandidateBase
+  alias ExICE.Priv.{CandidateBase, Transport}
 
   @type t() :: %__MODULE__{base: CandidateBase.t()}
 
@@ -28,7 +28,7 @@ defmodule ExICE.Priv.Candidate.Srflx 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
+    case Transport.send(cand.base.transport_module, cand.base.socket, {dst_ip, dst_port}, data) do
       :ok -> {:ok, cand}
       {:error, reason} -> {:error, reason, cand}
     end

lib/ex_ice/priv/candidate_base.ex

@@ -1,6 +1,6 @@
 defmodule ExICE.Priv.CandidateBase do
   @moduledoc false
-  alias ExICE.Priv.{Candidate, Utils}
+  alias ExICE.Priv.{Candidate, Transport, Utils}
 
   @type t :: %__MODULE__{
           id: integer(),
@@ -11,6 +11,7 @@ defmodule ExICE.Priv.CandidateBase do
           port: :inet.port_number(),
           priority: integer(),
           transport: :udp | :tcp,
+          # XXX
           transport_module: module(),
           socket: :inet.socket() | nil,
           type: Candidate.type(),
@@ -34,7 +35,7 @@ defmodule ExICE.Priv.CandidateBase do
   @spec new(Candidate.type(), Keyword.t()) :: t()
   def new(type, config) do
     transport_module = Keyword.fetch!(config, :transport_module)
-    transport = transport_module.transport()
+    transport = Transport.transport(transport_module)
     address = Keyword.fetch!(config, :address)
 
     %__MODULE__{

lib/ex_ice/priv/gatherer.ex

@@ -12,6 +12,7 @@ defmodule ExICE.Priv.Gatherer do
 
   @type t() :: %__MODULE__{
           if_discovery_module: module(),
+          # XXX
           transport_module: module(),
           ip_filter: (:inet.ip_address() -> boolean),
           ports: Enumerable.t(non_neg_integer())
@@ -41,7 +42,7 @@ defmodule ExICE.Priv.Gatherer do
         |> Stream.reject(&unsupported_ipv6?(&1))
         |> Enum.to_list()
 
-      for transport_opts <- gatherer.transport_module.socket_configs(),
+      for transport_opts <- Transport.socket_configs(gatherer.transport_module),
           ip <- ips do
         open_socket(gatherer, ip, transport_opts)
       end
@@ -67,9 +68,15 @@ defmodule ExICE.Priv.Gatherer do
     ]
 
     Enum.reduce_while(gatherer.ports, nil, fn port, _ ->
-      case gatherer.transport_module.setup_socket(ip, port, socket_opts, transport_opts) do
+      case Transport.setup_socket(
+             gatherer.transport_module,
+             ip,
+             port,
+             socket_opts,
+             transport_opts
+           ) do
         {:ok, socket} ->
-          {:ok, {^ip, sock_port}} = gatherer.transport_module.sockname(socket)
+          {:ok, {^ip, sock_port}} = Transport.sockname(gatherer.transport_module, socket)
 
           Logger.debug(
             "Successfully opened socket for: #{inspect(ip)}:#{sock_port}, socket: #{inspect(socket)}"
@@ -118,7 +125,7 @@ defmodule ExICE.Priv.Gatherer do
         ip = List.first(ips)
         port = stun_server.port
 
-        {:ok, {sock_ip, _sock_port}} = gatherer.transport_module.sockname(socket)
+        {:ok, {sock_ip, _sock_port}} = Transport.sockname(gatherer.transport_module, socket)
 
         cand_family = Utils.family(sock_ip)
         stun_family = Utils.family(ip)
@@ -127,7 +134,7 @@ defmodule ExICE.Priv.Gatherer do
           # Communication with STUN servers should be handled differently
           # than the rest of the TCP traffic: the messages are not RFC 4571-framed,
           # and we want to issue connection attempts from passive candidates as well.
-          gatherer.transport_module.send(socket, {ip, port}, binding_request,
+          Transport.send(gatherer.transport_module, socket, {ip, port}, binding_request,
             frame?: false,
             connect?: true
           )
@@ -281,7 +288,7 @@ defmodule ExICE.Priv.Gatherer do
          socket: socket,
          transport_opts: transport_opts
        }) do
-    {:ok, {sock_ip, sock_port}} = gatherer.transport_module.sockname(socket)
+    {:ok, {sock_ip, sock_port}} = Transport.sockname(gatherer.transport_module, socket)
 
     tcp_type = transport_opts[:tcp_type]
 

lib/ex_ice/priv/ice_agent.ex

@@ -142,9 +142,9 @@ defmodule ExICE.Priv.ICEAgent do
     {local_ufrag, local_pwd} = generate_credentials()
 
     controlling_process = Keyword.fetch!(opts, :controlling_process)
+    transport_module = Keyword.fetch!(opts, :transport_module)
 
     if_discovery_module = opts[:if_discovery_module] || IfDiscovery.Inet
-    transport_module = opts[:transport_module] || Transport.UDP
     ip_filter = opts[:ip_filter] || fn _ -> true end
     ports = opts[:ports] || [0]
 
@@ -944,7 +944,7 @@ defmodule ExICE.Priv.ICEAgent do
 
           {:send, dst, data, client} ->
             tr = %{tr | client: client}
-            :ok = ice_agent.transport_module.send(tr.socket, dst, data)
+            :ok = Transport.send(ice_agent.transport_module, tr.socket, dst, data)
             put_in(ice_agent.gathering_transactions[tr_id], tr)
 
           {:error, _reason, _client} ->
@@ -962,7 +962,7 @@ defmodule ExICE.Priv.ICEAgent do
             cand = %{cand | client: client}
             ice_agent = put_in(ice_agent.local_cands[cand.base.id], cand)
             # we can't use do_send here as it will try to create permission for the turn address
-            :ok = ice_agent.transport_module.send(cand.base.socket, dst, data)
+            :ok = Transport.send(ice_agent.transport_module, cand.base.socket, dst, data)
             ice_agent
 
           {:error, _reason, client} ->
@@ -1013,7 +1013,7 @@ defmodule ExICE.Priv.ICEAgent do
     for turn_server <- turn_servers, %{socket: socket} <- sockets do
       with {:ok, client} <-
              ExTURN.Client.new(turn_server.url, turn_server.username, turn_server.credential),
-           {:ok, {sock_ip, _sock_port}} <- ice_agent.transport_module.sockname(socket),
+           {:ok, {sock_ip, _sock_port}} <- Transport.sockname(ice_agent.transport_module, socket),
            {true, _, _} <-
              {Utils.family(client.turn_ip) == Utils.family(sock_ip), client, sock_ip} do
         t_id = {socket, {client.turn_ip, client.turn_port}}
@@ -1098,7 +1098,7 @@ defmodule ExICE.Priv.ICEAgent do
   end
 
   defp execute_gathering_transaction(ice_agent, %{stun_server: stun_server} = tr) do
-    {:ok, {sock_ip, sock_port}} = ice_agent.transport_module.sockname(tr.socket)
+    {:ok, {sock_ip, sock_port}} = Transport.sockname(ice_agent.transport_module, tr.socket)
 
     Logger.debug("""
     Sending binding request to gather srflx candidate for:
@@ -1124,7 +1124,7 @@ defmodule ExICE.Priv.ICEAgent do
   end
 
   defp execute_gathering_transaction(ice_agent, %{client: client} = tr) do
-    {:ok, {sock_ip, sock_port}} = ice_agent.transport_module.sockname(tr.socket)
+    {:ok, {sock_ip, sock_port}} = Transport.sockname(ice_agent.transport_module, tr.socket)
 
     Logger.debug("""
     Starting the process of gathering relay candidate for:
@@ -1135,7 +1135,7 @@ defmodule ExICE.Priv.ICEAgent do
     {:send, turn_addr, data, client} = ExTURN.Client.allocate(client)
     tr = Map.put(tr, :client, client)
 
-    case ice_agent.transport_module.send(tr.socket, turn_addr, data) do
+    case Transport.send(ice_agent.transport_module, tr.socket, turn_addr, data) do
       :ok ->
         tr = %{tr | state: :in_progress, send_time: now()}
         ice_agent = put_in(ice_agent.gathering_transactions[tr.t_id], tr)
@@ -1215,7 +1215,7 @@ defmodule ExICE.Priv.ICEAgent do
 
         # Use sock_addr for calculating priority.
         # In other case, we might get duplicates.
-        {:ok, {sock_addr, _sock_port}} = ice_agent.transport_module.sockname(tr.socket)
+        {:ok, {sock_addr, _sock_port}} = Transport.sockname(ice_agent.transport_module, tr.socket)
 
         # TODO: set correct tcp_type here
         {local_preferences, priority} =
@@ -1250,7 +1250,7 @@ defmodule ExICE.Priv.ICEAgent do
 
       {:send, turn_addr, data, client} ->
         tr = %{tr | client: client}
-        :ok = ice_agent.transport_module.send(tr.socket, turn_addr, data)
+        :ok = Transport.send(ice_agent.transport_module, tr.socket, turn_addr, data)
         put_in(ice_agent.gathering_transactions[tr_id], tr)
 
       {:error, _reason, _client} ->
@@ -1912,7 +1912,7 @@ defmodule ExICE.Priv.ICEAgent do
 
     case find_local_cand(Map.values(ice_agent.local_cands), xor_addr, xor_port) do
       nil ->
-        {:ok, {base_addr, base_port}} = ice_agent.transport_module.sockname(tr.socket)
+        {:ok, {base_addr, base_port}} = Transport.sockname(ice_agent.transport_module, tr.socket)
 
         host_cand = find_host_cand(Map.values(ice_agent.local_cands), tr.socket)
 
@@ -2423,7 +2423,7 @@ defmodule ExICE.Priv.ICEAgent do
     # Alternatively, we could create a callback for `:inet.info/1`,
     # but it's return type is not standardized - sometimes it's %{states: [:closed]},
     # some other time %{rstates: [:closed], wstates: [:closed]}.
-    case ice_agent.transport_module.sockname(socket) do
+    case Transport.sockname(ice_agent.transport_module, socket) do
       {:error, :closed} -> ice_agent
       _ -> do_close_socket(ice_agent, socket)
     end
@@ -2448,7 +2448,7 @@ defmodule ExICE.Priv.ICEAgent do
 
     tr_rtx = ice_agent.tr_rtx -- Map.keys(removed_gathering_transactions)
 
-    :ok = ice_agent.transport_module.close(socket)
+    :ok = Transport.close(ice_agent.transport_module, socket)
     :ok = flush_socket_msg(socket)
 
     %{ice_agent | tr_rtx: tr_rtx, gathering_transactions: gathering_transactions}
@@ -2513,12 +2513,12 @@ defmodule ExICE.Priv.ICEAgent do
 
   defp do_restart(ice_agent) do
     Enum.each(ice_agent.sockets, fn socket ->
-      case ice_agent.transport_module.sockname(socket) do
+      case Transport.sockname(ice_agent.transport_module, socket) do
         {:ok, {ip, port}} ->
           # we could use close_socket function here but because we are
           # clearing the whole state anyway, we can close the socket manually
           Logger.debug("Closing socket: #{inspect(ip)}:#{port}.")
-          :ok = ice_agent.transport_module.close(socket)
+          :ok = Transport.close(ice_agent.transport_module, socket)
           :ok = flush_socket_msg(socket)
 
         {:error, :closed} ->
@@ -3116,7 +3116,7 @@ defmodule ExICE.Priv.ICEAgent do
     priority =
       if local_candidate.base.type == :relay do
         {:ok, {sock_addr, _sock_port}} =
-          ice_agent.transport_module.sockname(local_candidate.base.socket)
+          Transport.sockname(ice_agent.transport_module, local_candidate.base.socket)
 
         Candidate.priority!(
           ice_agent.local_preferences,

lib/ex_ice/priv/transport.ex

@@ -2,6 +2,7 @@ defmodule ExICE.Priv.Transport do
   @moduledoc false
 
   @type socket :: term()
+  @type tp_ref :: term()
   @type open_option ::
           :inet.inet_backend()
           | :inet.address_family()
@@ -12,14 +13,50 @@ defmodule ExICE.Priv.Transport do
 
   @callback socket_configs() :: [map()]
 
-  @callback setup_socket(:inet.ip_address(), :inet.port_number(), [open_option()], map()) ::
+  @callback setup_socket(
+              tp_ref(),
+              :inet.ip_address(),
+              :inet.port_number(),
+              [open_option()],
+              map()
+            ) ::
               {:ok, socket()} | {:error, term()}
 
   @callback sockname(socket()) ::
               {:ok, {:inet.ip_address(), :inet.port_number()}} | {:error, term()}
 
-  @callback send(socket(), {:inet.ip_address(), :inet.port_number()}, binary(), Keyword.t()) ::
+  @callback send(
+              tp_ref(),
+              socket(),
+              {:inet.ip_address(), :inet.port_number()},
+              binary(),
+              Keyword.t()
+            ) ::
               :ok | {:error, term()}
 
-  @callback close(socket()) :: :ok
+  @callback close(tp_ref(), socket()) :: :ok
+
+  def transport(tp_spec) do
+    tp_spec.mod.transport()
+  end
+
+  def socket_configs(tp_spec) do
+    tp_spec.mod.socket_configs()
+  end
+
+  def setup_socket(tp_spec, ip, port, socket_opts, tp_opts \\ []) do
+    tp_spec.mod.setup_socket(tp_spec.ref, ip, port, socket_opts, tp_opts)
+  end
+
+  def sockname(tp_spec, socket) do
+    tp_spec.mod.sockname(socket)
+  end
+
+  def send(tp_spec, socket, dest, packet, tp_opts \\ []) do
+    tp_spec.mod.send(tp_spec.ref, socket, dest, packet, tp_opts)
+  end
+
+  def close(tp_spec, socket) do
+    tp_spec.mod.close(tp_spec.ref, socket)
+  end
 end