Move somme functionnality to SNet.Manager

4 files changed, 142 insertions, 22 deletions

diff --git a/shard/lib/net/group.ex b/shard/lib/net/group.ex
index d2c2537..692438a 100644
--- a/shard/lib/net/group.ex
+++ b/shard/lib/net/group.ex
@@ -36,12 +36,12 @@ defmodule SNet.PubShardGroup do
     def init_lookup(%SNet.PubShardGroup{id: id}, _notify_to) do
       # For now: ask all currently connected peers and connect to new peers we know of
       spawn fn ->
-        for {_, pid, _} <- Shard.Manager.list_connections do
+        for {_, pid, _} <- SNet.Manager.list_connections do
           GenServer.cast(pid, {:send_msg, {:interested, [id]}})
         end
         for peer_info <- Shard.Manager.get_shard_peers id do
-          if Shard.Manager.get_connections_to peer_info == [] do
-            Shard.Manager.add_peer(peer_info)      # TODO callback when connected
+          if SNet.Manager.get_connections_to peer_info == [] do
+            SNet.Manager.add_peer(peer_info)      # TODO callback when connected
           end
         end
       end
@@ -49,9 +49,10 @@ defmodule SNet.PubShardGroup do
     end
 
     def get_connections(%SNet.PubShardGroup{id: id}) do
-      for peer_info <- Shard.Manager.get_shard_peers(id),
-          [{pid, _auth}|_] = Shard.Manager.get_connections_to(peer_info),
-        do: pid
+      Shard.Manager.get_shard_peers(id)
+      |> Enum.map(&(SNet.Manager.get_connections_to(&1)))
+      |> Enum.filter(&(&1 != []))
+      |> Enum.map(fn [{pid, _auth}|_] -> pid end)
     end
 
     def broadcast(group, msg, nmax) do
@@ -63,10 +64,10 @@ defmodule SNet.PubShardGroup do
       |> Enum.count
       if nmax - nsent > 0 do
         Shard.Manager.get_shard_peers(id)
-        |> Enum.filter(&(Shard.Manager.get_connections_to(&1) == []))
+        |> Enum.filter(&(SNet.Manager.get_connections_to(&1) == []))
         |> Enum.shuffle
         |> Enum.take(nmax - nsent)
-        |> Enum.map(&(Shard.Manager.send(&1, msg)))
+        |> Enum.map(&(SNet.Manager.send(&1, msg)))
       end
     end
 
diff --git a/shard/lib/net/manager.ex b/shard/lib/net/manager.ex
new file mode 100644
index 0000000..17d6e06
--- /dev/null
+++ b/shard/lib/net/manager.ex
@@ -0,0 +1,119 @@
+defmodule SNet.Manager do
+  @moduledoc"""
+  - :connections (not persistent)
+
+      List of
+        { peer_info, pid, nil | {my_pk, his_pk} }
+  """
+
+  use GenServer
+
+  require Logger
+
+  def start_link(_) do
+    GenServer.start_link(__MODULE__, nil, name: __MODULE__)
+  end
+
+  def init(_) do
+    Process.flag(:trap_exit, true)
+
+    :ets.new(:connections, [:bag, :protected, :named_table])
+
+    {:ok, nil} 
+  end
+
+  def handle_call({:add_peer, peer_info}, _from, state) do
+    pid = add_peer_internal(peer_info)
+    {:reply, pid, state}
+  end
+
+  def handle_call({:accept, client}, _from, state) do
+    my_port = Application.get_env(:shard, :port)
+    {:ok, pid} = SNet.TCPConn.start_link(%{socket: client, my_port: my_port})
+    {:reply, pid, state}
+  end
+
+  def handle_call({:peer_up, pid, peer_info, auth}, _from, state) do
+    case :ets.match(:connections,  {peer_info, :_, auth}) do
+      [{_, pid2, _}] when pid2 != pid ->
+        {:reply, :redundant, state}
+      _ ->
+        :ets.insert(:connections, {peer_info, pid, auth})
+
+        # Send interested message for all our shards
+        id_list = (for {id, _, _} <- Shard.Manager.list_shards(), do: id)
+        GenServer.cast(pid, {:send_msg, {:interested, id_list}})
+
+        {:reply, :ok, state}
+    end
+  end
+
+  def handle_cast({:connect_and_send, peer_info, msg}, state) do
+    pid = add_peer_internal(peer_info)
+    GenServer.cast(pid, {:send_msg, msg})
+    {:noreply, state}
+  end
+
+  def handle_info({:EXIT, pid, _reason}, state) do
+    :ets.match_delete(:connections, {:_, pid, :_})
+    {:noreply, state}
+  end
+
+  defp add_peer_internal(peer_info) do
+    case :ets.lookup(:connections, peer_info) do
+      [{_, pid, _}|_] ->
+        pid
+      [] ->
+        my_port = Application.get_env(:shard, :port)
+        {:ok, pid} = SNet.TCPConn.start_link(%{connect_to: peer_info, my_port: my_port, auth: nil})
+        :ets.insert(:connections, {peer_info, pid, nil})
+        pid
+    end
+  end
+
+  # =========
+  # INTERFACE
+  # =========
+
+  @doc"""
+  Connect to a peer specified by ip address and port
+  """
+  def add_peer(peer_info) do
+    GenServer.call(__MODULE__, {:add_peer, peer_info})
+  end
+
+  @doc"""
+  Return the list of all connected peers
+  """
+  def list_connections() do
+    for [x] <- :ets.match(:connections, :"$1"), do: x
+  end
+
+  @doc"""
+  Return the list of connections to a given peer, possibly with different auth
+  """
+  def get_connections_to(peer_info) do
+    for {^peer_info, pid, auth} <- :ets.lookup(:connections, peer_info), do: {pid, auth}
+  end
+
+  @doc"""
+  Send message to a peer specified by peer info.
+  Opens a connection if necessary.
+  """
+  def send(peer_info, msg) do
+    case :ets.lookup(:connections, peer_info) do
+      [{^peer_info, pid, _auth}|_] ->
+        GenServer.cast(pid, {:send_msg, msg})
+      [] ->
+        GenServer.cast(__MODULE__, {:connect_and_send, peer_info, msg})
+    end
+  end
+
+  @doc"""
+  Send message to a peer specified by peer id
+  """
+  def send_pid(pid, msg) do
+    GenServer.cast(pid, {:send_msg, msg})
+  end
+end
+
diff --git a/shard/lib/net/tcpconn.ex b/shard/lib/net/tcpconn.ex
index 67d7f4c..476c426 100644
--- a/shard/lib/net/tcpconn.ex
+++ b/shard/lib/net/tcpconn.ex
@@ -33,23 +33,21 @@ defmodule SNet.TCPConn do
   end
 
   def init(state) do
-    if state.is_client do
-      GenServer.cast(self(), :client_handshake)
-    else
+    if Map.has_key?(state, :socket) do
       GenServer.cast(self(), :server_handshake)
+    else
+      GenServer.cast(self(), :client_handshake)
     end
-
     {:ok, state}
   end
 
-
   def handle_call(:get_peer_info, _from, state) do
     {:reply, state.peer_info, state}
   end
 
-
   def handle_cast(:client_handshake, state) do
-    socket = state.socket
+    {:inet, ip, port} = state.connect_to
+    {:ok, socket} = :gen_tcp.connect(ip, port, [:binary, packet: 2, active: false])
 
     net_key = Application.get_env(:shard, :network_key)
     %{public: cli_eph_pk, secret: cli_eph_sk} = :enacl.box_keypair
@@ -150,9 +148,11 @@ defmodule SNet.TCPConn do
             |> Map.put(:peer_info, {:inet, addr, port})
             |> Map.put(:my_port, state.my_port)
 
-    GenServer.cast(Shard.Manager, {:peer_up, self(), state.peer_info, state.auth})
-    Logger.info "New peer: #{print_id state} at #{inspect addr}:#{port}"
+    if GenServer.call(SNet.Manager, {:peer_up, self(), state.peer_info, state.auth}) == :redundant do
+      exit :redundant
+    end
 
+    Logger.info "New peer: #{print_id state} at #{inspect addr}:#{port}"
     {:noreply, state}
   end
 
@@ -262,9 +262,11 @@ defmodule SNet.TCPConn do
             |> Map.put(:peer_info, {:inet, addr, his_port})
             |> Map.put(:my_port, state.my_port)
 
-    GenServer.cast(Shard.Manager, {:peer_up, self(), state.peer_info, state.auth})
-    Logger.info "New peer: #{print_id state} at #{inspect state.peer_info} (#{port})"
+    if GenServer.call(SNet.Manager, {:peer_up, self(), state.peer_info, state.auth}) == :redundant do
+      exit :redundant
+    end
 
+    Logger.info "New peer: #{print_id state} at #{inspect state.peer_info} (#{port})"
     {:noreply, state}
   end
 
@@ -284,7 +286,6 @@ defmodule SNet.TCPConn do
 
   def handle_info({:tcp_closed, _socket}, state) do
     Logger.info "Disconnected: #{print_id state} at #{inspect state.peer_info}"
-    GenServer.cast(Shard.Manager, {:peer_down, self(), state.peer_info, state.auth})
     exit(:normal)
   end
 
diff --git a/shard/lib/net/tcpserver.ex b/shard/lib/net/tcpserver.ex
index 6cc3473..d7326ad 100644
--- a/shard/lib/net/tcpserver.ex
+++ b/shard/lib/net/tcpserver.ex
@@ -19,8 +19,7 @@ defmodule SNet.TCPServer do
 
   defp loop_acceptor(socket, my_port) do
     {:ok, client} = :gen_tcp.accept(socket)
-    {:ok, pid} = DynamicSupervisor.start_child(Shard.DynamicSupervisor,
-                    {SNet.TCPConn, %{socket: client, my_port: my_port, is_client: false}})
+    pid = GenServer.call(SNet.Manager, {:accept, client})
     :ok = :gen_tcp.controlling_process(client, pid)
     loop_acceptor(socket, my_port)
   end