test: cover blockingTask, awaitWaiters and Excache.CommandHandshakeTest

test/excache/blocking_test.exs

@@ -310,3 +310,135 @@ defmodule Excache.BlockingTest do
       assert Task.await(waiter) == "moved"
       assert {["moved"], conn} = Client.command(conn, ["LRANGE", "dst", "0", "-1"])
 
+      assert {nil, _conn} = Client.command(conn, ["BRPOPLPUSH", "empty", "dst", "0.1"])
+    end
+
+    test "BLMOVE honours both sides while blocked", %{conn: conn, port: port} do
+      waiter = blockingTask(port, ["BLMOVE", "src", "dst", "LEFT", "RIGHT", "0"])
+      awaitWaiters("src", 1)
+
+      assert {2, conn} = Client.command(conn, ["RPUSH", "src", "head", "tail"])
+      assert Task.await(waiter) == "head"
+
+      assert {["tail"], conn} = Client.command(conn, ["LRANGE", "src", "0", "-1"])
+      assert {2, conn} = Client.command(conn, ["RPUSH", "dst", "existing"])
+
+      assert {"tail", conn} = Client.command(conn, ["BLMOVE", "src", "dst", "RIGHT", "LEFT", "0"])
+
+      assert {["tail", "head", "existing"], _conn} =
+               Client.command(conn, ["LRANGE", "dst", "0", "-1"])
+    end
+
+    test "a BLMOVE destination wakes a client blocked on it", %{conn: conn, port: port} do
+      consumer = blockingTask(port, ["BLPOP", "stage2", "0"])
+      awaitWaiters("stage2", 1)
+
+      mover = blockingTask(port, ["BLMOVE", "stage1", "stage2", "LEFT", "RIGHT", "0"])
+      awaitWaiters("stage1", 1)
+
+      assert {1, conn} = Client.command(conn, ["LPUSH", "stage1", "item"])
+
+      assert Task.await(mover) == "item"
+      assert Task.await(consumer) == ["stage2", "item"]
+      assert {0, _conn} = Client.command(conn, ["EXISTS", "stage2"])
+    end
+
+    test "a rename that replaces a waited key with another type reports WRONGTYPE", %{
+      conn: conn,
+      port: port
+    } do
+      waiter = blockingTask(port, ["BLPOP", "renamed", "0"])
+      awaitWaiters("renamed", 1)
+
+      assert {{:simple, "OK"}, conn} = Client.command(conn, ["SET", "plain", "v"])
+      assert {{:simple, "OK"}, _conn} = Client.command(conn, ["RENAME", "plain", "renamed"])
+
+      assert {:error, "WRONGTYPE" <> _} = Task.await(waiter)
+    end
+  end
+
+  describe "BLMPOP" do
+    test "pops up to COUNT elements from the first non empty key", %{conn: conn} do
+      assert {3, conn} = Client.command(conn, ["RPUSH", "mp", "a", "b", "c"])
+
+      assert {["mp", ["a", "b"]], conn} =
+               Client.command(conn, ["BLMPOP", "0", "2", "missing", "mp", "LEFT", "COUNT", "2"])
+
+      assert {["mp", ["c"]], _conn} =
+               Client.command(conn, ["BLMPOP", "0", "1", "mp", "LEFT"])
+    end
+
+    test "blocks across several keys and answers RIGHT side", %{conn: conn, port: port} do
+      waiter = blockingTask(port, ["BLMPOP", "0", "2", "mp1", "mp2", "RIGHT", "COUNT", "2"])
+      awaitWaiters("mp2", 1)
+
+      assert {3, conn} = Client.command(conn, ["RPUSH", "mp2", "x", "y", "z"])
+      assert Task.await(waiter) == ["mp2", ["z", "y"]]
+      assert {["x"], _conn} = Client.command(conn, ["LRANGE", "mp2", "0", "-1"])
+    end
+
+    test "times out with a nil array", %{conn: conn} do
+      assert {nil, _conn} = Client.command(conn, ["BLMPOP", "0.1", "1", "mp", "LEFT"])
+    end
+
+    test "argument errors", %{conn: conn} do
+      assert {{:error, "ERR numkeys should be greater than 0"}, conn} =
+               Client.command(conn, ["BLMPOP", "0", "0", "mp", "LEFT"])
+
+      assert {{:error, "ERR syntax error"}, conn} =
+               Client.command(conn, ["BLMPOP", "0", "1", "mp", "SIDEWAYS"])
+
+      assert {{:error, "ERR count should be greater than 0"}, conn} =
+               Client.command(conn, ["BLMPOP", "0", "1", "mp", "LEFT", "COUNT", "0"])
+
+      assert {{:error, "ERR syntax error"}, _conn} =
+               Client.command(conn, ["BLMPOP", "0", "2", "mp", "LEFT"])
+    end
+  end
+
+  defp blockingTask(port, args) do
+    Task.async(fn ->
+      client = Client.connect(port)
+      {reply, _client} = Client.command(client, args)
+      Client.close(client)
+      reply
+    end)
+  end
+
+  defp awaitWaiters(key, count) do
+    awaitUntil(fn -> Blocking.waitingCount(key) == count end)
+  end
+
+  defp awaitUntil(check, attempts \\ 400) do
+    cond do
+      check.() ->
+        :ok
+
+      attempts == 0 ->
+        flunk("the expected registry state was never reached")
+
+      true ->
+        Process.sleep(5)
+        awaitUntil(check, attempts - 1)
+    end
+  end
+
+  defp rawConnect(port) do
+    {:ok, socket} = :gen_tcp.connect(~c"127.0.0.1", port, [:binary, active: false, packet: :raw])
+    socket
+  end
+
+  defp readBytes(socket, size) do
+    {:ok, bytes} = :gen_tcp.recv(socket, size, 2000)
+    bytes
+  end
+
+  defp encodeCommand(args) do
+    [
+      "*",
+      Integer.to_string(length(args)),
+      "\r\n",
+      Enum.map(args, fn arg -> ["$", Integer.to_string(byte_size(arg)), "\r\n", arg, "\r\n"] end)
+    ]
+  end
+end

test/excache/command_handshake_test.exs

@@ -0,0 +1,50 @@
+defmodule Excache.CommandHandshakeTest do
+  @moduledoc """
+  Client libraries ask the server what it can do before they send anything else.
+  Until the catalog was wired in, `COMMAND` answered with an empty array over the
+  wire - a server that claims to support nothing. These tests pin the answer to
+  the real dispatch table, over a real socket.
+  """
+
+  use Excache.Test.ServerCase, async: false
+
+  test "COMMAND reports the whole dispatch table over the wire", %{conn: conn} do
+    {entries, conn} = Client.command(conn, ["COMMAND"])
+
+    assert is_list(entries)
+    assert length(entries) == length(Enum.uniq(Excache.Command.known_commands()))
+
+    names =
+      entries
+      |> Enum.map(fn [name | _rest] -> String.upcase(name) end)
+      |> MapSet.new()
+
+    for command <- ["GET", "SET", "BLPOP", "MEMORY", "CONFIG", "SCAN", "XADD"] do
+      assert MapSet.member?(names, command), "#{command} missing from COMMAND"
+    end
+
+    {count, _conn} = Client.command(conn, ["COMMAND", "COUNT"])
+    assert count == length(entries)
+  end
+
+  test "COMMAND INFO carries the arity and key positions a router needs", %{conn: conn} do
+    {[["get", arity, flags, firstKey, lastKey, step]], conn} =
+      Client.command(conn, ["COMMAND", "INFO", "GET"])
+
+    assert arity == 2
+    assert firstKey == 1
+    assert lastKey == 1
+    assert step == 1
+    assert is_list(flags)
+
+    {[entry], _conn} = Client.command(conn, ["COMMAND", "INFO", "nosuchcommand"])
+    assert entry == nil
+  end
+
+  test "COMMAND GETKEYS extracts the keys a command touches", %{conn: conn} do
+    assert {["one", "two"], conn} =
+             Client.command(conn, ["COMMAND", "GETKEYS", "MSET", "one", "1", "two", "2"])
+
+    assert {["only"], _conn} = Client.command(conn, ["COMMAND", "GETKEYS", "GET", "only"])
+  end
+end