Big fixes (1 line), small changes (many lines)

4 files changed, 77 insertions, 11 deletions

diff --git a/lib/app/chat.ex b/lib/app/chat.ex
index 42991ce..a9cfb1e 100644
--- a/lib/app/chat.ex
+++ b/lib/app/chat.ex
@@ -39,7 +39,15 @@ defmodule SApp.Chat do
     GenServer.cast(Shard.Manager, {:register, id, manifest, self()})
     GenServer.cast(self(), :init_pull)
 
-    {:ok, %{channel: channel, id: id, manifest: manifest, store: store, peers: MapSet.new}}
+    {:ok,
+      %{channel: channel,
+        id: id,
+        manifest: manifest,
+        store: store, 
+        peers: MapSet.new,
+        subs: MapSet.new,
+      }
+    }
   end
 
   @doc """
@@ -49,6 +57,11 @@ defmodule SApp.Chat do
     {:reply, state.manifest, state}
   end
 
+  def handle_call({:read_history, start, num}, _from, state) do
+    ret = ML.read(state.store, start, num)
+    {:reply, ret, state}
+  end
+
   @doc """
   Implementation of the :redundant handler: if another process is already
   synchronizing this channel then we exit.
@@ -80,6 +93,12 @@ defmodule SApp.Chat do
                msg}
     new_state = %{state | store: ML.insert(state.store, msgitem)}
 
+    for pid <- state.subs do
+      if Process.alive?(pid) do
+        send(pid, {:chat_send, state.channel, msgitem})
+      end
+    end
+
     for peer <- state.peers do
       push_messages(new_state, peer, nil, 5)
     end
@@ -97,6 +116,11 @@ defmodule SApp.Chat do
     {:noreply, %{ state | peers: new_peers }}
   end
 
+  def handle_cast({:subscribe, pid}, state) do
+    new_subs = MapSet.put(state.subs, pid)
+    {:noreply, %{ state | subs: new_subs }}
+  end
+
   @doc """
   Implementation of the :msg handler, which is the main handler for messages
   comming from other peers concerning this chat room.
@@ -110,7 +134,7 @@ defmodule SApp.Chat do
     case msg do
       {:get_manifest} ->
         Shard.Manager.send(peer_id, {state.id, {:manifest, state.manifest}})
-      {:get, start} -> push_messages(peer_id, state, start, 20)
+      {:get, start} -> push_messages(state, peer_id, start, 20)
       {:info, _start, list, rest} ->
         if rest != nil and not ML.has(state.store, rest) do
           Shard.Manager.send(peer_id, {state.id, {:get, rest}})
@@ -131,7 +155,7 @@ defmodule SApp.Chat do
   end
 
   def handle_cast({:deferred_insert, list}, state) do
-    new_store = ML.insert_many(state.store, list, (fn msg -> msg_callback(state.channel, msg) end))
+    new_store = ML.insert_many(state.store, list, (fn msg -> msg_callback(state, msg) end))
     {:noreply, %{state | store: new_store}}
   end
 
@@ -143,8 +167,12 @@ defmodule SApp.Chat do
     end
   end
 
-  defp msg_callback(chan, {ts, nick, msg}) do
-    IO.puts "#{ts |> DateTime.from_unix! |> DateTime.to_iso8601}  ##{chan}  <#{nick}> #{msg}"
+  defp msg_callback(state, {ts, nick, msg}) do
+    for pid <- state.subs do
+      if Process.alive?(pid) do
+        send(pid, {:chat_recv, state.channel, {ts, nick, msg}})
+      end
+    end
   end
 
   defp msg_cmp({ts1, nick1, msg1}, {ts2, nick2, msg2}) do
diff --git a/lib/cli/cli.ex b/lib/cli/cli.ex
index c281d57..8928040 100644
--- a/lib/cli/cli.ex
+++ b/lib/cli/cli.ex
@@ -8,6 +8,8 @@ defmodule SCLI do
   end
 
   defp run(pid) do
+    handle_messages()
+
     nick = Shard.Identity.get_nickname
     prompt = case pid do
       nil -> "(no channel) #{nick}: "
@@ -25,11 +27,25 @@ defmodule SCLI do
         pid2 = handle_command(pid, command)
         run(pid2)
       true -> 
-        GenServer.cast(pid, {:chat_send, str})
+        if str != "" do
+          GenServer.cast(pid, {:chat_send, str})
+        end
         run(pid)
     end
   end
 
+  defp handle_messages() do
+    receive do
+      {:chat_recv, chan, {ts, nick, msg}} ->
+        IO.puts "#{ts |> DateTime.from_unix! |> DateTime.to_iso8601}  ##{chan}  <#{nick}> #{msg}"
+        handle_messages()
+      {:chat_send, _, _} ->
+        # do nothing
+        handle_messages()
+    after 10 -> nil
+    end
+  end
+
   defp handle_command(pid, ["connect", ipstr, portstr]) do
     {:ok, ip} = :inet.parse_address (to_charlist ipstr)
     {port, _} = Integer.parse portstr
@@ -47,6 +63,19 @@ defmodule SCLI do
     pid
   end
 
+  defp handle_command(pid, ["hist"]) do
+    case GenServer.call(pid, {:read_history, nil, 100}) do
+      {:ok, list, _rest} ->
+        list
+        |> Enum.reverse
+        |> Enum.each(fn {ts, nick, msg} ->
+          IO.puts "#{ts |> DateTime.from_unix! |> DateTime.to_iso8601}  <#{nick}> #{msg}"
+        end)
+      _ -> nil
+    end
+    pid
+  end
+
   defp handle_command(_pid, ["join", qchan]) do
     list = for {_chid, manifest, chpid} <- :ets.tab2list(:shard_db),
                {:chat, chan} = manifest,
@@ -54,6 +83,7 @@ defmodule SCLI do
     case List.keyfind(list, qchan, 0) do
       nil ->
         {:ok, pid} = DynamicSupervisor.start_child(Shard.DynamicSupervisor, {SApp.Chat, qchan})
+        GenServer.cast(pid, {:subscribe, self()})
         pid
       {_, pid} ->
         IO.puts "Switching to ##{qchan}"
diff --git a/lib/data/merklelist.ex b/lib/data/merklelist.ex
index 71483bd..357f336 100644
--- a/lib/data/merklelist.ex
+++ b/lib/data/merklelist.ex
@@ -78,7 +78,7 @@ defmodule SData.MerkleList do
             callback.(item)
             new_state
           :duplicate -> insert_many_aux(state, rest, callback)
-          :before -> push(insert_many_aux(state_rest, [item | rest], callback), item)
+          :before -> push(insert_many_aux(state_rest, [item | rest], callback), front)
         end
     end
   end
diff --git a/lib/manager.ex b/lib/manager.ex
index 45aae5f..87f95c5 100644
--- a/lib/manager.ex
+++ b/lib/manager.ex
@@ -28,6 +28,8 @@ defmodule Shard.Manager do
 
   use GenServer
 
+  require Logger
+
   def start_link(my_port) do
     GenServer.start_link(__MODULE__, my_port, name: __MODULE__)
   end
@@ -96,7 +98,9 @@ defmodule Shard.Manager do
         add_peer(ip, port, state)
         currtime = System.os_time :second
         :ets.insert(state.outbox, {peer_id, msg, currtime})
-        outbox_cleanup = :ets.fun2ms(fn {_, _, t} when t < currtime - 60 -> true end)
+        outbox_cleanup = [{{:_, :_, :'$1'},
+                           [{:<, :'$1', currtime - 60}],
+                           [:'$1']}]
         :ets.select_delete(state.outbox, outbox_cleanup)
       _ -> nil
     end
@@ -121,9 +125,13 @@ defmodule Shard.Manager do
 
   defp add_peer(ip, port, state) do
     spawn fn ->
-      {:ok, client} = :gen_tcp.connect(ip, port, [:binary, packet: 2, active: false])
-      {:ok, pid} = DynamicSupervisor.start_child(Shard.DynamicSupervisor, {SNet.TCPConn, %{socket: client, my_port: state.my_port}})
-      :ok = :gen_tcp.controlling_process(client, pid)
+      case :gen_tcp.connect(ip, port, [:binary, packet: 2, active: false]) do
+        {:ok, client} ->
+          {:ok, pid} = DynamicSupervisor.start_child(Shard.DynamicSupervisor, {SNet.TCPConn, %{socket: client, my_port: state.my_port}})
+          :ok = :gen_tcp.controlling_process(client, pid)
+        _ ->
+          Logger.info "Could not connect to #{inspect ip}:#{port}, some messages may be dropped"
+        end
     end
   end