feat(command): add core command dispatch
This commit is contained in:
dualfroz committed
1 parent
89d109fab0
commit
4438b77b2f
1 file changed
+392
@@ -0,0 +1,392 @@
|
||||
defmodule Excache.Command do
|
||||
@moduledoc false
|
||||
|
||||
alias Excache.{PubSub, Store, Persistence, Aof, Notify}
|
||||
alias Excache.Command.{Reply, Transaction}
|
||||
|
||||
import Reply,
|
||||
only: [
|
||||
reply: 2,
|
||||
with_int: 3,
|
||||
with_two_ints: 4,
|
||||
map_get: 1,
|
||||
map_int: 1,
|
||||
map_score: 1,
|
||||
map_list: 1,
|
||||
map_pop: 2,
|
||||
zrange_reply: 2,
|
||||
bulks: 1,
|
||||
bulks_nullable: 1,
|
||||
pairs: 1,
|
||||
parse_set_opts: 2,
|
||||
parse_zadd_args: 1,
|
||||
parse_bound: 1
|
||||
]
|
||||
|
||||
@known ~w(
|
||||
PING ECHO COMMAND DBSIZE FLUSHDB SELECT QUIT
|
||||
GET SET DEL EXISTS KEYS TYPE INCR DECR INCRBY DECRBY APPEND GETSET MSET MGET
|
||||
EXPIRE PEXPIRE TTL PTTL PERSIST
|
||||
LPUSH RPUSH LPOP RPOP LRANGE LLEN LINDEX
|
||||
HSET HGET HDEL HGETALL HKEYS HVALS HLEN HEXISTS HINCRBY
|
||||
SADD SREM SMEMBERS SISMEMBER SCARD SINTER SUNION SDIFF
|
||||
SINTERSTORE SUNIONSTORE SDIFFSTORE
|
||||
ZADD ZSCORE ZRANK ZINCRBY ZREM ZCARD ZRANGE ZRANGEBYSCORE ZPOPMIN ZPOPMAX
|
||||
MULTI EXEC DISCARD WATCH UNWATCH
|
||||
SUBSCRIBE UNSUBSCRIBE PUBLISH
|
||||
SAVE BGSAVE
|
||||
)
|
||||
|
||||
@write_commands ~w(
|
||||
SET DEL EXPIRE PEXPIRE PERSIST INCR DECR INCRBY DECRBY APPEND GETSET MSET FLUSHDB
|
||||
LPUSH RPUSH LPOP RPOP
|
||||
HSET HDEL HINCRBY
|
||||
SADD SREM SINTERSTORE SUNIONSTORE SDIFFSTORE
|
||||
ZADD ZREM ZINCRBY ZPOPMIN ZPOPMAX
|
||||
)
|
||||
|
||||
def known_commands, do: @known ++ Excache.Command.Extended.command_names()
|
||||
|
||||
@extended_write Excache.Command.Extended.write_command_names()
|
||||
|
||||
def execute([], state), do: {[], state}
|
||||
|
||||
def execute([name | args], state) do
|
||||
Transaction.execute(String.upcase(name), args, state, &dispatch_and_log/3)
|
||||
end
|
||||
|
||||
def replay_dispatch(command, args) do
|
||||
dispatch(command, args, replay_state())
|
||||
:ok
|
||||
end
|
||||
|
||||
def dispatch_and_log(command, args, state) do
|
||||
{replies, updated} = dispatch(command, args, state)
|
||||
|
||||
if writeCommand?(command) and not Enum.any?(replies, &match?({:error, _}, &1)) do
|
||||
Aof.append(command, aofArgs(command, args, replies))
|
||||
Notify.for_command(command, args)
|
||||
end
|
||||
|
||||
{replies, updated}
|
||||
end
|
||||
|
||||
defp writeCommand?(command), do: command in @write_commands or command in @extended_write
|
||||
|
||||
defp aofArgs("XADD", args, [{:bulk, id}]) when is_binary(id),
|
||||
do: Excache.Command.Streams.rewriteAddId(args, id)
|
||||
|
||||
defp aofArgs(_command, args, _replies), do: args
|
||||
|
||||
defp dispatch("PING", [], state), do: {[{:simple, "PONG"}], state}
|
||||
defp dispatch("PING", [message], state), do: {[{:bulk, message}], state}
|
||||
defp dispatch("ECHO", [message], state), do: {[{:bulk, message}], state}
|
||||
defp dispatch("COMMAND", _args, state), do: {[{:array, []}], state}
|
||||
defp dispatch("SELECT", [_db], state), do: {[{:simple, "OK"}], state}
|
||||
defp dispatch("QUIT", [], state), do: {[{:simple, "OK"}], state}
|
||||
defp dispatch("DBSIZE", [], state), do: {[{:integer, Store.dbsize()}], state}
|
||||
|
||||
defp dispatch("FLUSHDB", _args, state) do
|
||||
Store.flushdb()
|
||||
{[{:simple, "OK"}], state}
|
||||
end
|
||||
|
||||
defp dispatch("GET", [key], state), do: reply(map_get(Store.get(key)), state)
|
||||
|
||||
defp dispatch("SET", [key, value | rest], state) do
|
||||
case parse_set_opts(rest, []) do
|
||||
{:ok, opts} ->
|
||||
reply(
|
||||
case Store.set(key, value, opts) do
|
||||
:ok -> {:simple, "OK"}
|
||||
nil -> {:bulk, nil}
|
||||
end,
|
||||
state
|
||||
)
|
||||
|
||||
:error ->
|
||||
{[Reply.syntax()], state}
|
||||
end
|
||||
end
|
||||
|
||||
defp dispatch("DEL", [_ | _] = keys, state), do: {[{:integer, Store.del(keys)}], state}
|
||||
defp dispatch("EXISTS", [_ | _] = keys, state), do: {[{:integer, Store.exists(keys)}], state}
|
||||
defp dispatch("KEYS", [pattern], state), do: {[{:array, bulks(Store.keys(pattern))}], state}
|
||||
defp dispatch("TYPE", [key], state), do: {[{:simple, Store.type(key)}], state}
|
||||
|
||||
defp dispatch("INCR", [key], state), do: reply(map_int(Store.incrby(key, 1)), state)
|
||||
defp dispatch("DECR", [key], state), do: reply(map_int(Store.incrby(key, -1)), state)
|
||||
|
||||
defp dispatch("INCRBY", [key, delta], state) do
|
||||
with_int(delta, state, fn amount -> map_int(Store.incrby(key, amount)) end)
|
||||
end
|
||||
|
||||
defp dispatch("DECRBY", [key, delta], state) do
|
||||
with_int(delta, state, fn amount -> map_int(Store.incrby(key, -amount)) end)
|
||||
end
|
||||
|
||||
defp dispatch("APPEND", [key, value], state),
|
||||
do: reply(map_int(Store.append(key, value)), state)
|
||||
|
||||
defp dispatch("GETSET", [key, value], state),
|
||||
do: reply(map_get(Store.getset(key, value)), state)
|
||||
|
||||
defp dispatch("MSET", args, state) when args != [] do
|
||||
if rem(length(args), 2) == 0 do
|
||||
Store.mset(pairs(args))
|
||||
{[{:simple, "OK"}], state}
|
||||
else
|
||||
{[{:error, "ERR wrong number of arguments for 'mset' command"}], state}
|
||||
end
|
||||
end
|
||||
|
||||
defp dispatch("MGET", [_ | _] = keys, state) do
|
||||
{[{:array, bulks_nullable(Store.mget(keys))}], state}
|
||||
end
|
||||
|
||||
defp dispatch("EXPIRE", [key, seconds], state) do
|
||||
with_int(seconds, state, fn secs -> {:integer, Store.expire(key, secs * 1000)} end)
|
||||
end
|
||||
|
||||
defp dispatch("PEXPIRE", [key, ms], state) do
|
||||
with_int(ms, state, fn millis -> {:integer, Store.expire(key, millis)} end)
|
||||
end
|
||||
|
||||
defp dispatch("TTL", [key], state), do: {[{:integer, Store.ttl(key, :seconds)}], state}
|
||||
defp dispatch("PTTL", [key], state), do: {[{:integer, Store.ttl(key, :milliseconds)}], state}
|
||||
defp dispatch("PERSIST", [key], state), do: {[{:integer, Store.persist(key)}], state}
|
||||
|
||||
defp dispatch("LPUSH", [key | [_ | _] = values], state),
|
||||
do: reply(map_int(Store.lpush(key, values)), state)
|
||||
|
||||
defp dispatch("RPUSH", [key | [_ | _] = values], state),
|
||||
do: reply(map_int(Store.rpush(key, values)), state)
|
||||
|
||||
defp dispatch("LPOP", [key], state), do: reply(map_pop(Store.lpop(key, nil), nil), state)
|
||||
|
||||
defp dispatch("LPOP", [key, count], state) do
|
||||
with_int(count, state, fn n -> map_pop(Store.lpop(key, n), n) end)
|
||||
end
|
||||
|
||||
defp dispatch("RPOP", [key], state), do: reply(map_pop(Store.rpop(key, nil), nil), state)
|
||||
|
||||
defp dispatch("RPOP", [key, count], state) do
|
||||
with_int(count, state, fn n -> map_pop(Store.rpop(key, n), n) end)
|
||||
end
|
||||
|
||||
defp dispatch("LRANGE", [key, start, stop], state) do
|
||||
with_two_ints(start, stop, state, fn lo, hi -> map_list(Store.lrange(key, lo, hi)) end)
|
||||
end
|
||||
|
||||
defp dispatch("LLEN", [key], state), do: reply(map_int(Store.llen(key)), state)
|
||||
|
||||
defp dispatch("LINDEX", [key, index], state) do
|
||||
with_int(index, state, fn i -> map_get(Store.lindex(key, i)) end)
|
||||
end
|
||||
|
||||
defp dispatch("HSET", [key | [_, _ | _] = rest], state) do
|
||||
if rem(length(rest), 2) == 0 do
|
||||
reply(map_int(Store.hset(key, pairs(rest))), state)
|
||||
else
|
||||
{[{:error, "ERR wrong number of arguments for 'hset' command"}], state}
|
||||
end
|
||||
end
|
||||
|
||||
defp dispatch("HGET", [key, field], state), do: reply(map_get(Store.hget(key, field)), state)
|
||||
|
||||
defp dispatch("HDEL", [key | [_ | _] = fields], state),
|
||||
do: reply(map_int(Store.hdel(key, fields)), state)
|
||||
|
||||
defp dispatch("HGETALL", [key], state), do: reply(map_list(Store.hgetall(key)), state)
|
||||
defp dispatch("HKEYS", [key], state), do: reply(map_list(Store.hkeys(key)), state)
|
||||
defp dispatch("HVALS", [key], state), do: reply(map_list(Store.hvals(key)), state)
|
||||
defp dispatch("HLEN", [key], state), do: reply(map_int(Store.hlen(key)), state)
|
||||
|
||||
defp dispatch("HEXISTS", [key, field], state),
|
||||
do: reply(map_int(Store.hexists(key, field)), state)
|
||||
|
||||
defp dispatch("HINCRBY", [key, field, delta], state) do
|
||||
with_int(delta, state, fn amount -> map_int(Store.hincrby(key, field, amount)) end)
|
||||
end
|
||||
|
||||
defp dispatch("SADD", [key | [_ | _] = members], state),
|
||||
do: reply(map_int(Store.sadd(key, members)), state)
|
||||
|
||||
defp dispatch("SREM", [key | [_ | _] = members], state),
|
||||
do: reply(map_int(Store.srem(key, members)), state)
|
||||
|
||||
defp dispatch("SMEMBERS", [key], state), do: reply(map_list(Store.smembers(key)), state)
|
||||
|
||||
defp dispatch("SISMEMBER", [key, member], state),
|
||||
do: reply(map_int(Store.sismember(key, member)), state)
|
||||
|
||||
defp dispatch("SCARD", [key], state), do: reply(map_int(Store.scard(key)), state)
|
||||
|
||||
defp dispatch("SINTER", [_ | _] = keys, state), do: reply(map_list(Store.sinter(keys)), state)
|
||||
defp dispatch("SUNION", [_ | _] = keys, state), do: reply(map_list(Store.sunion(keys)), state)
|
||||
defp dispatch("SDIFF", [_ | _] = keys, state), do: reply(map_list(Store.sdiff(keys)), state)
|
||||
|
||||
defp dispatch("SINTERSTORE", [dest | [_ | _] = keys], state),
|
||||
do: reply(map_int(Store.sinterstore(dest, keys)), state)
|
||||
|
||||
defp dispatch("SUNIONSTORE", [dest | [_ | _] = keys], state),
|
||||
do: reply(map_int(Store.sunionstore(dest, keys)), state)
|
||||
|
||||
defp dispatch("SDIFFSTORE", [dest | [_ | _] = keys], state),
|
||||
do: reply(map_int(Store.sdiffstore(dest, keys)), state)
|
||||
|
||||
defp dispatch("ZADD", [key | [_ | _] = rest], state) do
|
||||
case parse_zadd_args(rest) do
|
||||
{:ok, opts, entries} -> reply(map_int(Store.zadd(key, entries, opts)), state)
|
||||
:error -> {[Reply.not_float()], state}
|
||||
end
|
||||
end
|
||||
|
||||
defp dispatch("ZSCORE", [key, member], state),
|
||||
do: reply(map_score(Store.zscore(key, member)), state)
|
||||
|
||||
defp dispatch("ZRANK", [key, member], state),
|
||||
do: reply(map_int(Store.zrank(key, member)), state)
|
||||
|
||||
defp dispatch("ZINCRBY", [key, delta, member], state) do
|
||||
case Reply.parse_score(delta) do
|
||||
{:ok, amount} -> reply(map_score(Store.zincrby(key, amount, member)), state)
|
||||
:error -> {[Reply.not_float()], state}
|
||||
end
|
||||
end
|
||||
|
||||
defp dispatch("ZREM", [key | [_ | _] = members], state),
|
||||
do: reply(map_int(Store.zrem(key, members)), state)
|
||||
|
||||
defp dispatch("ZCARD", [key], state), do: reply(map_int(Store.zcard(key)), state)
|
||||
|
||||
defp dispatch("ZRANGE", [key, start, stop], state) do
|
||||
with_two_ints(start, stop, state, fn lo, hi ->
|
||||
zrange_reply(Store.zrange(key, lo, hi), false)
|
||||
end)
|
||||
end
|
||||
|
||||
defp dispatch("ZRANGE", [key, start, stop, flag], state) do
|
||||
if String.upcase(flag) == "WITHSCORES" do
|
||||
with_two_ints(start, stop, state, fn lo, hi ->
|
||||
zrange_reply(Store.zrange(key, lo, hi), true)
|
||||
end)
|
||||
else
|
||||
{[Reply.syntax()], state}
|
||||
end
|
||||
end
|
||||
|
||||
defp dispatch("ZRANGEBYSCORE", [key, min, max], state),
|
||||
do: zrangebyscore(key, min, max, false, state)
|
||||
|
||||
defp dispatch("ZRANGEBYSCORE", [key, min, max, flag], state) do
|
||||
if String.upcase(flag) == "WITHSCORES" do
|
||||
zrangebyscore(key, min, max, true, state)
|
||||
else
|
||||
{[Reply.syntax()], state}
|
||||
end
|
||||
end
|
||||
|
||||
defp dispatch("ZPOPMIN", [key], state), do: reply(zrange_reply(Store.zpopmin(key, 1), true), state)
|
||||
|
||||
defp dispatch("ZPOPMIN", [key, count], state) do
|
||||
with_int(count, state, fn n -> zrange_reply(Store.zpopmin(key, n), true) end)
|
||||
end
|
||||
|
||||
defp dispatch("ZPOPMAX", [key], state), do: reply(zrange_reply(Store.zpopmax(key, 1), true), state)
|
||||
|
||||
defp dispatch("ZPOPMAX", [key, count], state) do
|
||||
with_int(count, state, fn n -> zrange_reply(Store.zpopmax(key, n), true) end)
|
||||
end
|
||||
|
||||
defp dispatch("SUBSCRIBE", [_ | _] = channels, state) do
|
||||
Enum.reduce(channels, {[], state}, fn channel, {replies, current} ->
|
||||
updated =
|
||||
if MapSet.member?(current.subscriptions, channel) do
|
||||
current
|
||||
else
|
||||
PubSub.subscribe(channel)
|
||||
%{current | subscriptions: MapSet.put(current.subscriptions, channel)}
|
||||
end
|
||||
|
||||
confirm =
|
||||
{:array,
|
||||
[{:bulk, "subscribe"}, {:bulk, channel}, {:integer, MapSet.size(updated.subscriptions)}]}
|
||||
|
||||
{replies ++ [confirm], updated}
|
||||
end)
|
||||
end
|
||||
|
||||
defp dispatch("UNSUBSCRIBE", [], state) do
|
||||
case MapSet.to_list(state.subscriptions) do
|
||||
[] ->
|
||||
{[{:array, [{:bulk, "unsubscribe"}, {:bulk, nil}, {:integer, 0}]}], state}
|
||||
|
||||
channels ->
|
||||
unsubscribe_each(channels, state)
|
||||
end
|
||||
end
|
||||
|
||||
defp dispatch("UNSUBSCRIBE", channels, state), do: unsubscribe_each(channels, state)
|
||||
|
||||
defp dispatch("PUBLISH", [channel, payload], state) do
|
||||
{[{:integer, PubSub.publish(channel, payload)}], state}
|
||||
end
|
||||
|
||||
defp dispatch("SAVE", [], state) do
|
||||
Persistence.save()
|
||||
{[{:simple, "OK"}], state}
|
||||
end
|
||||
|
||||
defp dispatch("BGSAVE", [], state) do
|
||||
Task.start(fn -> Persistence.save() end)
|
||||
{[{:simple, "Background saving started"}], state}
|
||||
end
|
||||
|
||||
defp dispatch(name, args, state) do
|
||||
case Excache.Command.Extended.dispatch(name, args, state) do
|
||||
:unhandled -> unknownOrArity(name, state)
|
||||
result -> result
|
||||
end
|
||||
end
|
||||
|
||||
defp unknownOrArity(name, state) do
|
||||
if name in known_commands() do
|
||||
{[{:error, "ERR wrong number of arguments for '#{String.downcase(name)}' command"}], state}
|
||||
else
|
||||
{[{:error, "ERR unknown command '#{name}'"}], state}
|
||||
end
|
||||
end
|
||||
|
||||
defp unsubscribe_each(channels, state) do
|
||||
Enum.reduce(channels, {[], state}, fn channel, {replies, current} ->
|
||||
updated =
|
||||
if MapSet.member?(current.subscriptions, channel) do
|
||||
PubSub.unsubscribe(channel)
|
||||
%{current | subscriptions: MapSet.delete(current.subscriptions, channel)}
|
||||
else
|
||||
current
|
||||
end
|
||||
|
||||
confirm =
|
||||
{:array,
|
||||
[
|
||||
{:bulk, "unsubscribe"},
|
||||
{:bulk, channel},
|
||||
{:integer, MapSet.size(updated.subscriptions)}
|
||||
]}
|
||||
|
||||
{replies ++ [confirm], updated}
|
||||
end)
|
||||
end
|
||||
|
||||
defp zrangebyscore(key, min, max, withscores, state) do
|
||||
with {:ok, lo} <- parse_bound(min), {:ok, hi} <- parse_bound(max) do
|
||||
reply(zrange_reply(Store.zrangebyscore(key, lo, hi), withscores), state)
|
||||
else
|
||||
:error -> {[Reply.not_float()], state}
|
||||
end
|
||||
end
|
||||
|
||||
defp replay_state, do: Map.put(Transaction.initial_state(), :subscriptions, MapSet.new())
|
||||
end
|
||||
Reference in new issue
Block a user