defmodule LibRedis.Client do
@moduledoc """
Redis client behaviour, there are 2 implementations of this behaviour:
- LibRedis.Standalone
- LibRedis.Cluster
"""
alias LibRedis.{Typespecs, Error}
@type t :: struct()
@opaque command_t :: Typespecs.command_t()
@type resp_t :: {:ok, term()} | Error.t()
@callback new(keyword()) :: t()
@callback start_link(keyword()) :: Typespecs.on_start()
@callback command(t(), command_t(), keyword()) :: resp_t()
@callback pipeline(t(), [command_t()], keyword()) :: resp_t()
@spec command(t(), command_t(), keyword()) :: resp_t()
def command(client, command, opts), do: delegate(client, :command, [command, opts])
@spec pipeline(t(), [command_t()], keyword()) :: resp_t()
def pipeline(client, commands, opts), do: delegate(client, :pipeline, [commands, opts])
defp delegate(%module{} = client, func, args),
do: apply(module, func, [client | args])
end
defmodule LibRedis do
@external_resource "README.md"
@moduledoc "README.md"
|> File.read!()
|> String.split("<!-- MDOC !-->")
|> Enum.fetch!(1)
alias LibRedis.{Standalone, Cluster, Typespecs, Client}
@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"
]
]
@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.t(),
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
@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
)
%{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 """
Standalone redis client is just a delegate to LibRedis.Pool
"""
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, Error}
require Logger
@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: ClientStore.t(),
slot_store: SlotStore.t()
}
@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()]
}
@opaque state :: %{cluster: t()}
@url_regex ~r/^redis:\/\/\w+:\d+$/
defstruct [
:name,
:urls,
:password,
:pool_size,
:refresh_interval_ms,
:client_store,
:slot_store
]
@impl Client
def new(opts \\ []) do
opts =
opts
|> Keyword.put_new(:name, :cluster)
|> Keyword.put_new(:password, "")
|> Keyword.put_new(:pool_size, 10)
|> Keyword.put_new(:refresh_interval_ms, 10_000)
|> Keyword.put_new(:client_store, LibRedis.ClientStore.new())
|> Keyword.put_new(:slot_store, LibRedis.SlotStore.new())
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 error_log(cmd, error),
do: Logger.error(%{"mode" => "cluster", "error" => error, "cmd" => cmd})
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(cluster: t()) :: Typespecs.on_start()
def start_link(cluster: cluster) do
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
@spec parse_url(Typespecs.url()) :: {bitstring(), non_neg_integer()}
defp parse_url(url) do
[ip, port] =
url
|> String.slice(8..-1)
|> String.split(":")
{ip, String.to_integer(port)}
end
def init(cluster) do
{:ok, _} = LibRedis.ClientStore.start_link(store: cluster.client_store)
{:ok, _} = LibRedis.SlotStore.start_link(store: cluster.slot_store)
cluster.urls
|> Enum.map(fn url ->
[
name: {:via, Registry, {cluster.client_store.name, parse_url(url)}},
pool_size: cluster.pool_size,
url: url_with_password(url, cluster.password)
]
end)
|> Enum.map(&LibRedis.Pool.new(&1))
|> Enum.map(&LibRedis.Pool.start_link(pool: &1))
send(self(), :refresh)
{:ok, %{cluster: cluster}}
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
@spec refetch_slot_info(ClientStore.t()) :: [slot_info()]
defp refetch_slot_info(client_store) do
client_store
|> LibRedis.ClientStore.random()
|> LibRedis.Pool.command(["CLUSTER", "SLOTS"])
|> case do
{:ok, slot_info} -> slot_info
{:error, _} -> []
end
|> parse_slot_info()
end
def handle_info(:refresh, %{cluster: cluster} = state) do
cluster.client_store
|> refetch_slot_info()
|> (fn x -> LibRedis.SlotStore.put(cluster.slot_store, x) end).()
Process.send_after(self(), :refresh, cluster.refresh_interval_ms)
{:noreply, state}
end
def handle_call({:command, command, opts}, _from, state) do
{:reply, try_command(state, command, opts, 3), state}
end
def handle_call({:pipeline, commands, opts}, _from, state) do
res = try_pipeline(state, commands, opts, 3)
# resort by original order
commands
|> Enum.map(fn cmd ->
{_, ret} =
Enum.find(res, fn
{^cmd, _} -> true
_ -> false
end)
ret
end)
|> (fn x -> {:reply, {:ok, x}, state} end).()
end
@spec try_command(state(), Typespecs.command_t(), keyword(), non_neg_integer()) ::
{:ok, any()} | {:error, any()}
defp try_command(state, command, opts, retries_left) do
case do_command(state, command, opts) do
{:ok, result} ->
{:ok, result}
{:error, %Redix.Error{message: msg}} when retries_left > 0 ->
if String.starts_with?(msg, "MOVED") do
send(self(), :refresh)
try_command(state, command, opts, retries_left - 1)
else
{:error, Error.new(msg)}
end
{:error, reason} ->
error_log(command, reason)
{:error, Error.new(inspect(reason))}
end
end
@spec try_pipeline(state(), [Typespecs.command_t()], keyword(), non_neg_integer()) ::
[{Typespecs.command_t(), any()}] | {:error, any()}
defp try_pipeline(state, commands, opts, retries_left) do
group_commands(commands, state, %{})
|> Map.to_list()
|> Enum.reduce_while([], fn {client, cmds}, acc ->
do_pipeline(client, cmds, opts)
|> case do
{:ok, res} ->
{:cont, acc ++ res}
{:error, :moved} when retries_left > 0 ->
res = try_pipeline(state, cmds, opts, retries_left - 1)
{:cont, acc ++ res}
err ->
{:halt, err}
end
end)
|> Enum.reverse()
end
@spec do_command(state(), Typespecs.command_t(), Keyword.t()) :: {:ok, term()} | {:error, any}
defp do_command(
%{cluster: cluster},
[_, key | _] = command,
opts
) do
get_client_by_key(cluster, key)
|> case do
nil -> {:error, "slot not found"}
client -> LibRedis.Pool.command(client, command, opts)
end
end
@spec do_pipeline(
LibRedis.Pool.t() | pid(),
[Typespecs.command_t()],
Keyword.t()
) ::
{:ok, [{Typespecs.command_t(), term()}]} | {:error, any}
defp do_pipeline(client, commands, opts) do
case LibRedis.Pool.pipeline(client, commands, opts) do
{:ok, result} ->
{:ok, Enum.zip(commands, result)}
{:error, %Redix.Error{message: msg}} ->
if String.starts_with?(msg, "MOVED") do
send(self(), :refresh)
{:error, :moved}
else
{:error, Error.new(msg)}
end
other ->
other
end
end
# group_commands takes a list of commands and groups them by slot.
# returns a map of standalone-client -> command list
@spec group_commands([Typespecs.command_t()], state(), map()) ::
%{
(LibRedis.Pool.t() | pid()) => [Typespecs.command_t()]
}
| {:error, any()}
defp group_commands([], _, acc), do: acc
defp group_commands([command | t], %{cluster: cluster} = state, acc) do
[_, key | _] = command
get_client_by_key(cluster, key)
|> case do
nil ->
{:error, "slot not found"}
cli ->
group_commands(t, state, Map.update(acc, cli, [command], &(&1 ++ [command])))
end
end
@spec load_client(t(), Typespecs.slot()) :: LibRedis.Pool.t() | pid()
defp load_client(cluster, slot) do
cluster.client_store
|> LibRedis.ClientStore.get({slot.master.ip, slot.master.port})
|> case do
nil ->
[
name: {:via, Registry, {cluster.client_store.name, {slot.master.ip, slot.master.port}}},
pool_size: cluster.pool_size,
url:
url_with_password(
"redis://:#{cluster.password}@#{slot.master.ip}:#{slot.master.port}",
""
)
]
|> LibRedis.Pool.new()
|> tap(&LibRedis.Pool.start_link(pool: &1))
client ->
client
end
end
@spec get_client_by_key(t(), bitstring()) :: LibRedis.Pool.t() | pid() | nil
defp get_client_by_key(cluster, key) do
target = LibRedis.SlotFinder.hash_slot(key)
cluster.slot_store
|> LibRedis.SlotStore.get()
|> Enum.find(fn x -> x.start_slot <= target and x.end_slot >= target end)
|> case do
nil ->
nil
slot ->
load_client(cluster, slot)
end
end
end
