aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--shard/lib/app/chat.ex123
-rw-r--r--shard/lib/app/identity.ex19
-rw-r--r--shard/lib/manager.ex4
-rw-r--r--shard/lib/net/addr.ex27
-rw-r--r--shard/lib/net/group.ex98
5 files changed, 196 insertions, 75 deletions
diff --git a/shard/lib/app/chat.ex b/shard/lib/app/chat.ex
index 0f4d573..051dfef 100644
--- a/shard/lib/app/chat.ex
+++ b/shard/lib/app/chat.ex
@@ -5,7 +5,7 @@ defmodule SApp.Chat do
Chat rooms are globally identified by their channel name.
A chat room manifest is of the form:
- {:chat, channel_name}
+ %SApp.Chat.Manifest{channel: channel_name}
Future improvements:
- message signing
@@ -28,7 +28,7 @@ defmodule SApp.Chat do
defimpl Shard.Manifest, for: Manifest do
def start(m) do
- DynamicSupervisor.start_child(Shard.DynamicSupervisor, {SApp.Chat, m.channel})
+ DynamicSupervisor.start_child(Shard.DynamicSupervisor, {SApp.Chat, m})
end
end
@@ -36,15 +36,15 @@ defmodule SApp.Chat do
@doc """
Start a process that connects to a given channel
"""
- def start_link(channel) do
- GenServer.start_link(__MODULE__, channel)
+ def start_link(manifest) do
+ GenServer.start_link(__MODULE__, manifest)
end
@doc """
Initialize channel process.
"""
- def init(channel) do
- manifest = %Manifest{channel: channel}
+ def init(manifest) do
+ %Manifest{channel: channel} = manifest
id = SData.term_hash manifest
case Shard.Manager.register(id, manifest, self()) do
@@ -62,10 +62,12 @@ defmodule SApp.Chat do
mst = %MST{store: %SApp.PageStore{pid: page_store},
cmp: &msg_cmp/2,
root: root}
- GenServer.cast(self(), :init_pull)
+ group = %SNet.PubShardGroup{id: id}
+ SNet.Group.init_lookup(group, self())
{:ok,
%{channel: channel,
id: id,
+ group: group,
manifest: manifest,
page_store: page_store,
mst: mst,
@@ -96,18 +98,6 @@ defmodule SApp.Chat do
end
@doc """
- Implementation of the :init_pull handler, which is called when the
- process starts. It contacts all currently connected peers and asks them to
- send data for this channel if they have some.
- """
- def handle_cast(:init_pull, state) do
- for {_, pid, _} <- Shard.Manager.list_connections do
- GenServer.cast(pid, {:send_msg, {:interested, [state.id]}})
- end
- {:noreply, state}
- end
-
- @doc """
Implementation of the :chat_send handler. This is the main handler that is used
to send a message to the chat room. Puts the message in the store and syncs
with all connected peers.
@@ -129,9 +119,7 @@ defmodule SApp.Chat do
end
notif = {:append, prev_root, msgitem, mst.root}
- for peer_info <- Shard.Manager.get_shard_peers(state.id) do
- Shard.Manager.send(peer_info, {state.id, nil, notif})
- end
+ SNet.Group.broadcast(state.group, notif)
{:noreply, state}
end
@@ -140,8 +128,10 @@ defmodule SApp.Chat do
Implementation of the :interested handler, this is called when a peer we are
connected to asks to recieve data for this channel.
"""
- def handle_cast({:interested, conn_pid, _auth}, state) do
- Shard.Manager.send_pid(conn_pid, {state.id, nil, {:root, state.mst.root}})
+ def handle_cast({:interested, conn_pid, auth}, state) do
+ if SNet.Group.in_group?(state.group, conn_pid, auth) do
+ Shard.Manager.send_pid(conn_pid, {state.id, nil, {:root, state.mst.root}})
+ end
{:noreply, state}
end
@@ -160,54 +150,59 @@ defmodule SApp.Chat do
- `{:info, start, list, rest}`: put some messages and informs of the
Merkle hash of the store of older messages.
"""
- def handle_cast({:msg, conn_pid, _auth, _shard_id, nil, msg}, state) do
- state = case msg do
- {:get_manifest} ->
- Shard.Manager.send_pid(conn_pid, {state.id, nil, {:manifest, state.manifest}})
- state
- {:append, prev_root, msgitem, new_root} ->
- # Append message: one single mesage has arrived
- if new_root == state.mst.root do
- # We already have the message, do nothing
+ def handle_cast({:msg, conn_pid, auth, _shard_id, nil, msg}, state) do
+ if not SNet.Group.in_group?(state.group, conn_pid, auth) do
+ # Ignore message
+ {:noreply, state}
+ else
+ state = case msg do
+ {:get_manifest} ->
+ Shard.Manager.send_pid(conn_pid, {state.id, nil, {:manifest, state.manifest}})
state
- else
- # Try adding the message
- {pk, bin, sign} = msgitem
- if Shard.Keys.verify(pk, bin, sign) == :ok do
- if prev_root == state.mst.root do
- # Only one new message, insert it directly
- mst2 = MST.insert(state.mst, msgitem)
- if mst2.root == new_root do
- state = %{state | mst: mst2}
- GenServer.cast(state.page_store, {:set_roots, [mst2.root]})
- Shard.Manager.save_state(state.id, mst2.root)
- msg_callback(state, msgitem)
- state
+ {:append, prev_root, msgitem, new_root} ->
+ # Append message: one single mesage has arrived
+ if new_root == state.mst.root do
+ # We already have the message, do nothing
+ state
+ else
+ # Try adding the message
+ {pk, bin, sign} = msgitem
+ if Shard.Keys.verify(pk, bin, sign) == :ok do
+ if prev_root == state.mst.root do
+ # Only one new message, insert it directly
+ mst2 = MST.insert(state.mst, msgitem)
+ if mst2.root == new_root do
+ state = %{state | mst: mst2}
+ GenServer.cast(state.page_store, {:set_roots, [mst2.root]})
+ Shard.Manager.save_state(state.id, mst2.root)
+ msg_callback(state, msgitem)
+ state
+ else
+ Logger.warn("Invalid new root after inserting same message item!")
+ state
+ end
else
- Logger.warn("Invalid new root after inserting same message item!")
- state
+ # Not a simple one-insertion transition, look at the whole tree
+ init_merge(state, new_root, conn_pid)
end
else
- # Not a simple one-insertion transition, look at the whole tree
- init_merge(state, new_root, conn_pid)
+ Logger.warn("Received message with invalid signature")
+ state
end
- else
- Logger.warn("Received message with invalid signature")
+ end
+ {:root, new_root} ->
+ if new_root == state.mst.root do
+ # already up to date, ignore
state
+ else
+ init_merge(state, new_root, conn_pid)
end
- end
- {:root, new_root} ->
- if new_root == state.mst.root do
- # already up to date, ignore
+ x ->
+ Logger.info("Unhandled message: #{inspect x}")
state
- else
- init_merge(state, new_root, conn_pid)
- end
- x ->
- Logger.info("Unhandled message: #{inspect x}")
- state
+ end
+ {:noreply, state}
end
- {:noreply, state}
end
defp init_merge(state, new_root, source_peer_pid) do
@@ -260,7 +255,7 @@ defmodule SApp.Chat do
end
end
- defp msg_cmp({pk1, msgbin1, _sign1}, {pk2, msgbin2, _sign2}) do
+ def msg_cmp({pk1, msgbin1, _sign1}, {pk2, msgbin2, _sign2}) do
{ts1, msg1} = SData.term_unbin msgbin1
{ts2, msg2} = SData.term_unbin msgbin2
cond do
diff --git a/shard/lib/app/identity.ex b/shard/lib/app/identity.ex
index 6909ad3..d2748a1 100644
--- a/shard/lib/app/identity.ex
+++ b/shard/lib/app/identity.ex
@@ -50,20 +50,17 @@ defmodule SApp.Identity do
def find_proc(pk) do
manifest = %Manifest{pk: pk}
id = SData.term_hash manifest
- Shard.Manager.find_proc id
+ case Shard.Manager.find_proc id do
+ nil ->
+ Shard.Manifest.start manifest
+ pid -> pid
+ end
end
def get_nick(pk) do
- case find_proc pk do
- nil ->
- if Shard.Keys.valid_identity_pk? pk do
- Shard.Manifest.start %Manifest{pk: pk}
- end
- default_nick pk
- pid ->
- info = GenServer.call(pid, :get_info)
- info.nick
- end
+ pid = find_proc pk
+ info = GenServer.call(pid, :get_info)
+ info.nick
end
def handle_call(:manifest, _from, state) do
diff --git a/shard/lib/manager.ex b/shard/lib/manager.ex
index 08e14c7..1e089ba 100644
--- a/shard/lib/manager.ex
+++ b/shard/lib/manager.ex
@@ -315,4 +315,8 @@ defmodule Shard.Manager do
def list_connections() do
for [x] <- :ets.match(:connections, :"$1"), do: x
end
+
+ def get_connections_to(peer_info) do
+ for {^peer_info, pid, auth} <- :ets.lookup(:connections, peer_info), do: {pid, auth}
+ end
end
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
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-21 13:08:07 +0000