From 6de76acbff93b5c3d800678b4efe4c87c59537b9 Mon Sep 17 00:00:00 2001 From: DualFroz Date: Fri, 4 Sep 2026 17:08:12 +0200 Subject: [PATCH] feat(command): extend blocking with Excache.Command.Blocking --- lib/excache/command/blocking.ex | 119 ++++++++++++++++++++++++++++++++ lib/excache/command/extended.ex | 14 +++- 2 files changed, 131 insertions(+), 2 deletions(-) create mode 100644 lib/excache/command/blocking.ex diff --git a/lib/excache/command/blocking.ex b/lib/excache/command/blocking.ex new file mode 100644 index 0000000..bb1a541 --- /dev/null +++ b/lib/excache/command/blocking.ex @@ -0,0 +1,119 @@ +defmodule Excache.Command.Blocking do + @moduledoc false + + alias Excache.Command.Reply + + @names ~w(BLPOP BRPOP BLMOVE BRPOPLPUSH BLMPOP) + + def commandNames, do: @names + + def dispatch("BLPOP", args, state), do: popCommand(args, :left, state) + def dispatch("BRPOP", args, state), do: popCommand(args, :right, state) + + def dispatch("BLMOVE", [source, destination, fromRaw, toRaw, timeoutRaw], state) do + with {:ok, fromSide} <- parseSide(fromRaw), + {:ok, toSide} <- parseSide(toRaw) do + withTimeout(timeoutRaw, state, fn timeout -> + submit([source], {:move, fromSide, toSide, destination}, timeout, state) + end) + else + :error -> {[Reply.syntax()], state} + end + end + + def dispatch("BRPOPLPUSH", [source, destination, timeoutRaw], state) do + withTimeout(timeoutRaw, state, fn timeout -> + submit([source], {:move, :right, :left, destination}, timeout, state) + end) + end + + def dispatch("BLMPOP", [timeoutRaw, numkeysRaw | rest], state) when rest != [] do + withTimeout(timeoutRaw, state, fn timeout -> + case parseMpop(numkeysRaw, rest) do + {:ok, keys, side, count} -> submit(keys, {:mpop, side, count}, timeout, state) + {:reply, term} -> {[term], state} + end + end) + end + + def dispatch(_name, _args, _state), do: :unhandled + + defp popCommand(args, side, state) when length(args) >= 2 do + {keys, [timeoutRaw]} = Enum.split(args, length(args) - 1) + + withTimeout(timeoutRaw, state, fn timeout -> + submit(keys, {:pop, side}, timeout, state) + end) + end + + defp popCommand(_args, _side, _state), do: :unhandled + + defp submit(keys, op, timeout, state) do + request = %{keys: keys, op: op, timeout: timeout} + + if state.in_multi do + {[Excache.Blocking.attempt(request)], state} + else + case Excache.Blocking.acquire(request) do + {:served, term} -> {[term], state} + {:blocked, ref} -> {[{:blocked, ref}], state} + end + end + end + + defp withTimeout(raw, state, fun) do + case parseTimeout(raw) do + {:ok, timeout} -> fun.(timeout) + {:reply, term} -> {[term], state} + end + end + + defp parseTimeout(raw) do + case Reply.parse_score(raw) do + {:ok, seconds} when seconds < 0 -> {:reply, {:error, "ERR timeout is negative"}} + {:ok, seconds} -> {:ok, trunc(seconds * 1000)} + :error -> {:reply, {:error, "ERR timeout is not a float or out of range"}} + end + end + + defp parseSide(raw) do + case String.upcase(raw) do + "LEFT" -> {:ok, :left} + "RIGHT" -> {:ok, :right} + _other -> :error + end + end + + defp parseMpop(numkeysRaw, rest) do + case Integer.parse(numkeysRaw) do + {numkeys, ""} when numkeys > 0 -> splitMpop(numkeys, rest) + _invalid -> {:reply, {:error, "ERR numkeys should be greater than 0"}} + end + end + + defp splitMpop(numkeys, rest) when length(rest) > numkeys do + {keys, [directionRaw | tail]} = Enum.split(rest, numkeys) + + case parseSide(directionRaw) do + {:ok, side} -> mpopCount(keys, side, tail) + :error -> {:reply, Reply.syntax()} + end + end + + defp splitMpop(_numkeys, _rest), do: {:reply, Reply.syntax()} + + defp mpopCount(keys, side, []), do: {:ok, keys, side, 1} + + defp mpopCount(keys, side, [countToken, countRaw]) do + if String.upcase(countToken) == "COUNT" do + case Integer.parse(countRaw) do + {count, ""} when count > 0 -> {:ok, keys, side, count} + _invalid -> {:reply, {:error, "ERR count should be greater than 0"}} + end + else + {:reply, Reply.syntax()} + end + end + + defp mpopCount(_keys, _side, _tail), do: {:reply, Reply.syntax()} +end diff --git a/lib/excache/command/extended.ex b/lib/excache/command/extended.ex index 1b9cd65..fcaff82 100644 --- a/lib/excache/command/extended.ex +++ b/lib/excache/command/extended.ex @@ -1,7 +1,7 @@ defmodule Excache.Command.Extended do @moduledoc false - alias Excache.Command.{Streams, Bitmaps, HyperLogLog, Keyspace} + alias Excache.Command.{Streams, Bitmaps, HyperLogLog, Keyspace, Server, Blocking, Introspection} @stream ~w(XADD XLEN XRANGE XREVRANGE XDEL XTRIM XSETID XREAD XINFO XGROUP XREADGROUP XACK XPENDING XCLAIM XAUTOCLAIM) @@ -17,17 +17,27 @@ defmodule Excache.Command.Extended do ZCOUNT ZREVRANGE ZREVRANK ZMSCORE RENAME RENAMENX COPY TOUCH UNLINK EXPIREAT EXPIRETIME PEXPIRETIME) + @server ~w(MEMORY CONFIG) + + @blocking Excache.Command.Blocking.commandNames() + + @introspection Excache.Command.Introspection.commandNames() + def dispatch(name, args, state) do cond do name in @stream -> Streams.dispatch(name, args, state) name in @bitmap -> Bitmaps.dispatch(name, args, state) name in @hll -> HyperLogLog.dispatch(name, args, state) name in @keyspace -> Keyspace.dispatch(name, args, state) + name in @server -> Server.dispatch(name, args, state) + name in @blocking -> Blocking.dispatch(name, args, state) + name in @introspection -> Introspection.dispatch(name, args, state) true -> :unhandled end end - def command_names, do: @stream ++ @bitmap ++ @hll ++ @keyspace + def command_names, + do: @stream ++ @bitmap ++ @hll ++ @keyspace ++ @server ++ @blocking ++ @introspection def write_command_names do ~w(XADD XDEL XTRIM XSETID XGROUP XREADGROUP XACK XCLAIM XAUTOCLAIM