execute room requests in a queue

7 files changed, 101 insertions, 67 deletions

diff --git a/lib/polyjuice/client.ex b/lib/polyjuice/client.ex
index 6cf1c84..39cadf0 100644
--- a/lib/polyjuice/client.ex
+++ b/lib/polyjuice/client.ex
@@ -178,6 +178,18 @@ defmodule Polyjuice.Client do
     def call(client_api, endpoint)
 
     @doc """
+    Execute a function in a queue for a room.
+
+    This is to make sure that, for example, messages are sent in order.
+    """
+    @spec room_queue(
+            client_api :: Polyjuice.Client.API.t(),
+            room_id :: String.t(),
+            func :: function
+          ) :: any
+    def room_queue(client_api, room_id, func)
+
+    @doc """
     Generate a unique transaction ID.
     """
     @spec transaction_id(client_api :: Polyjuice.Client.API.t()) :: String.t()
@@ -244,6 +256,10 @@ defmodule Polyjuice.Client do
       end
     end
 
+    def room_queue(%{id: id}, room_id, func) when is_binary(room_id) and is_function(func) do
+      Mutex.under(Polyjuice.Client.Mutex, {id, room_id}, func)
+    end
+
     def transaction_id(_) do
       "#{Node.self()}_#{:erlang.system_time(:millisecond)}_#{:erlang.unique_integer()}"
     end
diff --git a/lib/polyjuice/client/application.ex b/lib/polyjuice/client/application.ex
index b6c87cc..596fd37 100644
--- a/lib/polyjuice/client/application.ex
+++ b/lib/polyjuice/client/application.ex
@@ -21,7 +21,8 @@ defmodule Polyjuice.Client.Application do
       %{
         id: Polyjuice.Client.ID,
         start: {Agent, :start_link, [fn -> 0 end, [name: Polyjuice.Client.ID]]}
-      }
+      },
+      {Mutex, name: Polyjuice.Client.Mutex}
     ]
 
     Supervisor.start_link(children, strategy: :one_for_one)
diff --git a/lib/polyjuice/client/low_level.ex b/lib/polyjuice/client/low_level.ex
index 2b22c6c..d04f798 100644
--- a/lib/polyjuice/client/low_level.ex
+++ b/lib/polyjuice/client/low_level.ex
@@ -16,9 +16,13 @@ defmodule Polyjuice.Client.LowLevel do
   @moduledoc """
   A lower-level client than `Polyjuice.Client`.
 
-  The application is responsible for taking care of all the details.  For
-  example, after logging in, the application must create a new struct using the
-  access token provided.
+  Compared to `Polyjuice.Client`, this module:
+  - does not manage the access token, user ID, or device ID; whenever these
+    change (e.g. after logging in), the application must create a new client to
+    use the new values
+  - does not provide a process for syncing with the server
+  - does not ensure that room requests (such as sending messages) are run in a
+    queue
   """
 
   require Logger
@@ -93,12 +97,13 @@ defmodule Polyjuice.Client.LowLevel do
       end
     end
 
-    def transaction_id(_) do
-      "#{Node.self()}_#{:erlang.system_time(:millisecond)}_#{:erlang.unique_integer()}"
+    def room_queue(_client_api, _room_id, func) do
+      # this client doesn't have a queue.  Just run the function.
+      func.()
     end
 
-    def sync_child_spec(_client, _listener, _opts \\ []) do
-      # do nothing
+    def transaction_id(_) do
+      "#{Node.self()}_#{:erlang.system_time(:millisecond)}_#{:erlang.unique_integer()}"
     end
   end
 
diff --git a/lib/polyjuice/client/room.ex b/lib/polyjuice/client/room.ex
index 28c164c..db4b71e 100644
--- a/lib/polyjuice/client/room.ex
+++ b/lib/polyjuice/client/room.ex
@@ -61,32 +61,34 @@ defmodule Polyjuice.Client.Room do
           msg :: map | Polyjuice.Client.MsgBuilder.MsgData.t()
         ) :: {:ok, String.t()} | any
   def send_message(client_api, room, msg) when is_binary(room) do
-    cond do
-      Polyjuice.Client.MsgBuilder.MsgData.impl_for(msg) != nil ->
-        Polyjuice.Client.API.call(
-          client_api,
-          %Polyjuice.Client.Endpoint.PutRoomsSend{
-            txn_id: Polyjuice.Client.API.transaction_id(client_api),
-            room: room,
-            event_type: "m.room.message",
-            message: Polyjuice.Client.MsgBuilder.to_message(msg)
-          }
-        )
+    Polyjuice.Client.API.room_queue(client_api, room, fn ->
+      cond do
+        Polyjuice.Client.MsgBuilder.MsgData.impl_for(msg) != nil ->
+          Polyjuice.Client.API.call(
+            client_api,
+            %Polyjuice.Client.Endpoint.PutRoomsSend{
+              txn_id: Polyjuice.Client.API.transaction_id(client_api),
+              room: room,
+              event_type: "m.room.message",
+              message: Polyjuice.Client.MsgBuilder.to_message(msg)
+            }
+          )
 
-      is_map(msg) and not Map.has_key?(msg, :__struct__) ->
-        Polyjuice.Client.API.call(
-          client_api,
-          %Polyjuice.Client.Endpoint.PutRoomsSend{
-            txn_id: Polyjuice.Client.API.transaction_id(client_api),
-            room: room,
-            event_type: "m.room.message",
-            message: msg
-          }
-        )
+        is_map(msg) and not Map.has_key?(msg, :__struct__) ->
+          Polyjuice.Client.API.call(
+            client_api,
+            %Polyjuice.Client.Endpoint.PutRoomsSend{
+              txn_id: Polyjuice.Client.API.transaction_id(client_api),
+              room: room,
+              event_type: "m.room.message",
+              message: msg
+            }
+          )
 
-      true ->
-        raise ArgumentError, message: "invalid argument msg"
-    end
+        true ->
+          raise ArgumentError, message: "invalid argument msg"
+      end
+    end)
   end
 
   @doc """
@@ -100,15 +102,17 @@ defmodule Polyjuice.Client.Room do
         ) :: {:ok, String.t()} | any
   def send_event(client_api, room, event_type, event)
       when is_binary(event_type) and is_map(event) and is_binary(room) do
-    Polyjuice.Client.API.call(
-      client_api,
-      %Polyjuice.Client.Endpoint.PutRoomsSend{
-        txn_id: Polyjuice.Client.API.transaction_id(client_api),
-        room: room,
-        event_type: event_type,
-        message: event
-      }
-    )
+    Polyjuice.Client.API.room_queue(client_api, room, fn ->
+      Polyjuice.Client.API.call(
+        client_api,
+        %Polyjuice.Client.Endpoint.PutRoomsSend{
+          txn_id: Polyjuice.Client.API.transaction_id(client_api),
+          room: room,
+          event_type: event_type,
+          message: event
+        }
+      )
+    end)
   end
 
   @doc """
@@ -123,15 +127,17 @@ defmodule Polyjuice.Client.Room do
         ) :: {:ok, String.t()} | any
   def send_state_event(client_api, room, event_type, state_key \\ "", event)
       when is_binary(event_type) and is_binary(state_key) and is_map(event) and is_binary(room) do
-    Polyjuice.Client.API.call(
-      client_api,
-      %Polyjuice.Client.Endpoint.PutRoomsState{
-        room: room,
-        event_type: event_type,
-        state_key: state_key,
-        content: event
-      }
-    )
+    Polyjuice.Client.API.room_queue(client_api, room, fn ->
+      Polyjuice.Client.API.call(
+        client_api,
+        %Polyjuice.Client.Endpoint.PutRoomsState{
+          room: room,
+          event_type: event_type,
+          state_key: state_key,
+          content: event
+        }
+      )
+    end)
   end
 
   @doc """
@@ -168,14 +174,16 @@ defmodule Polyjuice.Client.Room do
   def join(client_api, room, servers \\ [], third_party_signed \\ nil)
       when is_binary(room) and is_list(servers) and
              (is_map(third_party_signed) or third_party_signed == nil) do
-    Polyjuice.Client.API.call(
-      client_api,
-      %Polyjuice.Client.Endpoint.PostJoin{
-        room: room,
-        servers: servers,
-        third_party_signed: third_party_signed
-      }
-    )
+    Polyjuice.Client.API.room_queue(client_api, room, fn ->
+      Polyjuice.Client.API.call(
+        client_api,
+        %Polyjuice.Client.Endpoint.PostJoin{
+          room: room,
+          servers: servers,
+          third_party_signed: third_party_signed
+        }
+      )
+    end)
   end
 
   @doc """
diff --git a/mix.exs b/mix.exs
index 1600134..4addbbd 100644
--- a/mix.exs
+++ b/mix.exs
@@ -52,7 +52,8 @@ defmodule PolyjuiceClient.MixProject do
       {:ex_doc, "~> 0.21", only: :dev, runtime: false},
       {:hackney, "~> 1.12"},
       {:jason, "~> 1.2"},
-      {:polyjuice_util, "~> 0.1.0"}
+      {:polyjuice_util, "~> 0.1.0"},
+      {:mutex, "~> 1.1.3"}
     ]
   end
 
diff --git a/mix.lock b/mix.lock
index 97bf556..401aaa0 100644
--- a/mix.lock
+++ b/mix.lock
@@ -9,6 +9,7 @@
   "makeup_elixir": {:hex, :makeup_elixir, "0.14.0", "cf8b7c66ad1cff4c14679698d532f0b5d45a3968ffbcbfd590339cb57742f1ae", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}], "hexpm"},
   "metrics": {:hex, :metrics, "1.0.1", "25f094dea2cda98213cecc3aeff09e940299d950904393b2a29d191c346a8486", [:rebar3], [], "hexpm"},
   "mimerl": {:hex, :mimerl, "1.2.0", "67e2d3f571088d5cfd3e550c383094b47159f3eee8ffa08e64106cdf5e981be3", [:rebar3], [], "hexpm"},
+  "mutex": {:hex, :mutex, "1.1.3", "d7e19f96fe19d6d97583bf12ca1ec182bbf14619b7568592cc461135de1c3b81", [:mix], [], "hexpm"},
   "nimble_parsec": {:hex, :nimble_parsec, "0.5.1", "c90796ecee0289dbb5ad16d3ad06f957b0cd1199769641c961cfe0b97db190e0", [:mix], [], "hexpm"},
   "parse_trans": {:hex, :parse_trans, "3.3.0", "09765507a3c7590a784615cfd421d101aec25098d50b89d7aa1d66646bc571c1", [:rebar3], [], "hexpm"},
   "polyjuice_util": {:hex, :polyjuice_util, "0.1.0", "69901959c143245b47829c8302d0605dff6c0e1c3b116730c162982e0f512ee0", [:mix], [], "hexpm"},
diff --git a/test/support/dummy_client.ex b/test/support/dummy_client.ex
index 0a6ad21..7c1556e 100644
--- a/test/support/dummy_client.ex
+++ b/test/support/dummy_client.ex
@@ -26,11 +26,12 @@ defmodule DummyClient do
       result
     end
 
-    def transaction_id(_), do: "txn_id"
-
-    def sync_child_spec(_client_api, _listener, _opts \\ []) do
-      %{}
+    def room_queue(_client_api, _room_id, func) do
+      # this client doesn't have a queue.  Just run the function.
+      func.()
     end
+
+    def transaction_id(_), do: "txn_id"
   end
 
   defmodule MultiReq do
@@ -71,11 +72,12 @@ defmodule DummyClient do
         result
       end
 
-      def transaction_id(_), do: "txn_id"
-
-      def sync_child_spec(_client_api, _listener, _opts \\ []) do
-        %{}
+      def room_queue(_client_api, _room_id, func) do
+        # this client doesn't have a queue.  Just run the function.
+        func.()
       end
+
+      def transaction_id(_), do: "txn_id"
     end
   end
 end