diff --git a/lib/excache/command/streams.ex b/lib/excache/command/streams.ex index 66f2543..25ec915 100644 --- a/lib/excache/command/streams.ex +++ b/lib/excache/command/streams.ex @@ -373,3 +373,400 @@ defmodule Excache.Command.Streams do defp parseAddOptions([], _key, _nomkstream, _trim), do: :error + defp parseTrim([]), do: {:ok, {nil, []}} + + defp parseTrim([token | rest]) do + case String.upcase(token) do + "MAXLEN" -> + case parseTrimBody(rest, :maxlen) do + {:ok, parsed, remaining} -> {:ok, {parsed, remaining}} + :error -> :error + end + + "MINID" -> + case parseTrimBody(rest, :minid) do + {:ok, parsed, remaining} -> {:ok, {parsed, remaining}} + :error -> :error + end + + _ -> + :error + end + end + + defp parseTrimBody(tokens, kind) do + tokens = dropApproxMarker(tokens) + + case tokens do + [threshold | rest] -> + rest = dropLimit(rest) + + case buildTrim(kind, threshold) do + {:ok, parsed} -> {:ok, parsed, rest} + :error -> :error + end + + [] -> + :error + end + end + + defp dropApproxMarker([marker | rest]) when marker in ["~", "="], do: rest + defp dropApproxMarker(tokens), do: tokens + + defp dropLimit(["LIMIT", _value | rest]), do: rest + defp dropLimit(["limit", _value | rest]), do: rest + defp dropLimit(tokens), do: tokens + + defp buildTrim(:maxlen, threshold) do + case Integer.parse(threshold) do + {value, ""} when value >= 0 -> {:ok, {:maxlen, value}} + _ -> :error + end + end + + defp buildTrim(:minid, threshold) do + case parseThresholdId(threshold) do + {:ok, id} -> {:ok, {:minid, id}} + :error -> :error + end + end + + defp trimTokens(nil), do: [] + defp trimTokens({:maxlen, count}), do: ["MAXLEN", Integer.to_string(count)] + defp trimTokens({:minid, {ms, seq}}), do: ["MINID", "#{ms}-#{seq}"] + + defp parseRead(args) do + {count, rest} = takeReadCount(args, :all) + + case splitStreams(rest) do + {:ok, keys, ids} when length(keys) == length(ids) and keys != [] -> + {:ok, count, Enum.zip(keys, ids)} + + _ -> + :error + end + end + + defp parseReadGroup(["GROUP", group, consumer | rest]) do + {count, rest} = takeReadCount(rest, :all) + {noack, rest} = takeNoack(rest) + + case splitStreams(rest) do + {:ok, keys, ids} when length(keys) == length(ids) and keys != [] -> + {:ok, group, consumer, count, noack, Enum.zip(keys, ids)} + + _ -> + :error + end + end + + defp parseReadGroup(_args), do: :error + + defp takeReadCount(["COUNT", raw | rest], _default) do + case Integer.parse(raw) do + {value, ""} when value >= 0 -> {value, rest} + _ -> {:all, rest} + end + end + + defp takeReadCount([token, raw | rest], default) when token in ["count"] do + takeReadCount(["COUNT", raw | rest], default) + end + + defp takeReadCount(rest, default), do: {default, rest} + + defp takeNoack(["NOACK" | rest]), do: {true, rest} + defp takeNoack(["noack" | rest]), do: {true, rest} + defp takeNoack(rest), do: {false, rest} + + defp splitStreams(tokens) do + case Enum.split_while(tokens, &(String.upcase(&1) != "STREAMS")) do + {leading, ["STREAMS" | rest]} -> + # any leading BLOCK ms is tolerated and ignored here + _ = leading + half = div(length(rest), 2) + + if rem(length(rest), 2) == 0 and half > 0 do + {:ok, Enum.take(rest, half), Enum.drop(rest, half)} + else + :error + end + + _ -> + :error + end + end + + defp resolveReadId(key, "$"), do: lastIdOf(key) + defp resolveReadId(_key, "+"), do: {maxMs(), maxSeq()} + + defp resolveReadId(_key, raw) do + case parseThresholdId(raw) do + {:ok, id} -> id + :error -> {0, 0} + end + end + + defp lastIdOf(key) do + case Store.xinfoStream(key) do + {:ok, info} -> info.lastGeneratedId + _ -> {0, 0} + end + end + + defp parseGroupStart("$"), do: {:ok, :last} + defp parseGroupStart("0"), do: {:ok, {:id, {0, 0}}} + + defp parseGroupStart(raw) do + case parseThresholdId(raw) do + {:ok, id} -> {:ok, {:id, id}} + :error -> :error + end + end + + defp parsePendingRange(["IDLE", idleRaw | rest]) do + case Integer.parse(idleRaw) do + {idle, ""} -> withPendingCore(rest, idle) + _ -> :error + end + end + + defp parsePendingRange(rest), do: withPendingCore(rest, nil) + + defp withPendingCore([startRaw, endRaw, countRaw | rest], minIdle) do + case Integer.parse(countRaw) do + {count, ""} when count >= 0 -> + consumer = List.first(rest) + {:ok, startRaw, endRaw, count, consumer, minIdle} + + _ -> + :error + end + end + + defp withPendingCore(_rest, _minIdle), do: :error + + defp pendingBound("-", _default), do: {:ok, {0, 0}} + defp pendingBound("+", _default), do: {:ok, {maxMs(), maxSeq()}} + + defp pendingBound(raw, _default) do + case parseThresholdId(raw) do + {:ok, id} -> {:ok, id} + :error -> :error + end + end + + defp parseClaim(tokens), do: parseClaimIds(tokens, []) + + defp parseClaimIds([token | rest], acc) do + case parseThresholdId(token) do + {:ok, id} -> + parseClaimIds(rest, [id | acc]) + + :error -> + case acc do + [] -> :error + _ -> parseClaimOpts([token | rest], Enum.reverse(acc), defaultClaimOpts()) + end + end + end + + defp parseClaimIds([], acc) when acc != [], do: {:ok, Enum.reverse(acc), defaultClaimOpts()} + defp parseClaimIds([], _acc), do: :error + + defp parseClaimOpts([], ids, opts), do: {:ok, ids, opts} + + defp parseClaimOpts([token | rest], ids, opts) do + case String.upcase(token) do + "IDLE" -> claimIntOpt(rest, ids, opts, :idle) + "TIME" -> claimIntOpt(rest, ids, opts, :time) + "RETRYCOUNT" -> claimIntOpt(rest, ids, opts, :retryCount) + "FORCE" -> parseClaimOpts(rest, ids, %{opts | force: true}) + "JUSTID" -> parseClaimOpts(rest, ids, %{opts | justId: true}) + "LASTID" -> parseClaimOpts(tl(rest), ids, opts) + _ -> :error + end + end + + defp claimIntOpt([raw | rest], ids, opts, field) do + case Integer.parse(raw) do + {value, ""} -> parseClaimOpts(rest, ids, Map.put(opts, field, value)) + _ -> :error + end + end + + defp claimIntOpt([], _ids, _opts, _field), do: :error + + defp defaultClaimOpts, + do: %{idle: nil, time: nil, retryCount: nil, force: false, justId: false} + + defp parseAutoclaim(["COUNT", raw | rest]) do + case Integer.parse(raw) do + {value, ""} when value > 0 -> finishAutoclaim(rest, value) + _ -> :error + end + end + + defp parseAutoclaim(rest), do: finishAutoclaim(rest, 100) + + defp finishAutoclaim([], count), do: {:ok, count, false} + defp finishAutoclaim(["JUSTID"], count), do: {:ok, count, true} + defp finishAutoclaim(["justid"], count), do: {:ok, count, true} + defp finishAutoclaim(_rest, _count), do: :error + + defp parseIds(idsRaw) do + Enum.reduce_while(idsRaw, {:ok, []}, fn raw, {:ok, acc} -> + case parseThresholdId(raw) do + {:ok, id} -> {:cont, {:ok, [id | acc]}} + :error -> {:halt, :error} + end + end) + |> case do + {:ok, ids} -> {:ok, Enum.reverse(ids)} + :error -> :error + end + end + + defp parseThresholdId(raw) do + case String.split(raw, "-", parts: 2) do + [ms] -> withMs(ms, 0) + [ms, seq] -> withMsSeq(ms, seq) + end + end + + defp parseThresholdBang(raw) do + case parseThresholdId(raw) do + {:ok, id} -> id + :error -> {0, 0} + end + end + + defp withMs(ms, seq) do + case Integer.parse(ms) do + {value, ""} when value >= 0 -> {:ok, {value, seq}} + _ -> :error + end + end + + defp withMsSeq(ms, seq) do + with {msValue, ""} <- Integer.parse(ms), + {seqValue, ""} <- Integer.parse(seq), + true <- msValue >= 0 and seqValue >= 0 do + {:ok, {msValue, seqValue}} + else + _ -> :error + end + end + + defp entryReply({id, fields}) do + {:array, [{:bulk, Stream.formatId(id)}, fieldsReply(fields)]} + end + + defp fieldsReply(nil), do: {:array, nil} + defp fieldsReply(fields), do: {:array, Enum.map(fields, &{:bulk, &1})} + + defp streamEntriesReply(key, entries) do + {:array, [{:bulk, key}, {:array, Enum.map(entries, &entryReply/1)}]} + end + + defp infoReply(info) do + [ + {:bulk, "length"}, + {:integer, info.length}, + {:bulk, "last-generated-id"}, + {:bulk, Stream.formatId(info.lastGeneratedId)}, + {:bulk, "max-deleted-entry-id"}, + {:bulk, Stream.formatId(info.maxDeletedId)}, + {:bulk, "entries-added"}, + {:integer, info.entriesAdded}, + {:bulk, "groups"}, + {:integer, info.groups}, + {:bulk, "first-entry"}, + firstOrNil(info.firstEntry), + {:bulk, "last-entry"}, + firstOrNil(info.lastEntry) + ] + end + + defp firstOrNil(nil), do: {:array, nil} + defp firstOrNil(entry), do: entryReply(entry) + + defp pendingSummaryReply({0, _minId, _maxId, _perConsumer}) do + {:array, [{:integer, 0}, {:bulk, nil}, {:bulk, nil}, {:array, nil}]} + end + + defp pendingSummaryReply({total, minId, maxId, perConsumer}) do + consumers = + Enum.map(perConsumer, fn {consumer, count} -> + {:array, [{:bulk, consumer}, {:bulk, Integer.to_string(count)}]} + end) + + {:array, + [ + {:integer, total}, + {:bulk, Stream.formatId(minId)}, + {:bulk, Stream.formatId(maxId)}, + {:array, consumers} + ]} + end + + defp pendingRowReply({id, consumer, idle, deliveryCount}) do + {:array, + [ + {:bulk, Stream.formatId(id)}, + {:bulk, consumer}, + {:integer, idle}, + {:integer, deliveryCount} + ]} + end + + defp claimReply(results, true) do + {:array, Enum.map(results, fn {id, _fields} -> {:bulk, Stream.formatId(id)} end)} + end + + defp claimReply(results, false) do + {:array, Enum.map(results, &entryReply/1)} + end + + defp autoclaimReply(cursor, claimed, deleted, justId) do + entriesReply = + if justId do + {:array, Enum.map(claimed, fn {id, _fields} -> {:bulk, Stream.formatId(id)} end)} + else + {:array, Enum.map(claimed, &entryReply/1)} + end + + deletedReply = {:array, Enum.map(deleted, fn id -> {:bulk, Stream.formatId(id)} end)} + {:array, [{:bulk, Stream.formatId(cursor)}, entriesReply, deletedReply]} + end + + defp reply(term, state), do: {[term], state} + + defp orZero({:ok, value}), do: value + defp orZero({:error, _}), do: 0 + + defp streamError(:wrongtype), do: Reply.wrongtype() + defp streamError(:smaller), + do: {:error, "ERR The ID specified in XADD is equal or smaller than the target stream top item"} + + defp streamError(:zero), do: {:error, "ERR The ID specified in XADD must be greater than 0-0"} + defp streamError(other), do: {:error, "ERR #{other}"} + + defp nogroup(key, group) do + {:error, + "NOGROUP No such key '#{key}' or consumer group '#{group}' in XREADGROUP with GROUP option"} + end + + defp noKeyGroupMessage do + "ERR The XGROUP subcommand requires the key to exist. Note that for CREATE you may want to use the MKSTREAM option to create an empty stream automatically." + end + + defp setidSmallerMessage do + "ERR The ID specified in XSETID is smaller than the target stream top item" + end + + defp now, do: System.system_time(:millisecond) + defp maxMs, do: 0xFFFFFFFFFFFFFFFF + defp maxSeq, do: 0xFFFFFFFFFFFFFFFF +end