feat(command): add stream command handlers
This commit is contained in:
dualfroz committed
1 parent
a1279d682f
commit
18f8f68088
1 file changed
+375
@@ -0,0 +1,375 @@
|
||||
defmodule Excache.Command.Streams do
|
||||
@moduledoc false
|
||||
|
||||
alias Excache.{Store, Stream}
|
||||
alias Excache.Command.Reply
|
||||
|
||||
def dispatch("XADD", args, state), do: xadd(args, state)
|
||||
def dispatch("XLEN", [key], state), do: reply({:integer, orZero(Store.xlen(key))}, state)
|
||||
def dispatch("XRANGE", args, state), do: xrange(args, state, :forward)
|
||||
def dispatch("XREVRANGE", args, state), do: xrange(args, state, :reverse)
|
||||
def dispatch("XDEL", [key | [_ | _] = ids], state), do: xdel(key, ids, state)
|
||||
def dispatch("XTRIM", args, state), do: xtrim(args, state)
|
||||
def dispatch("XSETID", args, state), do: xsetid(args, state)
|
||||
def dispatch("XREAD", args, state), do: xread(args, state)
|
||||
def dispatch("XINFO", args, state), do: xinfo(args, state)
|
||||
def dispatch("XGROUP", args, state), do: xgroup(args, state)
|
||||
def dispatch("XREADGROUP", args, state), do: xreadgroup(args, state)
|
||||
def dispatch("XACK", [key, group | [_ | _] = ids], state), do: xack(key, group, ids, state)
|
||||
def dispatch("XPENDING", args, state), do: xpending(args, state)
|
||||
def dispatch("XCLAIM", args, state), do: xclaim(args, state)
|
||||
def dispatch("XAUTOCLAIM", args, state), do: xautoclaim(args, state)
|
||||
def dispatch(_name, _args, _state), do: :unhandled
|
||||
|
||||
def rewriteAddId(args, id) do
|
||||
case parseAdd(args) do
|
||||
{:ok, key, _nomkstream, trim, _idToken, fields} ->
|
||||
[key] ++ trimTokens(trim) ++ [id] ++ fields
|
||||
|
||||
:error ->
|
||||
args
|
||||
end
|
||||
end
|
||||
|
||||
defp xadd(args, state) do
|
||||
case parseAdd(args) do
|
||||
{:ok, key, nomkstream, trim, idToken, fields} ->
|
||||
case Stream.parseAddId(idToken, now()) do
|
||||
{:ok, idSpec} ->
|
||||
case Store.xadd(key, idSpec, fields, trim, nomkstream) do
|
||||
{:ok, nil} -> reply({:bulk, nil}, state)
|
||||
{:ok, id} -> reply({:bulk, id}, state)
|
||||
{:error, reason} -> reply(streamError(reason), state)
|
||||
end
|
||||
|
||||
:error ->
|
||||
reply({:error, "ERR Invalid stream ID specified as stream command argument"}, state)
|
||||
end
|
||||
|
||||
:error ->
|
||||
reply(Reply.syntax(), state)
|
||||
end
|
||||
end
|
||||
|
||||
defp xrange([key, startRaw, endRaw], state, direction),
|
||||
do: xrangeWith(key, startRaw, endRaw, :all, state, direction)
|
||||
|
||||
defp xrange([key, startRaw, endRaw, count | rest], state, direction) do
|
||||
case {String.upcase(count), rest} do
|
||||
{"COUNT", [countRaw]} ->
|
||||
case Integer.parse(countRaw) do
|
||||
{value, ""} when value >= 0 ->
|
||||
xrangeWith(key, startRaw, endRaw, value, state, direction)
|
||||
|
||||
_ ->
|
||||
reply(Reply.not_integer(), state)
|
||||
end
|
||||
|
||||
_ ->
|
||||
reply(Reply.syntax(), state)
|
||||
end
|
||||
end
|
||||
|
||||
defp xrange(_args, state, _direction), do: reply(Reply.syntax(), state)
|
||||
|
||||
defp xrangeWith(key, startRaw, endRaw, count, state, direction) do
|
||||
with {:ok, startBound} <- Stream.parseRangeStart(startRaw),
|
||||
{:ok, endBound} <- Stream.parseRangeEnd(endRaw) do
|
||||
result =
|
||||
case direction do
|
||||
:forward -> Store.xrange(key, startBound, endBound, count)
|
||||
:reverse -> Store.xrevrange(key, endBound, startBound, count)
|
||||
end
|
||||
|
||||
case result do
|
||||
{:ok, entries} -> reply({:array, Enum.map(entries, &entryReply/1)}, state)
|
||||
{:error, reason} -> reply(streamError(reason), state)
|
||||
end
|
||||
else
|
||||
:error ->
|
||||
reply({:error, "ERR Invalid stream ID specified as stream command argument"}, state)
|
||||
end
|
||||
end
|
||||
|
||||
defp xdel(key, idsRaw, state) do
|
||||
case parseIds(idsRaw) do
|
||||
{:ok, ids} ->
|
||||
case Store.xdel(key, ids) do
|
||||
{:ok, removed} -> reply({:integer, removed}, state)
|
||||
{:error, reason} -> reply(streamError(reason), state)
|
||||
end
|
||||
|
||||
:error ->
|
||||
reply({:error, "ERR Invalid stream ID specified as stream command argument"}, state)
|
||||
end
|
||||
end
|
||||
|
||||
defp xtrim([key | rest], state) do
|
||||
case parseTrim(rest) do
|
||||
{:ok, {trim, _leftover}} when trim != nil ->
|
||||
case Store.xtrim(key, trim) do
|
||||
{:ok, removed} -> reply({:integer, removed}, state)
|
||||
{:error, reason} -> reply(streamError(reason), state)
|
||||
end
|
||||
|
||||
_ ->
|
||||
reply(Reply.syntax(), state)
|
||||
end
|
||||
end
|
||||
|
||||
defp xtrim(_args, state), do: reply(Reply.syntax(), state)
|
||||
|
||||
defp xsetid([key, idRaw | _opts], state) do
|
||||
case parseThresholdId(idRaw) do
|
||||
{:ok, id} ->
|
||||
case Store.xsetid(key, id) do
|
||||
{:ok, "OK"} -> reply({:simple, "OK"}, state)
|
||||
{:error, :nokey} -> reply({:error, "ERR no such key"}, state)
|
||||
{:error, :setid_smaller} -> reply({:error, setidSmallerMessage()}, state)
|
||||
{:error, reason} -> reply(streamError(reason), state)
|
||||
end
|
||||
|
||||
:error ->
|
||||
reply({:error, "ERR Invalid stream ID specified as stream command argument"}, state)
|
||||
end
|
||||
end
|
||||
|
||||
defp xsetid(_args, state), do: reply(Reply.syntax(), state)
|
||||
|
||||
defp xread(args, state) do
|
||||
case parseRead(args) do
|
||||
{:ok, count, pairs} ->
|
||||
collected =
|
||||
Enum.reduce(pairs, [], fn {key, idRaw}, acc ->
|
||||
afterId = resolveReadId(key, idRaw)
|
||||
|
||||
case Store.xread(key, afterId, count) do
|
||||
{:ok, nil} -> acc
|
||||
{:ok, entries} -> [streamEntriesReply(key, entries) | acc]
|
||||
{:error, _} -> acc
|
||||
end
|
||||
end)
|
||||
|> Enum.reverse()
|
||||
|
||||
case collected do
|
||||
[] -> reply({:array, nil}, state)
|
||||
results -> reply({:array, results}, state)
|
||||
end
|
||||
|
||||
:error ->
|
||||
reply(Reply.syntax(), state)
|
||||
end
|
||||
end
|
||||
|
||||
defp xreadgroup(args, state) do
|
||||
case parseReadGroup(args) do
|
||||
{:ok, group, consumer, count, noack, pairs} ->
|
||||
collected =
|
||||
Enum.reduce_while(pairs, {:ok, []}, fn {key, idRaw}, {:ok, acc} ->
|
||||
mode = if idRaw == ">", do: {:new}, else: {:history, parseThresholdBang(idRaw)}
|
||||
|
||||
case Store.xreadgroup(key, group, consumer, mode, count, noack) do
|
||||
{:ok, []} -> {:cont, {:ok, acc}}
|
||||
{:ok, entries} -> {:cont, {:ok, [streamEntriesReply(key, entries) | acc]}}
|
||||
{:error, :nogroup} -> {:halt, {:error, nogroup(key, group)}}
|
||||
{:error, reason} -> {:halt, {:error, streamError(reason)}}
|
||||
end
|
||||
end)
|
||||
|
||||
case collected do
|
||||
{:ok, []} -> reply({:array, nil}, state)
|
||||
{:ok, results} -> reply({:array, Enum.reverse(results)}, state)
|
||||
{:error, term} -> reply(term, state)
|
||||
end
|
||||
|
||||
:error ->
|
||||
reply(Reply.syntax(), state)
|
||||
end
|
||||
end
|
||||
|
||||
defp xinfo(["STREAM", key | _], state) do
|
||||
case Store.xinfoStream(key) do
|
||||
{:ok, info} -> reply({:array, infoReply(info)}, state)
|
||||
{:error, :nokey} -> reply({:error, "ERR no such key"}, state)
|
||||
{:error, reason} -> reply(streamError(reason), state)
|
||||
end
|
||||
end
|
||||
|
||||
defp xinfo(_args, state), do: reply(Reply.syntax(), state)
|
||||
|
||||
defp xgroup(["CREATE", key, group, startRaw | opts], state) do
|
||||
mkstream = Enum.any?(opts, &(String.upcase(&1) == "MKSTREAM"))
|
||||
|
||||
case parseGroupStart(startRaw) do
|
||||
{:ok, startSpec} ->
|
||||
case Store.xgroupCreate(key, group, startSpec, mkstream) do
|
||||
{:ok, "OK"} -> reply({:simple, "OK"}, state)
|
||||
{:error, :busygroup} -> reply({:error, "BUSYGROUP Consumer Group name already exists"}, state)
|
||||
{:error, :nokey} -> reply({:error, noKeyGroupMessage()}, state)
|
||||
{:error, reason} -> reply(streamError(reason), state)
|
||||
end
|
||||
|
||||
:error ->
|
||||
reply({:error, "ERR Invalid stream ID specified as stream command argument"}, state)
|
||||
end
|
||||
end
|
||||
|
||||
defp xgroup(["DESTROY", key, group], state) do
|
||||
case Store.xgroupDestroy(key, group) do
|
||||
{:ok, removed} -> reply({:integer, removed}, state)
|
||||
{:error, reason} -> reply(streamError(reason), state)
|
||||
end
|
||||
end
|
||||
|
||||
defp xgroup(["CREATECONSUMER", key, group, consumer], state) do
|
||||
case Store.xgroupCreateConsumer(key, group, consumer) do
|
||||
{:ok, added} -> reply({:integer, added}, state)
|
||||
{:error, :nogroup} -> reply(nogroup(key, group), state)
|
||||
{:error, reason} -> reply(streamError(reason), state)
|
||||
end
|
||||
end
|
||||
|
||||
defp xgroup(["DELCONSUMER", key, group, consumer], state) do
|
||||
case Store.xgroupDelConsumer(key, group, consumer) do
|
||||
{:ok, removed} -> reply({:integer, removed}, state)
|
||||
{:error, :nogroup} -> reply(nogroup(key, group), state)
|
||||
{:error, reason} -> reply(streamError(reason), state)
|
||||
end
|
||||
end
|
||||
|
||||
defp xgroup(["SETID", key, group, idRaw | _opts], state) do
|
||||
case parseGroupStart(idRaw) do
|
||||
{:ok, startSpec} ->
|
||||
id =
|
||||
case startSpec do
|
||||
:last -> lastIdOf(key)
|
||||
{:id, value} -> value
|
||||
end
|
||||
|
||||
case Store.xgroupSetId(key, group, id) do
|
||||
{:ok, "OK"} -> reply({:simple, "OK"}, state)
|
||||
{:error, :nogroup} -> reply(nogroup(key, group), state)
|
||||
{:error, reason} -> reply(streamError(reason), state)
|
||||
end
|
||||
|
||||
:error ->
|
||||
reply({:error, "ERR Invalid stream ID specified as stream command argument"}, state)
|
||||
end
|
||||
end
|
||||
|
||||
defp xgroup(_args, state), do: reply(Reply.syntax(), state)
|
||||
|
||||
defp xack(key, group, idsRaw, state) do
|
||||
case parseIds(idsRaw) do
|
||||
{:ok, ids} ->
|
||||
case Store.xack(key, group, ids) do
|
||||
{:ok, count} -> reply({:integer, count}, state)
|
||||
{:error, reason} -> reply(streamError(reason), state)
|
||||
end
|
||||
|
||||
:error ->
|
||||
reply({:error, "ERR Invalid stream ID specified as stream command argument"}, state)
|
||||
end
|
||||
end
|
||||
|
||||
defp xpending([key, group], state) do
|
||||
case Store.xpendingSummary(key, group) do
|
||||
{:ok, summary} -> reply(pendingSummaryReply(summary), state)
|
||||
{:error, :nogroup} -> reply(nogroup(key, group), state)
|
||||
{:error, reason} -> reply(streamError(reason), state)
|
||||
end
|
||||
end
|
||||
|
||||
defp xpending([key, group | rest], state) do
|
||||
case parsePendingRange(rest) do
|
||||
{:ok, startRaw, endRaw, count, consumer, minIdle} ->
|
||||
with {:ok, startId} <- pendingBound(startRaw, {0, 0}),
|
||||
{:ok, endId} <- pendingBound(endRaw, {maxMs(), maxSeq()}) do
|
||||
case Store.xpendingRange(key, group, startId, endId, count, consumer, minIdle) do
|
||||
{:ok, rows} -> reply({:array, Enum.map(rows, &pendingRowReply/1)}, state)
|
||||
{:error, :nogroup} -> reply(nogroup(key, group), state)
|
||||
{:error, reason} -> reply(streamError(reason), state)
|
||||
end
|
||||
else
|
||||
:error ->
|
||||
reply({:error, "ERR Invalid stream ID specified as stream command argument"}, state)
|
||||
end
|
||||
|
||||
:error ->
|
||||
reply(Reply.syntax(), state)
|
||||
end
|
||||
end
|
||||
|
||||
defp xpending(_args, state), do: reply(Reply.syntax(), state)
|
||||
|
||||
defp xclaim([key, group, consumer, minIdleRaw | rest], state) do
|
||||
with {minIdle, ""} <- Integer.parse(minIdleRaw),
|
||||
{:ok, ids, opts} <- parseClaim(rest) do
|
||||
case Store.xclaim(key, group, consumer, minIdle, ids, opts) do
|
||||
{:ok, results} -> reply(claimReply(results, opts.justId), state)
|
||||
{:error, :nogroup} -> reply(nogroup(key, group), state)
|
||||
{:error, reason} -> reply(streamError(reason), state)
|
||||
end
|
||||
else
|
||||
_ -> reply(Reply.syntax(), state)
|
||||
end
|
||||
end
|
||||
|
||||
defp xclaim(_args, state), do: reply(Reply.syntax(), state)
|
||||
|
||||
defp xautoclaim([key, group, consumer, minIdleRaw, startRaw | rest], state) do
|
||||
with {minIdle, ""} <- Integer.parse(minIdleRaw),
|
||||
{:ok, startId} <- Stream.parseRangeStart(startRaw),
|
||||
{:ok, count, justId} <- parseAutoclaim(rest) do
|
||||
{_kind, startTuple} = startId
|
||||
|
||||
case Store.xautoclaim(key, group, consumer, minIdle, startTuple, count) do
|
||||
{:ok, {cursor, claimed, deleted}} ->
|
||||
reply(autoclaimReply(cursor, claimed, deleted, justId), state)
|
||||
|
||||
{:error, :nogroup} ->
|
||||
reply(nogroup(key, group), state)
|
||||
|
||||
{:error, reason} ->
|
||||
reply(streamError(reason), state)
|
||||
end
|
||||
else
|
||||
_ -> reply(Reply.syntax(), state)
|
||||
end
|
||||
end
|
||||
|
||||
defp xautoclaim(_args, state), do: reply(Reply.syntax(), state)
|
||||
|
||||
defp parseAdd([key | rest]), do: parseAddOptions(rest, key, false, nil)
|
||||
defp parseAdd(_args), do: :error
|
||||
|
||||
defp parseAddOptions([token | rest], key, nomkstream, trim) do
|
||||
case String.upcase(token) do
|
||||
"NOMKSTREAM" ->
|
||||
parseAddOptions(rest, key, true, trim)
|
||||
|
||||
"MAXLEN" ->
|
||||
case parseTrimBody(rest, :maxlen) do
|
||||
{:ok, parsed, remaining} -> parseAddOptions(remaining, key, nomkstream, parsed)
|
||||
:error -> :error
|
||||
end
|
||||
|
||||
"MINID" ->
|
||||
case parseTrimBody(rest, :minid) do
|
||||
{:ok, parsed, remaining} -> parseAddOptions(remaining, key, nomkstream, parsed)
|
||||
:error -> :error
|
||||
end
|
||||
|
||||
_ ->
|
||||
case rest do
|
||||
[_ | _] = fields ->
|
||||
if rem(length(fields), 2) == 0, do: {:ok, key, nomkstream, trim, token, fields}, else: :error
|
||||
|
||||
[] ->
|
||||
:error
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
defp parseAddOptions([], _key, _nomkstream, _trim), do: :error
|
||||
|
||||
Reference in new issue
Block a user