Avoid messaging handled_event_ids and write/1

Author: progval

lib/matrix_client/poller.ex

@@ -33,18 +33,17 @@ defmodule M51.MatrixClient.Poller do
 
     # If we are being restarted, pick up from where the last process stopped.
     since = M51.MatrixClient.State.poll_since_marker(state)
-    handled_event_ids = M51.MatrixClient.State.handled_events(state)
 
     if M51.IrcConn.State.registered(irc_state) do
-      loop_poll(sup_pid, since, handled_event_ids)
+      loop_poll(sup_pid, since)
     else
       receive do
-        :start_polling -> loop_poll(sup_pid, since, handled_event_ids)
+        :start_polling -> loop_poll(sup_pid, since)
       end
     end
   end
 
-  def loop_poll(sup_pid, since, handled_event_ids \\ MapSet.new()) do
+  def loop_poll(sup_pid, since) do
     client = M51.IrcConn.Supervisor.matrix_client(sup_pid)
     state = M51.IrcConn.Supervisor.matrix_state(sup_pid)
 
@@ -56,14 +55,13 @@ defmodule M51.MatrixClient.Poller do
         end
 
       raw_client ->
-        since = poll_one(sup_pid, since, handled_event_ids, raw_client)
+        since = poll_one(sup_pid, since, raw_client)
         M51.MatrixClient.State.update_poll_since_marker(state, since)
-        # do not pass handled_event_ids, no longer needed
         loop_poll(sup_pid, since)
     end
   end
 
-  defp poll_one(sup_pid, since, handled_event_ids, raw_client) do
+  defp poll_one(sup_pid, since, raw_client) do
     query = %{
       # Completely arbitrary value. Just make sure it's lower than recv_timeout below
       "timeout" => "600000"
@@ -79,30 +77,19 @@ defmodule M51.MatrixClient.Poller do
 
     case M51.Matrix.RawClient.get(raw_client, path, [], options) do
       {:ok, events} ->
-        handle_events(sup_pid, is_backlog, events, handled_event_ids)
+        handle_events(sup_pid, is_backlog, events)
         events["next_batch"]
 
       {:error, code, _} when code >= 500 and code < 600 ->
         # server error, try again
-        poll_one(sup_pid, since, handled_event_ids, raw_client)
+        poll_one(sup_pid, since, raw_client)
     end
   end
 
   @doc """
     Internal method that dispatches event; public only so it can be unit-tested.
   """
-  def handle_events(sup_pid, is_backlog, events, handled_event_ids \\ MapSet.new()) do
-    irc_state = M51.IrcConn.Supervisor.state(sup_pid)
-    capabilities = M51.IrcConn.State.capabilities(irc_state)
-    writer = M51.IrcConn.Supervisor.writer(sup_pid)
-
-    write = fn cmd ->
-      M51.IrcConn.Writer.write_command(
-        writer,
-        M51.Irc.Command.downgrade(cmd, capabilities)
-      )
-    end
-
+  def handle_events(sup_pid, is_backlog, events) do
     events
     |> Map.get("rooms", %{})
     |> Map.get("join", %{})
@@ -113,8 +100,6 @@ defmodule M51.MatrixClient.Poller do
         room_id,
         :join,
         is_backlog,
-        handled_event_ids,
-        write,
         event
       )
     end)
@@ -129,8 +114,6 @@ defmodule M51.MatrixClient.Poller do
         room_id,
         :leave,
         is_backlog,
-        handled_event_ids,
-        write,
         event
       )
     end)
@@ -145,8 +128,6 @@ defmodule M51.MatrixClient.Poller do
         room_id,
         :invite,
         is_backlog,
-        handled_event_ids,
-        write,
         event
       )
     end)
@@ -204,7 +185,7 @@ defmodule M51.MatrixClient.Poller do
           handle_event(sup_pid, room_id, sender, is_backlog, write, event)
           # Don't mark it handled right now, there is still some processing to
           # do below.
-          # M51.MatrixClient.State.mark_handled_event(state, event_id)
+          # M51.MatrixClient.State.mark_handled_event(state, room_id, event_id)
         end
       end)
 
@@ -244,7 +225,7 @@ defmodule M51.MatrixClient.Poller do
 
         handle_event(sup_pid, room_id, sender, is_backlog, write, event)
 
-        M51.MatrixClient.State.mark_handled_event(state, event_id)
+        M51.MatrixClient.State.mark_handled_event(state, room_id, event_id)
       end
     end)
   end
@@ -1121,7 +1102,7 @@ defmodule M51.MatrixClient.Poller do
             nil
         end
 
-        M51.MatrixClient.State.mark_handled_event(state, event_id)
+        M51.MatrixClient.State.mark_handled_event(state, room_id, event_id)
       end
     end)
   end

lib/matrix_client/room_handler.ex

@@ -37,49 +37,52 @@ defmodule M51.MatrixClient.RoomHandler do
   end
 
   @impl true
-  def handle_cast({:events, :join, is_backlog, handled_event_ids, write, events}, state) do
+  def handle_cast({:events, room_type, is_backlog, events}, state) do
     {sup_pid, room_id} = state
+    state_pid = M51.IrcConn.Supervisor.matrix_state(sup_pid)
+    irc_state = M51.IrcConn.Supervisor.state(sup_pid)
+    capabilities = M51.IrcConn.State.capabilities(irc_state)
+    writer = M51.IrcConn.Supervisor.writer(sup_pid)
+    handled_event_ids = M51.MatrixClient.State.handled_events(state_pid, room_id)
 
-    M51.MatrixClient.Poller.handle_joined_room(
-      sup_pid,
-      is_backlog,
-      handled_event_ids,
-      room_id,
-      write,
-      events
-    )
-
-    {:noreply, state}
-  end
-
-  @impl true
-  def handle_cast({:events, :leave, is_backlog, handled_event_ids, write, events}, state) do
-    {sup_pid, room_id} = state
+    write = fn cmd ->
+      M51.IrcConn.Writer.write_command(
+        writer,
+        M51.Irc.Command.downgrade(cmd, capabilities)
+      )
+    end
 
-    M51.MatrixClient.Poller.handle_left_room(
-      sup_pid,
-      is_backlog,
-      handled_event_ids,
-      room_id,
-      write,
-      events
-    )
+    case room_type do
+      :join ->
+        M51.MatrixClient.Poller.handle_joined_room(
+          sup_pid,
+          is_backlog,
+          handled_event_ids,
+          room_id,
+          write,
+          events
+        )
 
-    {:noreply, state}
-  end
+      :leave ->
+        M51.MatrixClient.Poller.handle_left_room(
+          sup_pid,
+          is_backlog,
+          handled_event_ids,
+          room_id,
+          write,
+          events
+        )
 
-  @impl true
-  def handle_cast({:events, :invite, is_backlog, handled_event_ids, write, events}, state) do
-    {sup_pid, room_id} = state
-
-    M51.MatrixClient.Poller.handle_invited_room(
-      sup_pid,
-      is_backlog,
-      handled_event_ids,
-      room_id,
-      write,
-      events
-    )
+      :invite ->
+        M51.MatrixClient.Poller.handle_invited_room(
+          sup_pid,
+          is_backlog,
+          handled_event_ids,
+          room_id,
+          write,
+          events
+        )
+    end
 
     {:noreply, state}
   end

lib/matrix_client/room_supervisor.ex

@@ -56,15 +56,9 @@ defmodule M51.MatrixClient.RoomSupervisor do
     end
   end
 
-  def handle_events(sup_pid, room_id, type, is_backlog, handled_event_ids, write, events) do
-    # TODO: fetch handled_event_ids in the server instead of message-passing it
-    # TODO: define write/1 in the server instead of message-passing it
-
+  def handle_events(sup_pid, room_id, type, is_backlog, events) do
     room_handler_pid = M51.MatrixClient.RoomSupervisor.start_or_get_room_handler(sup_pid, room_id)
 
-    GenServer.cast(
-      room_handler_pid,
-      {:events, type, is_backlog, handled_event_ids, write, events}
-    )
+    GenServer.cast(room_handler_pid, {:events, type, is_backlog, events})
   end
 end

lib/matrix_client/state.ex

@@ -26,7 +26,8 @@ defmodule M51.MatrixClient.State do
     # events handled since the last update to :poll_since (the poller updates
     # this set as it handles events in a batch; then updates :poll_since
     # an resets this set when it is done with a batch).
-    handled_events: MapSet.new(),
+    # Stored as a Map from room ids to the set of event ids.
+    handled_events: Map.new(),
     # %{channel name => list of callbacks to run when a room
     #                   with that channel name is completely synced }
     channel_sync_callbacks: Map.new()
@@ -263,23 +264,28 @@ defmodule M51.MatrixClient.State do
     Agent.get(pid, fn state -> state.poll_since end)
   end
 
-  def handled_events(pid) do
-    Agent.get(pid, fn state -> state.handled_events end)
+  def handled_events(pid, room_id) do
+    Agent.get(pid, fn state -> Map.get(state.handled_events, room_id) || MapSet.new() end)
   end
 
   @doc """
     Updates the 'since' marker, and resets the 'handled_events' set.
   """
   def update_poll_since_marker(pid, new_since_marker) do
     Agent.update(pid, fn state ->
-      %{state | poll_since: new_since_marker, handled_events: MapSet.new()}
+      %{state | poll_since: new_since_marker, handled_events: Map.new()}
     end)
   end
 
-  def mark_handled_event(pid, event_id) do
+  def mark_handled_event(pid, room_id, event_id) do
     if event_id != nil do
       Agent.update(pid, fn state ->
-        %{state | handled_events: MapSet.put(state.handled_events, event_id)}
+        handled_events =
+          Map.update(state.handled_events, room_id, nil, fn event_ids ->
+            MapSet.put(event_ids || MapSet.new(), event_id)
+          end)
+
+        %{state | handled_events: handled_events}
       end)
     end
   end