defmodule LibRedis do
@external_resource "README.md"
@moduledoc "README.md"
|> File.read!()
|> String.split("<!-- MDOC !-->")
|> Enum.fetch!(1)
alias LibRedis.{Standalone, Cluster, Typespecs}
@client_options_schema [
name: [
type: :atom,
default: :redis,
doc: "The name of the redis client"
],
mode: [
type: {:in, [:standalone, :cluster]},
default: :standalone,
doc: "The mode of the redis client, :standalone or :cluster"
],
url: [
type: :string,
required: true,
doc: "The url of the redis server, like redis://:123456@localhost:6379"
],
password: [
type: :string,
default: "",
doc: "The password of the redis server, this option is useful in cluster mode"
],
pool_size: [
type: :non_neg_integer,
default: 10,
doc: "The pool size of the redis client's pool"
],
client_store_module: [
type: :any,
default: LibRedis.ClientStore.Default,
doc: "The implementation of the client store, see LibRedis.ClientStore for more details"
],
slot_store_module: [
type: :any,
default: LibRedis.SlotStore.Default,
doc: "The implementation of the slot store, see LibRedis.SlotStore for more details"
]
]
@type options() :: [unquote(NimbleOptions.option_typespec(@client_options_schema))]
# types
@type t :: %__MODULE__{
name: Typespecs.name(),
mode: :cluster | :standalone,
url: Typespecs.url(),
password: Typespecs.password(),
pool_size: non_neg_integer(),
client: Client.client(),
client_store_module: module(),
slot_store_module: module()
}
@enforce_keys ~w(name mode url password pool_size client client_store_module slot_store_module)a
defstruct @enforce_keys
defmodule Client do
alias LibRedis.Typespecs
@type client :: struct()
@type command_t :: [binary() | bitstring()]
@type resp_t :: {:ok, term()} | {:error, term()}
@callback new(keyword()) :: client()
@callback start_link(keyword()) :: Typespecs.on_start()
@callback command(client(), command_t(), keyword()) :: resp_t()
@callback pipeline(client(), [command_t()], keyword()) :: resp_t()
def command(client, command, opts), do: delegate(client, :command, [command, opts])
def pipeline(client, commands, opts), do: delegate(client, :pipeline, [commands, opts])
defp delegate(%module{} = client, func, args),
do: apply(module, func, [client | args])
end
@doc """
create a new redis client
## Options
#{NimbleOptions.docs(@client_options_schema)}
## Examples
iex> standalone_options = [
name: :redis,
mode: :standalone,
url: "redis://:pwd@localhost:6379",
pool_size: 5
]
iex> LibRedis.new(standalone_options)
"""
@spec new(options()) :: t()
def new(opts \\ []) do
opts = opts |> NimbleOptions.validate!(@client_options_schema)
__MODULE__
|> struct(opts)
|> with_client()
end
defp with_client(%__MODULE__{mode: :standalone} = redis) do
client =
Standalone.new(
name: redis.name,
url: redis.url,
pool_size: redis.pool_size
)
%{redis | client: client}
end
defp with_client(%__MODULE__{mode: :cluster} = redis) do
client =
Cluster.new(
name: redis.name,
urls: redis.url |> url_to_cluster_urls(),
password: redis.password,
pool_size: redis.pool_size,
client_store_module: redis.client_store_module,
slot_store_module: redis.slot_store_module
)
%{redis | client: client}
end
@doc """
execute a redis command
## Examples
iex> LibRedis.command(redis, ["SET", "key", "value"])
{:ok, "OK"}
"""
@spec command(t(), Client.command_t(), keyword()) :: Client.resp_t()
def command(redis, command, opts \\ []) do
Client.command(redis.client, command, opts)
end
@doc """
execute a redis pipeline
## Examples
iex> LibRedis.pipeline(redis, [["SET", "key", "value"], ["GET", "key"]])
{:ok, ["OK", "value"]}
"""
@spec pipeline(t(), [Client.command_t()], keyword()) :: Client.resp_t()
def pipeline(redis, commands, opts \\ []) do
Client.pipeline(redis.client, commands, opts)
end
def child_spec(opts) do
redis = Keyword.fetch!(opts, :redis)
%{id: {__MODULE__, redis.name}, start: {__MODULE__, :start_link, [opts]}}
end
@spec start_link(keyword()) :: GenServer.on_start()
def start_link(opts) do
{redis, _opts} = Keyword.pop!(opts, :redis)
case redis.mode do
:standalone ->
LibRedis.Standalone.start_link(pool: redis.client)
:cluster ->
LibRedis.Cluster.start_link(cluster: redis.client)
end
end
defp url_to_cluster_urls(url) do
url
|> String.split(",")
end
end
defmodule LibRedis.Standalone do
@moduledoc false
alias LibRedis.{Client, Pool}
@behaviour Client
@impl Client
defdelegate new(opts), to: Pool
@impl Client
defdelegate command(cli, command, opts), to: Pool
@impl Client
defdelegate pipeline(cli, commands, opts), to: Pool
@impl Client
defdelegate start_link(opts), to: Pool
end
defmodule LibRedis.Cluster do
@moduledoc """
cluster redis client
"""
alias LibRedis.{Client, Typespecs}
@behaviour Client
# types
@type t :: %__MODULE__{
name: Typespecs.name(),
urls: [Typespecs.url()],
password: Typespecs.password(),
pool_size: non_neg_integer(),
refresh_interval_ms: non_neg_integer(),
client_store_module: module(),
slot_store_module: module()
}
@type node_info :: %{ip: bitstring(), port: non_neg_integer()}
@type slot_info :: %{
start_slot: non_neg_integer(),
end_slot: non_neg_integer(),
master: node_info(),
replicas: [node_info()]
}
@url_regex ~r/^redis:\/\/\w+:\d+$/
@enforce_keys ~w(name urls password pool_size refresh_interval_ms client_store_module slot_store_module)a
defstruct @enforce_keys
@impl Client
def new(opts \\ []) do
opts =
opts
|> Keyword.put_new(:name, :cluster)
|> Keyword.put_new(:urls, ["redis://localhost:6379"])
|> Keyword.put_new(:password, "")
|> Keyword.put_new(:pool_size, 10)
|> Keyword.put_new(:refresh_interval_ms, 10_000)
|> Keyword.put_new(:client_store_module, LibRedis.ClientStore.Default)
|> Keyword.put_new(:slot_store_module, LibRedis.SlotStore.Default)
opts[:urls]
|> Enum.map(fn url ->
if not valid_redis_url?(url) do
raise ArgumentError, "invalid url: #{url}"
end
end)
struct(__MODULE__, opts)
end
defp valid_redis_url?(url), do: Regex.match?(@url_regex, url)
def child_spec(opts) do
cluster = Keyword.fetch!(opts, :cluster)
%{id: {__MODULE__, cluster.name}, start: {__MODULE__, :start_link, [opts]}}
end
@impl Client
@spec start_link(keyword()) :: GenServer.on_start()
def start_link(opts) do
{cluster, _opts} = Keyword.pop!(opts, :cluster)
GenServer.start(__MODULE__, cluster, name: cluster.name)
end
@impl Client
def command(client, command, opts \\ []) do
GenServer.call(client.name, {:command, command, opts})
end
@impl Client
def pipeline(client, commands, opts \\ []) do
GenServer.call(client.name, {:pipeline, commands, opts})
end
defp url_with_password(url, ""), do: url
defp url_with_password(url, password) do
prefix = String.slice(url, 0, 8)
prefix <> ":" <> password <> "@" <> String.slice(url, 8..-1)
end
defp random_name() do
:crypto.strong_rand_bytes(16)
|> Base.encode64()
|> String.to_atom()
end
defp parse_url(url) do
[ip, port] =
url
|> String.slice(8..-1)
|> String.split(":")
{ip, String.to_integer(port)}
end
def init(cluster) do
slot_store = cluster.slot_store_module.new()
client_store = cluster.client_store_module.new()
{:ok, _} = cluster.slot_store_module.start_link(store: slot_store)
{:ok, _} = cluster.client_store_module.start_link(store: client_store)
cluster.urls
|> Enum.map(fn url ->
{host, port} = parse_url(url)
cluster.client_store_module.new(
host: host,
port: port,
opts: [
name: random_name(),
pool_size: cluster.pool_size,
url: url_with_password(url, cluster.password)
]
)
end)
|> Enum.each(&LibRedis.ClientStore.get(&1))
send(self(), :refresh)
{:ok, %{cluster: cluster, client_store: client_store, slot_store: slot_store}}
end
# Response Format:
# - Start slot range
# - End slot range
# - Master for slot range represented as nested IP/Port array
# - First replica of master for slot range
# - Second replica
# ...continues until all replicas for this master are returned.
# Ref - https://redis.io/commands/cluster-slots#nested-result-array
@spec parse_slot_info(any) :: [slot_info()]
defp parse_slot_info(slot_info) do
Enum.map(slot_info, fn [start_slot, end_slot, master | replicas] ->
%{
start_slot: start_slot,
end_slot: end_slot,
master: parse_node_info(master),
replicas: Enum.map(replicas, &parse_node_info/1)
}
end)
end
defp parse_node_info([node_ip, node_port, _node_id | _] = _node) do
%{
ip: node_ip,
port: node_port
}
end
defp refetch_slot_info(client_store) do
client_store
|> LibRedis.ClientStore.random()
|> LibRedis.Standalone.command(["CLUSTER", "SLOTS"], [])
|> case do
{:ok, slot_info} -> slot_info
{:error, _} -> []
end
|> parse_slot_info()
end
def handle_info(:refresh, %{client_store: cs, slot_store: ss, cluster: cluster} = state) do
cs
|> refetch_slot_info()
|> (fn x -> LibRedis.SlotStore.put(ss, x) end).()
Process.send_after(self(), :refresh, cluster.refresh_interval_ms)
{:noreply, state}
end
def handle_call({:command, command, opts}, _from, state) do
try_command(state, command, opts, 3)
end
def handle_call({:pipeline, commands, opts}, _from, state) do
try_pipeline(state, commands, opts)
end
defp try_command(state, command, opts, retries_left) do
case do_command(state, command, opts) do
{:ok, result} ->
{:reply, {:ok, result}, state}
{:error, %Redix.Error{message: msg}} when retries_left > 0 ->
if String.starts_with?(msg, "MOVED") do
send(self(), :refresh)
end
try_command(state, command, opts, retries_left - 1)
{:error, _reason} when retries_left > 0 ->
try_command(state, command, opts, retries_left - 1)
{:error, reason} ->
{:reply, {:error, reason}, state}
end
end
defp try_pipeline(state, commands, opts) do
res =
group_commands(commands, state, %{})
|> Map.to_list()
|> Enum.map(fn {client, cmds} ->
do_pipeline(client, cmds, opts, 3)
|> case do
{:ok, res} ->
res
{:error, :moved} ->
try_pipeline(state, cmds, opts)
{:error, error} ->
raise error
end
end)
|> List.flatten()
reply =
commands
|> Enum.map(fn cmd ->
{_, ret} =
Enum.find(res, fn
{^cmd, _} -> true
_ -> false
end)
ret
end)
{:reply, {:ok, reply}, state}
end
defp do_command(
%{cluster: cluster, slot_store: ss},
[_, key | _] = command,
opts
) do
target = LibRedis.SlotFinder.hash_slot(key)
ss
|> LibRedis.SlotStore.get()
|> Enum.find(fn x -> x.start_slot <= target and x.end_slot >= target end)
|> case do
nil ->
{:error, "slot not found"}
slot ->
cluster.client_store_module.new(
host: slot.master.ip,
port: slot.master.port,
opts: [
name: random_name(),
pool_size: cluster.pool_size,
url:
url_with_password(
"redis://:#{cluster.password}@#{slot.master.ip}:#{slot.master.port}",
""
)
]
)
|> LibRedis.ClientStore.get()
|> LibRedis.Standalone.command(command, opts)
end
end
defp do_pipeline(client, commands, opts, retries_left) do
case LibRedis.Standalone.pipeline(client, commands, opts) do
{:ok, result} ->
{:ok, Enum.zip(commands, result)}
{:error, %Redix.Error{message: msg}} when retries_left > 0 ->
if String.starts_with?(msg, "MOVED") do
send(self(), :refresh)
{:error, :moved}
else
do_pipeline(client, commands, opts, retries_left - 1)
end
{:error, _reason} when retries_left > 0 ->
do_pipeline(client, commands, opts, retries_left - 1)
{:error, reason} ->
{:error, reason}
end
end
# group_commands takes a list of commands and groups them by slot.
# returns a map of standalone-client -> command list
defp group_commands([], _, acc), do: acc
defp group_commands([command | t], %{slot_store: ss, cluster: cluster} = state, acc) do
[_, key | _] = command
target = LibRedis.SlotFinder.hash_slot(key)
ss
|> LibRedis.SlotStore.get()
|> Enum.find(fn x -> x.start_slot <= target and x.end_slot >= target end)
|> case do
nil ->
{:error, "slot not found"}
slot ->
cli =
cluster.client_store_module.new(
host: slot.master.ip,
port: slot.master.port,
opts: [
name: random_name(),
pool_size: cluster.pool_size,
url:
url_with_password(
"redis://:#{cluster.password}@#{slot.master.ip}:#{slot.master.port}",
""
)
]
)
|> LibRedis.ClientStore.get()
group_commands(t, state, Map.update(acc, cli, [command], &(&1 ++ [command])))
end
end
end
