elib

-module(redis_proxy).
-include_lib("kernel/include/logger.hrl").
%% API

-export([

  start_link/1,

  start_link/2,

  stop/0,

  stop/1,

  status/0,

  status/1

]).

-export([

  q/1,

  q/2,

  q/3,

  qp/1,

  qp/2,

  qp/3,

  foreach_redis/4,

  foreach_redis/5

]).

%% 利用redis自旋锁

-export([

  trans/2,

  trans/3,

  trans/5

]).

-export([

  del/1, del/3,

  exists/1, exists/3,

  expire/2, expire/4,

  ttl/1, ttl/3,

  type/1, type/3

]).

-export([

  append/2, append/4,

  decr/1, decr/3,

  decrby/2, decrby/4,

  get/1, get/3,

  get_del/1, get_del/3,

  get_ex/2, get_ex/4,

  get_range/3, get_range/5,

  incr/1, incr/3,

  incrby/2, incrby/4,

  mget/1, mget/3,

  mset/1, mset/3,

  mset_nx/1, mset_nx/3,

  set/3, set/5,

  set_ex/3, set_ex/5,

  set_nx/2, set_nx/4,

  set_range/3, set_range/5

]).

-define(SERVER, ?MODULE).
-define(DEFAULT_POOL_SIZE, 16).
-define(TIMEOUT, 5000).
-define(REDIS_LOCK_TIME, 5000).
-define(REDIS_TRANS_TIMEOUT, infinity).
-define(REDIS_TIMEOUT, 5000).
-define(TRANS_SLEEP_TIME_MAX, 100).
-define(BOOL_REPLY_FUN, fun(<<"OK">>) -> true;(_) -> false end).
-define(INTEGER_REPLY_FUN, fun(X) -> erlang:binary_to_integer(X) end).
-define(REPLY_FUN, fun(X) -> X end).

start_link(Args) ->
  L = parse_config(Args),
  poolboy:start_link(L).

start_link(Name, Args) when is_atom(Name) ->
  L = parse_config(Args),
  L2 = [{name, {local, Name}} | L],
  poolboy:start_link(L2).

parse_config(M) when is_map(M) ->
  Config = #{worker_module => redis_worker, size => ?DEFAULT_POOL_SIZE},
  M2 = maps:merge(Config, M),
  maps:to_list(M2);
parse_config(L) when is_list(L) ->
  parse_config(maps:from_list(L)).


stop() ->
  stop(?SERVER).

stop(Name) when is_atom(Name) ->
  poolboy:stop(Name).

status() ->
  status(?SERVER).

status(Name) when is_atom(Name) ->
  poolboy:status(Name).

q(Command) -> q(Command, ?TIMEOUT).

q(Command, Timeout) -> q(?SERVER, Command, Timeout).

q(Name, Command, Timeout) ->
  poolboy:transaction(Name, fun(PID) -> eredis:q(PID, Command) end, Timeout).

qp(Command) -> qp(Command, ?TIMEOUT).

qp(Command, Timeout) -> qp(?SERVER, Command, Timeout).

qp(Name, Command, Timeout) ->
  poolboy:transaction(Name, fun(PID) -> eredis:qp(PID, Command) end, Timeout).

foreach_redis(REG, Fun, CntPerTime, Timeout) ->
  foreach_redis(?SERVER, REG, Fun, CntPerTime, Timeout).

foreach_redis(Name, REG, Fun, CntPerTime, Timeout) ->
  foreach_redis(Name, <<"0">>, 0, REG, CntPerTime, Timeout, Fun).

foreach_redis(_Name, <<"0">>, Times, _REG, _CNT, _Timeout, _Fun) when Times > 0 ->
  ok;
foreach_redis(Name, Cur, Times, REG, CNT, Timeout, Fun) ->
  CMD = case REG == "*" of
          true -> ["SCAN", Cur, "COUNT", erlang:integer_to_list(CNT)];
          false -> ["SCAN", Cur, "MATCH", REG, "COUNT", erlang:integer_to_list(CNT)]
        end,
  case q(Name, CMD, Timeout) of
    {ok, [Cur2, []]} ->
      foreach_redis(Name, Cur2, Times + 1, REG, CNT, Timeout, Fun);
    {ok, [Cur2, L]} ->
      case Fun of
        {M, F} -> M:F(L);
        _ -> Fun(L)
      end,
      foreach_redis(Name, Cur2, Times + 1, REG, CNT, Timeout, Fun)
  end.

%% ====================

%% Fun = {M,F,A} | fun/0 | {fun/x,Args}

%% @return {ok, Result} | {error,timeout}

trans(LockName, Fun) when is_binary(LockName) ->
  trans(?SERVER, LockName, Fun, ?REDIS_LOCK_TIME, ?REDIS_TRANS_TIMEOUT).

trans(Name, LockName, Fun) when is_binary(LockName) ->
  trans(Name, LockName, Fun, ?REDIS_LOCK_TIME, ?REDIS_TRANS_TIMEOUT).

trans(Name, LockName, Fun, LockTime, Timeout) when Timeout =:= infinity ->
  RandBin = erlang:term_to_binary(erlang:make_ref()),
  trans_i(Name, LockName, RandBin, Fun, LockTime, infinity);
trans(Name, LockName, Fun, LockTime, Timeout) when is_integer(Timeout) ->
  RandBin = erlang:term_to_binary(erlang:make_ref()),
  Now = os:perf_counter(millisecond),
  EndTime = Now + Timeout,
  trans_i(Name, LockName, RandBin, Fun, LockTime, EndTime).

trans_i(Name, LockKey, RandBin, Fun, LockTime, EndTime) ->
  Now = os:perf_counter(millisecond),
  Timeout = case EndTime of
              infinity -> false;
              _ -> Now >= EndTime
            end,
  if
    Timeout ->
      {error, timeout};
    true ->
      case redis_proxy:q(Name, ["SET", LockKey, RandBin, "PX", erlang:integer_to_list(LockTime), "NX"], ?REDIS_TIMEOUT) of
        {ok, <<"OK">>} ->
          Ret = case Fun of
                  {M, F, A} -> catch apply(M, F, A);
                  {F, A} when is_function(F) -> catch erlang:apply(F, A);
                  _ when is_function(Fun, 0) -> catch Fun()
                end,
          case redis_proxy:q(Name, ["GET", LockKey], ?REDIS_TIMEOUT) of
            {ok, RandBin} -> redis_proxy:q(Name, ["DEL", LockKey], ?REDIS_TIMEOUT);
            _ -> pass
          end,
          {ok, Ret};
        {ok, undefined} ->
          Left = EndTime - os:perf_counter(millisecond),
          if
            Left =< 1 ->
              {error, timeout};
            true ->
              Random = rand:uniform(?TRANS_SLEEP_TIME_MAX),
              Sleep = erlang:min(Random, Left - 1),
              timer:sleep(Sleep),
              trans_i(Name, LockKey, RandBin, Fun, LockTime, EndTime)
          end;
        Err ->
          % redis error

          Err
      end
  end.
%%%%%%%

join_command(C, K) when is_binary(K) ->
  [C, K];
join_command(C, L) when is_list(L) ->
  join_command(L, C, []);
join_command(C, M) when is_map(M) ->
  join_command(maps:to_list(M), C, []).

join_command([], C, Acc) ->
  [C | lists:reverse(Acc)];
join_command([E | L], C, Acc) when is_binary(E);is_list(E) ->
  join_command(L, C, [E | Acc]);
join_command([{K, V} | L], C, Acc) ->
  join_command(L, C, [V, K | Acc]).

exec(Name, CMD, Timeout, Parse) ->
  case q(Name, CMD, Timeout) of
    {ok, Result} -> {ok, Parse(Result)};
    Err -> Err
  end.

%% keys


%% @doc del(<<"foo">>) or del([<<"foo">>,<<"bar">>...])

del(L) ->
  del(?SERVER, L, ?TIMEOUT).


del(Name, L, Timeout) ->
  exec(Name, join_command(<<"DEL">>, L), Timeout, ?INTEGER_REPLY_FUN).

exists(L) ->
  exists(?SERVER, L, ?TIMEOUT).


exists(Name, L, Timeout) ->
  exec(Name, join_command(<<"EXISTS">>, L), Timeout, ?INTEGER_REPLY_FUN).

expire(K, Sec) ->
  expire(?SERVER, K, Sec, ?TIMEOUT).

expire(Name, K, Sec, Timeout) when is_binary(K), is_integer(Sec) ->
  exec(Name, join_command(<<"EXPIRE">>, [K, erlang:integer_to_binary(Sec)]), Timeout, ?INTEGER_REPLY_FUN).


ttl(K) ->
  ttl(?SERVER, K, ?TIMEOUT).

ttl(Name, K, Timeout) when is_binary(K) ->
  exec(Name, join_command(<<"EXISTS">>, K), Timeout, ?INTEGER_REPLY_FUN).

type(K) ->
  type(?SERVER, K, ?TIMEOUT).

%% @doc string,list,set,zset,hash,stream

type(Name, K, Timeout) when is_binary(K) ->
  Func = fun(X) ->
    if
      X == <<"string">> -> string;
      X == <<"list">> -> list;
      X == <<"set">> -> set;
      X == <<"zset">> -> zset;
      X == <<"hash">> -> hash;
      X == <<"stream">> -> stream;
      true -> erlang:binary_to_atom(X)
    end end,
  exec(Name, join_command(<<"TYPE">>, K), Timeout, Func).

%% string

append(K, V) ->
  append(?SERVER, K, V, ?TIMEOUT).

append(Name, K, V, Timeout) ->
  exec(Name, join_command(<<"APPEND">>, [K, V]), Timeout, ?INTEGER_REPLY_FUN).

decr(K) ->
  decr(?SERVER, K, ?TIMEOUT).

decr(Name, K, Timeout) when is_binary(K) ->
  exec(Name, join_command(<<"DECR">>, K), Timeout, ?INTEGER_REPLY_FUN).

decrby(K, V) ->
  decrby(?SERVER, K, V, ?TIMEOUT).

decrby(Name, K, V, Timeout) when is_binary(K), is_integer(V) ->
  exec(Name, join_command(<<"DECRBY">>, [K, erlang:integer_to_binary(V)]), Timeout, ?INTEGER_REPLY_FUN).

get(K) -> get(?SERVER, K, ?TIMEOUT).

get(Name, K, Timeout) when is_binary(K) ->
  exec(Name, join_command(<<"GET">>, K), Timeout, ?REPLY_FUN).

get_del(K) -> get_del(?SERVER, K, ?TIMEOUT).

get_del(Name, K, Timeout) when is_binary(K) ->
  exec(Name, join_command(<<"GETDEL">>, K), Timeout, ?REPLY_FUN).

get_ex(K, Args) ->
  get_ex(?SERVER, K, Args, ?TIMEOUT).

get_ex(Name, K, Args, Timeout) when is_binary(K), is_list(Args) ->
  exec(Name, join_command(<<"GETEX">>, [K | Args]), Timeout, ?REPLY_FUN).

get_range(K, S, E) ->
  get_range(?SERVER, K, S, E, ?TIMEOUT).

get_range(Name, K, S, E, Timeout) when is_binary(K), is_integer(S), is_integer(E), S =< E ->
  exec(Name, join_command(<<"DECRBY">>, [K, erlang:integer_to_binary(S), erlang:integer_to_binary(E)]), Timeout, ?REPLY_FUN).

incr(K) ->
  incr(?SERVER, K, ?TIMEOUT).
incr(Name, K, Timeout) ->
  exec(Name, join_command(<<"INCR">>, K), Timeout, ?INTEGER_REPLY_FUN).

incrby(K, V) -> incrby(?SERVER, K, V, ?TIMEOUT).

incrby(Name, K, V, Timeout) when is_binary(K), is_integer(V) ->
  exec(Name, join_command(<<"INCRBY">>, [K, erlang:integer_to_binary(V)]), Timeout, ?INTEGER_REPLY_FUN).

mget(L) ->
  mget(?SERVER, L, ?TIMEOUT).

mget(Name, L, Timeout) ->
  exec(Name, join_command(<<"MGET">>, L), Timeout, ?REPLY_FUN).

mset(MapOrList) ->
  mset(?SERVER, MapOrList, ?TIMEOUT).

mset(Name, MapOrList, Timeout) ->
  exec(Name, join_command(<<"MSET">>, MapOrList), Timeout, ?BOOL_REPLY_FUN).

mset_nx(MapOrList) ->
  mset_nx(?SERVER, MapOrList, ?TIMEOUT).

mset_nx(Name, MapOrList, Timeout) ->
  exec(Name, join_command(<<"MSETNX">>, MapOrList), Timeout, ?BOOL_REPLY_FUN).

set(K, V, Args) ->
  set(?SERVER, K, V, Args, ?TIMEOUT).

set(Name, K, V, Args, Timeout) ->
  CMD = [<<"SET">>, K, V | Args],
  exec(Name, CMD, Timeout, ?BOOL_REPLY_FUN).

set_ex(K, V, Expire) ->
  set_ex(?SERVER, K, V, Expire, ?TIMEOUT).

set_ex(Name, K, V, Expire, Timeout) when is_binary(K), is_binary(V), is_integer(Expire) ->
  exec(Name, join_command(<<"SETEX">>, [K, erlang:integer_to_binary(Expire), V]), Timeout, ?BOOL_REPLY_FUN).

set_nx(K, V) ->
  set_nx(?SERVER, K, V, ?TIMEOUT).

set_nx(Name, K, V, Timeout) when is_binary(K), is_binary(V) ->
  exec(Name, join_command(<<"SETEX">>, [K, V]), Timeout, ?INTEGER_REPLY_FUN).

set_range(K, Offset, V) ->
  set_range(?SERVER, K, Offset, V, ?TIMEOUT).

set_range(Name, K, Offset, V, Timeout) when is_binary(K), is_binary(V), is_integer(Offset) ->
  exec(Name, join_command(<<"SETRANGE">>, [K, erlang:integer_to_binary(Offset), V]), Timeout, ?INTEGER_REPLY_FUN).

%% bitmaps