Address, broadcast group management

2 files changed, 125 insertions, 0 deletions

diff --git a/shard/lib/net/addr.ex b/shard/lib/net/addr.ex
new file mode 100644
index 0000000..645e109
--- /dev/null
+++ b/shard/lib/net/addr.ex
@@ -0,0 +1,27 @@
+defmodule SNet.Addr do
+
+  def get_if_inet4 do
+    {:ok, ifs} = :inet.getifaddrs
+    for {_, opts} <- ifs,
+        {:addr, addr} <- opts,
+        tuple_size(addr) == 4,
+        addr != {127,0,0,1}
+    do
+      addr
+    end
+  end
+
+  def get_pub_inet4 do
+    Application.ensure_all_started(:inets)
+    Application.ensure_all_started(:ssl)
+    {:ok, {_, _, body}} = :httpc.request('http://api.ipify.org')
+    {:ok, addr} = :inet.parse_address body
+    addr
+  end
+
+  def get_all_inet4 do
+    addrset = for x <- get_if_inet4(), into: %MapSet{}, do: x
+    addrset = MapSet.put(addrset, get_pub_inet4())
+    MapSet.to_list addrset
+  end
+end
diff --git a/shard/lib/net/group.ex b/shard/lib/net/group.ex
new file mode 100644
index 0000000..d2c2537
--- /dev/null
+++ b/shard/lib/net/group.ex
@@ -0,0 +1,98 @@
+defprotocol SNet.Group do
+  @moduledoc"""
+  A group is a specification of a bunch of peers we want and accept to talk to
+  about some things. It supports a number of abstract operations for finding peers,
+  broadcasting/gossiping, authenticating, etc.
+  """
+
+  @doc"""
+  Find new peers for this group, open connections and notify us when connections are open.
+
+  Launches background processes if necessary, returns immediately.
+  """
+  def init_lookup(group, notify_to) 
+
+  @doc"""
+  Get all currently open connections to peers in this group.
+  """
+  def get_connections(group)
+
+  @doc"""
+  Broadcast a message to peers of the group.
+  Will send to at most nmax peers, so this is a good primitive for gossip.
+  """
+  def broadcast(group, msg, nmax \\ 10)
+
+  @doc"""
+  Check if a peer is allowed to participate in this group.
+  """
+  def in_group?(group, conn_pid, auth)
+end
+
+defmodule SNet.PubShardGroup do
+  defstruct [:id]
+
+  defimpl SNet.Group 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
+          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
+          end
+        end
+      end
+      # TODO: use a DHT to find peers
+    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
+    end
+
+    def broadcast(group, msg, nmax) do
+      %SNet.PubShardGroup{id: id} = group
+      nsent = get_connections(group)
+      |> Enum.shuffle
+      |> Enum.take(nmax)
+      |> Enum.map(&(GenServer.cast(&1, {:send_msg, msg})))
+      |> Enum.count
+      if nmax - nsent > 0 do
+        Shard.Manager.get_shard_peers(id)
+        |> Enum.filter(&(Shard.Manager.get_connections_to(&1) == []))
+        |> Enum.shuffle
+        |> Enum.take(nmax - nsent)
+        |> Enum.map(&(Shard.Manager.send(&1, msg)))
+      end
+    end
+
+    def in_group?(%SNet.PubShardGroup{id: _id}, _peer_pid, _auth) do
+      true     # No access control
+    end
+  end
+end
+
+defmodule SNet.PrivGroup do
+  defstruct [:pk_list]
+
+  defimpl SNet.Group do
+    def init_lookup(%SNet.PubShardGroup{id: id}, notify_to) do
+      # TODO
+    end
+
+    def get_connections(%SNet.PubShardGroup{id: id}) do
+      # TODO
+    end
+
+    def broadcast(group, msg, nmax) do
+    end
+
+    def in_group?(%SNet.PubShardGroup{id: _id}, peer_pid, auth) do
+      # TODO
+    end
+  end
+end