aboutsummaryrefslogtreecommitdiff
path: root/shard/lib/net/manager.ex
blob: dd803470d9a4cf15b3edb762fc339da990caafb1 (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
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, auth, callback}, _from, state) do
    pid = add_peer_internal(peer_info, auth)
    if callback != nil do
      GenServer.cast(pid, {:callback, callback})
    end
    {: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, :'$1', auth}) do
      [[pid2]|_] when pid2 != pid ->
        {:reply, :redundant, state}
      _ ->
        :ets.match_delete(:connections, {peer_info, pid, :_})
        :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, auth, msg}, state) do
    pid = add_peer_internal(peer_info, auth)
    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, auth) do
    if SNet.Addr.is_local? peer_info do
      nil
    else
      case :ets.match(:connections, {peer_info, :'$1', (if auth != nil do auth else :_ end)}) 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: auth})
          :ets.insert(:connections, {peer_info, pid, auth})
          pid
      end
    end
  end

  # =========
  # INTERFACE
  # =========

  @doc"""
  Connect to a peer specified by ip address and port
  """
  def add_peer(peer_info, opts \\ []) do
    GenServer.call(__MODULE__, {:add_peer, peer_info, opts[:auth], opts[:callback]})
  end

  @doc"""
  Return the list of all connected peers
  """
  def list_connections() do
    :ets.tab2list(:connections)
  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"""
  Return the list of connections to a given peer that match a given auth spec
  """
  def get_auth_connections_to(peer_info, my_auth, his_auth) do
    for {^peer_info, pid, %SNet.Auth{my_pk: my_pk, his_pk: his_pk}} <- :ets.lookup(:connections, peer_info),
      my_pk == my_auth or my_pk in my_auth,
      his_pk == his_auth or his_pk in his_auth,
    do: pid
  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, nil, msg})
    end
  end

  def send_auth(peer_info, auth, msg) do
    case :ets.match(:connections, {peer_info, :'$1', auth}) do
      [[pid]|_] ->
        GenServer.cast(pid, {:send_msg, msg})
      [] ->
        GenServer.cast(__MODULE__, {:connect_and_send, peer_info, auth, 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