feat(store): add store client api
This commit is contained in:
dualfroz committed
1 parent
9707dd9498
commit
c21152f0b5
1 file changed
+445
@@ -0,0 +1,445 @@
|
||||
defmodule Excache.Store do
|
||||
@moduledoc false
|
||||
|
||||
alias Excache.{Bitmap, HyperLogLog}
|
||||
|
||||
def shard_count, do: Application.get_env(:excache, :shard_count, 8)
|
||||
|
||||
def shard_name(index), do: :"Excache.Store.Shard.#{index}"
|
||||
|
||||
defp shard_for(key), do: shard_name(rem(:erlang.phash2(key), shard_count()))
|
||||
|
||||
defp call_key(key, op), do: GenServer.call(shard_for(key), {:op, op})
|
||||
|
||||
defp fan_out(op) do
|
||||
for index <- 0..(shard_count() - 1) do
|
||||
GenServer.call(shard_name(index), {:op, op})
|
||||
end
|
||||
end
|
||||
|
||||
def get(key), do: call_key(key, {:get, key})
|
||||
def set(key, value, opts), do: call_key(key, {:set, key, value, opts})
|
||||
def type(key), do: call_key(key, {:type, key})
|
||||
def incrby(key, delta), do: call_key(key, {:incrby, key, delta})
|
||||
def append(key, value), do: call_key(key, {:append, key, value})
|
||||
def getset(key, value), do: call_key(key, {:getset, key, value})
|
||||
|
||||
def del(keys), do: Enum.reduce(keys, 0, fn key, acc -> acc + call_key(key, {:del, key}) end)
|
||||
|
||||
def exists(keys),
|
||||
do: Enum.reduce(keys, 0, fn key, acc -> acc + call_key(key, {:exists, key}) end)
|
||||
|
||||
def mset(pairs) do
|
||||
Enum.each(pairs, fn {key, value} -> call_key(key, {:set, key, value, []}) end)
|
||||
:ok
|
||||
end
|
||||
|
||||
def mget(keys) do
|
||||
Enum.map(keys, fn key ->
|
||||
case call_key(key, {:get, key}) do
|
||||
{:ok, value} -> value
|
||||
_ -> nil
|
||||
end
|
||||
end)
|
||||
end
|
||||
|
||||
def expire(key, ms), do: call_key(key, {:expire, key, ms})
|
||||
def ttl(key, unit), do: call_key(key, {:ttl, key, unit})
|
||||
def persist(key), do: call_key(key, {:persist, key})
|
||||
|
||||
def lpush(key, values), do: call_key(key, {:lpush, key, values})
|
||||
def rpush(key, values), do: call_key(key, {:rpush, key, values})
|
||||
def lpop(key, count), do: call_key(key, {:lpop, key, count})
|
||||
def rpop(key, count), do: call_key(key, {:rpop, key, count})
|
||||
def lrange(key, start, stop), do: call_key(key, {:lrange, key, start, stop})
|
||||
def llen(key), do: call_key(key, {:llen, key})
|
||||
def lindex(key, index), do: call_key(key, {:lindex, key, index})
|
||||
|
||||
def hset(key, pairs), do: call_key(key, {:hset, key, pairs})
|
||||
def hget(key, field), do: call_key(key, {:hget, key, field})
|
||||
def hdel(key, fields), do: call_key(key, {:hdel, key, fields})
|
||||
def hgetall(key), do: call_key(key, {:hgetall, key})
|
||||
def hkeys(key), do: call_key(key, {:hkeys, key})
|
||||
def hvals(key), do: call_key(key, {:hvals, key})
|
||||
def hlen(key), do: call_key(key, {:hlen, key})
|
||||
def hexists(key, field), do: call_key(key, {:hexists, key, field})
|
||||
def hincrby(key, field, delta), do: call_key(key, {:hincrby, key, field, delta})
|
||||
|
||||
def sadd(key, members), do: call_key(key, {:sadd, key, members})
|
||||
def srem(key, members), do: call_key(key, {:srem, key, members})
|
||||
def smembers(key), do: call_key(key, {:smembers, key})
|
||||
def sismember(key, member), do: call_key(key, {:sismember, key, member})
|
||||
def scard(key), do: call_key(key, {:scard, key})
|
||||
def sinter(keys), do: set_op(keys, &MapSet.intersection/2)
|
||||
def sunion(keys), do: set_op(keys, &MapSet.union/2)
|
||||
def sdiff(keys), do: set_op(keys, &MapSet.difference/2)
|
||||
def sinterstore(dest, keys), do: store_set_op(dest, keys, &MapSet.intersection/2)
|
||||
def sunionstore(dest, keys), do: store_set_op(dest, keys, &MapSet.union/2)
|
||||
def sdiffstore(dest, keys), do: store_set_op(dest, keys, &MapSet.difference/2)
|
||||
|
||||
def zadd(key, entries, opts \\ []), do: call_key(key, {:zadd, key, entries, opts})
|
||||
def zscore(key, member), do: call_key(key, {:zscore, key, member})
|
||||
def zrank(key, member), do: call_key(key, {:zrank, key, member})
|
||||
def zincrby(key, delta, member), do: call_key(key, {:zincrby, key, delta, member})
|
||||
def zrem(key, members), do: call_key(key, {:zrem, key, members})
|
||||
def zcard(key), do: call_key(key, {:zcard, key})
|
||||
def zrange(key, start, stop), do: call_key(key, {:zrange, key, start, stop})
|
||||
def zrangebyscore(key, min, max), do: call_key(key, {:zrangebyscore, key, min, max})
|
||||
def zpopmin(key, count), do: call_key(key, {:zpopmin, key, count})
|
||||
def zpopmax(key, count), do: call_key(key, {:zpopmax, key, count})
|
||||
|
||||
def xadd(key, idSpec, fields, trim, nomkstream),
|
||||
do: call_key(key, {:stream, {:xadd, key, idSpec, fields, trim, nomkstream}})
|
||||
|
||||
def xlen(key), do: call_key(key, {:stream, {:xlen, key}})
|
||||
|
||||
def xrange(key, startBound, endBound, count),
|
||||
do: call_key(key, {:stream, {:xrange, key, startBound, endBound, count}})
|
||||
|
||||
def xrevrange(key, endBound, startBound, count),
|
||||
do: call_key(key, {:stream, {:xrevrange, key, endBound, startBound, count}})
|
||||
|
||||
def xread(key, afterId, count), do: call_key(key, {:stream, {:xread, key, afterId, count}})
|
||||
def xdel(key, ids), do: call_key(key, {:stream, {:xdel, key, ids}})
|
||||
def xtrim(key, strategy), do: call_key(key, {:stream, {:xtrim, key, strategy}})
|
||||
def xsetid(key, id), do: call_key(key, {:stream, {:xsetid, key, id}})
|
||||
def xinfoStream(key), do: call_key(key, {:stream, {:xinfoStream, key}})
|
||||
|
||||
def xgroupCreate(key, name, startSpec, mkstream),
|
||||
do: call_key(key, {:stream, {:xgroupCreate, key, name, startSpec, mkstream}})
|
||||
|
||||
def xgroupDestroy(key, name), do: call_key(key, {:stream, {:xgroupDestroy, key, name}})
|
||||
|
||||
def xgroupCreateConsumer(key, name, consumer),
|
||||
do: call_key(key, {:stream, {:xgroupCreateConsumer, key, name, consumer}})
|
||||
|
||||
def xgroupDelConsumer(key, name, consumer),
|
||||
do: call_key(key, {:stream, {:xgroupDelConsumer, key, name, consumer}})
|
||||
|
||||
def xgroupSetId(key, name, id), do: call_key(key, {:stream, {:xgroupSetId, key, name, id}})
|
||||
|
||||
def xreadgroup(key, name, consumer, mode, count, noack),
|
||||
do: call_key(key, {:stream, {:xreadgroup, key, name, consumer, mode, count, noack}})
|
||||
|
||||
def xack(key, name, ids), do: call_key(key, {:stream, {:xack, key, name, ids}})
|
||||
|
||||
def xpendingSummary(key, name), do: call_key(key, {:stream, {:xpendingSummary, key, name}})
|
||||
|
||||
def xpendingRange(key, name, startId, endId, count, consumer, minIdle),
|
||||
do:
|
||||
call_key(
|
||||
key,
|
||||
{:stream, {:xpendingRange, key, name, startId, endId, count, consumer, minIdle}}
|
||||
)
|
||||
|
||||
def xclaim(key, name, consumer, minIdle, ids, opts),
|
||||
do: call_key(key, {:stream, {:xclaim, key, name, consumer, minIdle, ids, opts}})
|
||||
|
||||
def xautoclaim(key, name, consumer, minIdle, startId, count),
|
||||
do: call_key(key, {:stream, {:xautoclaim, key, name, consumer, minIdle, startId, count}})
|
||||
|
||||
def setbit(key, offset, value), do: call_key(key, {:binary, {:setbit, key, offset, value}})
|
||||
def getbit(key, offset), do: call_key(key, {:binary, {:getbit, key, offset}})
|
||||
def bitcount(key, range), do: call_key(key, {:binary, {:bitcount, key, range}})
|
||||
|
||||
def bitpos(key, bit, range, endGiven),
|
||||
do: call_key(key, {:binary, {:bitpos, key, bit, range, endGiven}})
|
||||
|
||||
def bitfield(key, operations, writes?),
|
||||
do: call_key(key, {:binary, {:bitfield, key, operations, writes?}})
|
||||
|
||||
def strlen(key), do: call_key(key, {:binary, {:strlen, key}})
|
||||
def getrange(key, start, stop), do: call_key(key, {:binary, {:getrange, key, start, stop}})
|
||||
def setrange(key, offset, value), do: call_key(key, {:binary, {:setrange, key, offset, value}})
|
||||
|
||||
def pfadd(key, elements), do: call_key(key, {:binary, {:pfadd, key, elements}})
|
||||
def pfget(key), do: call_key(key, {:binary, {:pfget, key}})
|
||||
def pfstore(key, binary), do: call_key(key, {:binary, {:pfstore, key, binary}})
|
||||
|
||||
def bitop(:not, dest, [source]) do
|
||||
with {:ok, binary} <- fetchRawString(source) do
|
||||
result = Bitmap.combine(:not, [binary || ""])
|
||||
storeStringResult(dest, result)
|
||||
end
|
||||
end
|
||||
|
||||
def bitop(operation, dest, sources) do
|
||||
with {:ok, binaries} <- fetchRawStrings(sources) do
|
||||
result = Bitmap.combine(operation, Enum.map(binaries, &(&1 || "")))
|
||||
storeStringResult(dest, result)
|
||||
end
|
||||
end
|
||||
|
||||
def pfcount(keys) do
|
||||
with {:ok, binaries} <- fetchHllBinaries(keys) do
|
||||
present = Enum.reject(binaries, &is_nil/1)
|
||||
count = if present == [], do: 0, else: HyperLogLog.countMerged(present)
|
||||
{:ok, count}
|
||||
end
|
||||
end
|
||||
|
||||
def pfmerge(dest, sources) do
|
||||
with {:ok, binaries} <- fetchHllBinaries([dest | sources]) do
|
||||
present = Enum.reject(binaries, &is_nil/1)
|
||||
merged = if present == [], do: HyperLogLog.empty(), else: HyperLogLog.merge(present)
|
||||
pfstore(dest, merged)
|
||||
{:ok, "OK"}
|
||||
end
|
||||
end
|
||||
|
||||
defp fetchRawString(key) do
|
||||
case call_key(key, {:get, key}) do
|
||||
{:ok, value} -> {:ok, value}
|
||||
nil -> {:ok, nil}
|
||||
{:error, _} = error -> error
|
||||
end
|
||||
end
|
||||
|
||||
defp fetchRawStrings(keys) do
|
||||
Enum.reduce_while(keys, {:ok, []}, fn key, {:ok, acc} ->
|
||||
case fetchRawString(key) do
|
||||
{:ok, value} -> {:cont, {:ok, [value | acc]}}
|
||||
error -> {:halt, error}
|
||||
end
|
||||
end)
|
||||
|> case do
|
||||
{:ok, values} -> {:ok, Enum.reverse(values)}
|
||||
error -> error
|
||||
end
|
||||
end
|
||||
|
||||
defp fetchHllBinaries(keys) do
|
||||
Enum.reduce_while(keys, {:ok, []}, fn key, {:ok, acc} ->
|
||||
case pfget(key) do
|
||||
{:ok, value} -> {:cont, {:ok, [value | acc]}}
|
||||
error -> {:halt, error}
|
||||
end
|
||||
end)
|
||||
|> case do
|
||||
{:ok, values} -> {:ok, Enum.reverse(values)}
|
||||
error -> error
|
||||
end
|
||||
end
|
||||
|
||||
defp storeStringResult(dest, result) do
|
||||
if result == <<>> do
|
||||
call_key(dest, {:del, dest})
|
||||
else
|
||||
call_key(dest, {:set, dest, result, []})
|
||||
end
|
||||
|
||||
{:ok, byte_size(result)}
|
||||
end
|
||||
|
||||
def getdel(key), do: call_key(key, {:extra, {:getdel, key}})
|
||||
def incrbyfloat(key, delta), do: call_key(key, {:extra, {:incrbyfloat, key, delta}})
|
||||
def hsetnx(key, field, value), do: call_key(key, {:extra, {:hsetnx, key, field, value}})
|
||||
def spop(key, count), do: call_key(key, {:extra, {:spop, key, count}})
|
||||
def srandmember(key, count), do: call_key(key, {:extra, {:srandmember, key, count}})
|
||||
def lrem(key, count, value), do: call_key(key, {:extra, {:lrem, key, count, value}})
|
||||
def lset(key, index, value), do: call_key(key, {:extra, {:lset, key, index, value}})
|
||||
def ltrim(key, start, stop), do: call_key(key, {:extra, {:ltrim, key, start, stop}})
|
||||
|
||||
def linsert(key, position, pivot, value),
|
||||
do: call_key(key, {:extra, {:linsert, key, position, pivot, value}})
|
||||
|
||||
def zcount(key, minBound, maxBound),
|
||||
do: call_key(key, {:extra, {:zcount, key, minBound, maxBound}})
|
||||
|
||||
def zrevrange(key, start, stop), do: call_key(key, {:extra, {:zrevrange, key, start, stop}})
|
||||
def zrevrank(key, member), do: call_key(key, {:extra, {:zrevrank, key, member}})
|
||||
def zmscore(key, members), do: call_key(key, {:extra, {:zmscore, key, members}})
|
||||
def expireat(key, msAbs), do: call_key(key, {:extra, {:expireat, key, msAbs}})
|
||||
def expiretime(key, unit), do: call_key(key, {:extra, {:expiretime, key, unit}})
|
||||
|
||||
def lpos(key, element, rank, count, maxlen),
|
||||
do: call_key(key, {:extra, {:lpos, key, element, rank, count, maxlen}})
|
||||
|
||||
def setnx(key, value) do
|
||||
case set(key, value, nx: true) do
|
||||
:ok -> 1
|
||||
nil -> 0
|
||||
end
|
||||
end
|
||||
|
||||
def touch(keys), do: exists(keys)
|
||||
|
||||
def hmget(key, fields) do
|
||||
Enum.map(fields, fn field ->
|
||||
case call_key(key, {:hget, key, field}) do
|
||||
{:ok, value} -> value
|
||||
_ -> nil
|
||||
end
|
||||
end)
|
||||
end
|
||||
|
||||
def smismember(key, members) do
|
||||
Enum.map(members, fn member ->
|
||||
case call_key(key, {:sismember, key, member}) do
|
||||
{:ok, 1} -> 1
|
||||
_ -> 0
|
||||
end
|
||||
end)
|
||||
end
|
||||
|
||||
def sintercard(keys, limit) do
|
||||
with {:ok, members} <- set_op(keys, &MapSet.intersection/2) do
|
||||
count = length(members)
|
||||
{:ok, if(limit == 0, do: count, else: min(count, limit))}
|
||||
end
|
||||
end
|
||||
|
||||
def smove(source, dest, member) do
|
||||
case call_key(source, {:srem, source, [member]}) do
|
||||
{:ok, 0} ->
|
||||
{:ok, 0}
|
||||
|
||||
{:ok, _} ->
|
||||
call_key(dest, {:sadd, dest, [member]})
|
||||
{:ok, 1}
|
||||
|
||||
error ->
|
||||
error
|
||||
end
|
||||
end
|
||||
|
||||
def rpoplpush(source, dest) do
|
||||
case call_key(source, {:rpop, source, nil}) do
|
||||
{:ok, value} ->
|
||||
call_key(dest, {:lpush, dest, [value]})
|
||||
{:ok, value}
|
||||
|
||||
nil ->
|
||||
{:ok, nil}
|
||||
|
||||
error ->
|
||||
error
|
||||
end
|
||||
end
|
||||
|
||||
def rename(source, dest) do
|
||||
case call_key(source, {:peek, source}) do
|
||||
:miss ->
|
||||
{:error, :nokey}
|
||||
|
||||
{kind, value, ea} ->
|
||||
call_key(dest, {:extra, {:raw_put, dest, kind, value, ea}})
|
||||
call_key(source, {:del, source})
|
||||
{:ok, "OK"}
|
||||
end
|
||||
end
|
||||
|
||||
def renamenx(source, dest) do
|
||||
case call_key(source, {:peek, source}) do
|
||||
:miss ->
|
||||
{:error, :nokey}
|
||||
|
||||
entry ->
|
||||
case call_key(dest, {:exists, dest}) do
|
||||
1 ->
|
||||
{:ok, 0}
|
||||
|
||||
0 ->
|
||||
{kind, value, ea} = entry
|
||||
call_key(dest, {:extra, {:raw_put, dest, kind, value, ea}})
|
||||
call_key(source, {:del, source})
|
||||
{:ok, 1}
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
def copy(source, dest, replace) do
|
||||
case call_key(source, {:peek, source}) do
|
||||
:miss ->
|
||||
{:ok, 0}
|
||||
|
||||
{kind, value, ea} ->
|
||||
if not replace and call_key(dest, {:exists, dest}) == 1 do
|
||||
{:ok, 0}
|
||||
else
|
||||
call_key(dest, {:del, dest})
|
||||
{:ok, call_key(dest, {:extra, {:raw_put, dest, kind, value, ea}})}
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
def encoding(key) do
|
||||
case call_key(key, {:peek, key}) do
|
||||
:miss -> nil
|
||||
entry -> {:ok, entry}
|
||||
end
|
||||
end
|
||||
|
||||
def scan(cursor, regex, typeFilter) do
|
||||
shard = cursor
|
||||
keys = GenServer.call(shard_name(shard), {:op, {:extra, {:scan_shard, regex, typeFilter}}})
|
||||
nextCursor = if shard + 1 < shard_count(), do: shard + 1, else: 0
|
||||
{nextCursor, keys}
|
||||
end
|
||||
|
||||
def watch_snapshot(key), do: call_key(key, {:peek, key})
|
||||
def unchanged?(key, snapshot), do: watch_snapshot(key) == snapshot
|
||||
|
||||
def keys(pattern), do: fan_out({:keys, pattern}) |> List.flatten()
|
||||
def dbsize, do: fan_out(:dbsize) |> Enum.sum()
|
||||
|
||||
def flushdb do
|
||||
fan_out(:flushdb)
|
||||
:ok
|
||||
end
|
||||
|
||||
def sweep do
|
||||
fan_out(:sweep) |> List.flatten()
|
||||
end
|
||||
|
||||
def dump do
|
||||
for index <- 0..(shard_count() - 1),
|
||||
entry <- GenServer.call(shard_name(index), :dump) do
|
||||
entry
|
||||
end
|
||||
end
|
||||
|
||||
def restart do
|
||||
for index <- 0..(shard_count() - 1) do
|
||||
Supervisor.terminate_child(Excache.Store.Supervisor, {:shard, index})
|
||||
Supervisor.restart_child(Excache.Store.Supervisor, {:shard, index})
|
||||
end
|
||||
|
||||
:ok
|
||||
end
|
||||
|
||||
defp set_op([first | rest], combine) do
|
||||
with {:ok, initial} <- fetch_set(first) do
|
||||
Enum.reduce_while(rest, {:ok, initial}, fn key, {:ok, acc} ->
|
||||
case fetch_set(key) do
|
||||
{:ok, set} -> {:cont, {:ok, combine.(acc, set)}}
|
||||
error -> {:halt, error}
|
||||
end
|
||||
end)
|
||||
|> case do
|
||||
{:ok, set} -> {:ok, MapSet.to_list(set)}
|
||||
error -> error
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
defp fetch_set(key) do
|
||||
case call_key(key, {:smembers, key}) do
|
||||
{:ok, members} -> {:ok, MapSet.new(members)}
|
||||
error -> error
|
||||
end
|
||||
end
|
||||
|
||||
defp store_set_op(dest, keys, combine) do
|
||||
with {:ok, members} <- set_op(keys, combine) do
|
||||
call_key(dest, {:del, dest})
|
||||
|
||||
if members == [] do
|
||||
{:ok, 0}
|
||||
else
|
||||
{:ok, added} = call_key(dest, {:sadd, dest, members})
|
||||
{:ok, added}
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
Reference in new issue
Block a user