Chapril
/
xmpp.chapril.org-ejabberd
mirror of https://github.com/processone/ejabberd.git
24
1
Fork 0
Code Releases Activity

Improve redis related code

This commit is contained in:
Evgeniy Khramtsov 2017-04-02 11:56:09 +03:00
parent 5087e9c2df
commit 9d9037856c
5 changed files with 265 additions and 149 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

133
src/ejabberd_redis.erl
View File

@ -32,7 +32,9 @@
%% API %% API
-export([start_link/0, q/1, qp/1, config_reloaded/0, opt_type/1]). -export([start_link/0, q/1, qp/1, config_reloaded/0, opt_type/1]).
%% Commands %% Commands
-export([multi/1, get/1, set/2, del/1, sadd/2, srem/2, smembers/1, scard/1]). -export([multi/1, get/1, set/2, del/1,
sadd/2, srem/2, smembers/1, sismember/2, scard/1,
hget/2, hset/3, hdel/2, hlen/1, hgetall/1, hkeys/1]).
%% gen_server callbacks %% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
@ -57,12 +59,14 @@ start_link() ->
q(Command) -> q(Command) ->
try eredis:q(?PROCNAME, Command) try eredis:q(?PROCNAME, Command)
catch _:Reason -> {error, Reason} catch _:{noproc, _} -> {error, disconnected};
_:{timeout, _} -> {error, timeout}
end. end.
qp(Pipeline) -> qp(Pipeline) ->
try eredis:qp(?PROCNAME, Pipeline) try eredis:qp(?PROCNAME, Pipeline)
catch _:Reason -> {error, Reason} catch _:{noproc, _} -> {error, disconnected};
_:{timeout, _} -> {error, timeout}
end. end.
-spec multi(fun(() -> any())) -> {ok, list()} | redis_error(). -spec multi(fun(() -> any())) -> {ok, list()} | redis_error().
@ -95,6 +99,7 @@ config_reloaded() ->
?MODULE ! disconnect ?MODULE ! disconnect
end. end.
-spec get(iodata()) -> {ok, undefined | binary()} | redis_error().
get(Key) -> get(Key) ->
case erlang:get(?TR_STACK) of case erlang:get(?TR_STACK) of
undefined -> undefined ->
@ -113,11 +118,12 @@ set(Key, Val) ->
{error, _} = Err -> Err {error, _} = Err -> Err
end; end;
Stack -> Stack ->
erlang:put(?TR_STACK, [Cmd|Stack]), tr_enq(Cmd, Stack)
queued
end. end.
-spec del(list()) -> {ok, non_neg_integer()} | redis_error() | queued. -spec del(list()) -> {ok, non_neg_integer()} | redis_error() | queued.
del([]) ->
reply(0);
del(Keys) -> del(Keys) ->
Cmd = [<<"DEL">>|Keys], Cmd = [<<"DEL">>|Keys],
case erlang:get(?TR_STACK) of case erlang:get(?TR_STACK) of
@ -127,11 +133,12 @@ del(Keys) ->
{error, _} = Err -> Err {error, _} = Err -> Err
end; end;
Stack -> Stack ->
erlang:put(?TR_STACK, [Cmd|Stack]), tr_enq(Cmd, Stack)
queued
end. end.
-spec sadd(iodata(), list()) -> {ok, non_neg_integer()} | redis_error() | queued. -spec sadd(iodata(), list()) -> {ok, non_neg_integer()} | redis_error() | queued.
sadd(_Set, []) ->
reply(0);
sadd(Set, Members) -> sadd(Set, Members) ->
Cmd = [<<"SADD">>, Set|Members], Cmd = [<<"SADD">>, Set|Members],
case erlang:get(?TR_STACK) of case erlang:get(?TR_STACK) of
@ -141,11 +148,12 @@ sadd(Set, Members) ->
{error, _} = Err -> Err {error, _} = Err -> Err
end; end;
Stack -> Stack ->
erlang:put(?TR_STACK, [Cmd|Stack]), tr_enq(Cmd, Stack)
queued
end. end.
-spec srem(iodata(), list()) -> {ok, non_neg_integer()} | redis_error() | queued. -spec srem(iodata(), list()) -> {ok, non_neg_integer()} | redis_error() | queued.
srem(_Set, []) ->
reply(0);
srem(Set, Members) -> srem(Set, Members) ->
Cmd = [<<"SREM">>, Set|Members], Cmd = [<<"SREM">>, Set|Members],
case erlang:get(?TR_STACK) of case erlang:get(?TR_STACK) of
@ -155,8 +163,7 @@ srem(Set, Members) ->
{error, _} = Err -> Err {error, _} = Err -> Err
end; end;
Stack -> Stack ->
erlang:put(?TR_STACK, [Cmd|Stack]), tr_enq(Cmd, Stack)
queued
end. end.
-spec smembers(iodata()) -> {ok, [binary()]} | redis_error(). -spec smembers(iodata()) -> {ok, [binary()]} | redis_error().
@ -168,6 +175,18 @@ smembers(Set) ->
{error, transaction_unsupported} {error, transaction_unsupported}
end. end.
-spec sismember(iodata(), iodata()) -> boolean() | redis_error().
sismember(Set, Member) ->
case erlang:get(?TR_STACK) of
undefined ->
case q([<<"SISMEMBER">>, Set, Member]) of
{ok, Flag} -> {ok, dec_bool(Flag)};
{error, _} = Err -> Err
end;
_ ->
{error, transaction_unsupported}
end.
-spec scard(iodata()) -> {ok, non_neg_integer()} | redis_error(). -spec scard(iodata()) -> {ok, non_neg_integer()} | redis_error().
scard(Set) -> scard(Set) ->
case erlang:get(?TR_STACK) of case erlang:get(?TR_STACK) of
@ -182,6 +201,76 @@ scard(Set) ->
{error, transaction_unsupported} {error, transaction_unsupported}
end. end.
-spec hget(iodata(), iodata()) -> {ok, undefined | binary()} | redis_error().
hget(Key, Field) ->
case erlang:get(?TR_STACK) of
undefined ->
q([<<"HGET">>, Key, Field]);
_ ->
{error, transaction_unsupported}
end.
-spec hset(iodata(), iodata(), iodata()) -> {ok, boolean()} | redis_error() | queued.
hset(Key, Field, Val) ->
Cmd = [<<"HSET">>, Key, Field, Val],
case erlang:get(?TR_STACK) of
undefined ->
case q(Cmd) of
{ok, Flag} -> {ok, dec_bool(Flag)};
{error, _} = Err -> Err
end;
Stack ->
tr_enq(Cmd, Stack)
end.
-spec hdel(iodata(), list()) -> {ok, non_neg_integer()} | redis_error() | queued.
hdel(_Key, []) ->
reply(0);
hdel(Key, Fields) ->
Cmd = [<<"HDEL">>, Key|Fields],
case erlang:get(?TR_STACK) of
undefined ->
case q(Cmd) of
{ok, N} -> {ok, binary_to_integer(N)};
{error, _} = Err -> Err
end;
Stack ->
tr_enq(Cmd, Stack)
end.
-spec hgetall(iodata()) -> {ok, [{binary(), binary()}]} | redis_error().
hgetall(Key) ->
case erlang:get(?TR_STACK) of
undefined ->
case q([<<"HGETALL">>, Key]) of
{ok, Pairs} -> {ok, decode_pairs(Pairs)};
{error, _} = Err -> Err
end;
_ ->
{error, transaction_unsupported}
end.
-spec hlen(iodata()) -> {ok, non_neg_integer()} | redis_error().
hlen(Key) ->
case erlang:get(?TR_STACK) of
undefined ->
case q([<<"HLEN">>, Key]) of
{ok, N} -> {ok, binary_to_integer(N)};
{error, _} = Err -> Err
end;
_ ->
{error, transaction_unsupported}
end.
-spec hkeys(iodata()) -> {ok, [binary()]} | redis_error().
hkeys(Key) ->
case erlang:get(?TR_STACK) of
undefined ->
q([<<"HKEYS">>, Key]);
_ ->
{error, transaction_unsupported}
end.
%%%=================================================================== %%%===================================================================
%%% gen_server callbacks %%% gen_server callbacks
%%%=================================================================== %%%===================================================================
@ -325,6 +414,28 @@ get_result([{ok, _} = OK]) ->
get_result([_|T]) -> get_result([_|T]) ->
get_result(T). get_result(T).
-spec tr_enq([iodata()], list()) -> queued.
tr_enq(Cmd, Stack) ->
erlang:put(?TR_STACK, [Cmd|Stack]),
queued.
decode_pairs(Pairs) ->
decode_pairs(Pairs, []).
decode_pairs([Field, Val|Pairs], Acc) ->
decode_pairs(Pairs, [{Field, Val}|Acc]);
decode_pairs([], Acc) ->
lists:reverse(Acc).
dec_bool(<<$1>>) -> true;
dec_bool(<<$0>>) -> false.
reply(Val) ->
case erlang:get(?TR_STACK) of
undefined -> {ok, Val};
_ -> queued
end.
opt_type(redis_connect_timeout) -> opt_type(redis_connect_timeout) ->
fun (I) when is_integer(I), I > 0 -> I end; fun (I) when is_integer(I), I > 0 -> I end;
opt_type(redis_db) -> opt_type(redis_db) ->

75
src/ejabberd_router_redis.erl
View File

@ -44,9 +44,12 @@ register_route(Domain, ServerHost, LocalHint, _, Pid) ->
DomKey = domain_key(Domain), DomKey = domain_key(Domain),
PidKey = term_to_binary(Pid), PidKey = term_to_binary(Pid),
T = term_to_binary({ServerHost, LocalHint}), T = term_to_binary({ServerHost, LocalHint}),
case ejabberd_redis:qp([["HSET", DomKey, PidKey, T], case ejabberd_redis:multi(
["SADD", ?ROUTES_KEY, Domain]]) of fun() ->
[{ok, _}, {ok, _}] -> ejabberd_redis:hset(DomKey, PidKey, T),
ejabberd_redis:sadd(?ROUTES_KEY, [Domain])
end) of
{ok, _} ->
ok; ok;
Err -> Err ->
?ERROR_MSG("failed to register route in redis: ~p", [Err]), ?ERROR_MSG("failed to register route in redis: ~p", [Err]),
@ -57,13 +60,20 @@ unregister_route(Domain, _, Pid) ->
DomKey = domain_key(Domain), DomKey = domain_key(Domain),
PidKey = term_to_binary(Pid), PidKey = term_to_binary(Pid),
try try
{ok, _} = ejabberd_redis:q(["HDEL", DomKey, PidKey]), {ok, Num} = ejabberd_redis:hdel(DomKey, [PidKey]),
{ok, Num} = ejabberd_redis:q(["HLEN", DomKey]), if Num > 0 ->
case binary_to_integer(Num) of {ok, Len} = ejabberd_redis:hlen(DomKey),
0 -> if Len == 0 ->
{ok, _} = ejabberd_redis:q(["SREM", ?ROUTES_KEY, Domain]), {ok, _} = ejabberd_redis:multi(
ok; fun() ->
_ -> ejabberd_redis:del([DomKey]),
ejabberd_redis:srem(?ROUTES_KEY, [Domain])
end),
ok;
true ->
ok
end;
true ->
ok ok
end end
catch _:{badmatch, Err} -> catch _:{badmatch, Err} ->
@ -73,7 +83,7 @@ unregister_route(Domain, _, Pid) ->
find_routes(Domain) -> find_routes(Domain) ->
DomKey = domain_key(Domain), DomKey = domain_key(Domain),
case ejabberd_redis:q(["HGETALL", DomKey]) of case ejabberd_redis:hgetall(DomKey) of
{ok, Vals} -> {ok, Vals} ->
decode_routes(Domain, Vals); decode_routes(Domain, Vals);
Err -> Err ->
@ -83,8 +93,8 @@ find_routes(Domain) ->
host_of_route(Domain) -> host_of_route(Domain) ->
DomKey = domain_key(Domain), DomKey = domain_key(Domain),
case ejabberd_redis:q(["HGETALL", DomKey]) of case ejabberd_redis:hgetall(DomKey) of
{ok, [_, Data|_]} -> {ok, [{_Pid, Data}|_]} ->
{ServerHost, _} = binary_to_term(Data), {ServerHost, _} = binary_to_term(Data),
{ok, ServerHost}; {ok, ServerHost};
{ok, []} -> {ok, []} ->
@ -95,9 +105,9 @@ host_of_route(Domain) ->
end. end.
is_my_route(Domain) -> is_my_route(Domain) ->
case ejabberd_redis:q(["SISMEMBER", ?ROUTES_KEY, Domain]) of case ejabberd_redis:sismember(?ROUTES_KEY, Domain) of
{ok, <<"1">>} -> true; {ok, Bool} ->
{ok, _} -> false; Bool;
Err -> Err ->
?ERROR_MSG("failed to check route in redis: ~p", [Err]), ?ERROR_MSG("failed to check route in redis: ~p", [Err]),
false false
@ -107,7 +117,7 @@ is_my_host(Domain) ->
{ok, Domain} == host_of_route(Domain). {ok, Domain} == host_of_route(Domain).
get_all_routes() -> get_all_routes() ->
case ejabberd_redis:q(["SMEMBERS", ?ROUTES_KEY]) of case ejabberd_redis:smembers(?ROUTES_KEY) of
{ok, Routes} -> {ok, Routes} ->
Routes; Routes;
Err -> Err ->
@ -116,18 +126,7 @@ get_all_routes() ->
end. end.
find_routes() -> find_routes() ->
lists:flatmap( lists:flatmap(fun find_routes/1, get_all_routes()).
fun(Domain) ->
DomKey = domain_key(Domain),
case ejabberd_redis:q(["HGETALL", DomKey]) of
{ok, Vals} ->
decode_routes(Domain, Vals);
Err ->
?ERROR_MSG("failed to fetch routes from redis: ~p",
[Err]),
[]
end
end, get_all_routes()).
%%%=================================================================== %%%===================================================================
%%% Internal functions %%% Internal functions
@ -143,12 +142,12 @@ clean_table() ->
domain_key(Domain) -> domain_key(Domain) ->
<<"ejabberd:route:", Domain/binary>>. <<"ejabberd:route:", Domain/binary>>.
decode_routes(Domain, [Pid, Data|Vals]) -> decode_routes(Domain, Vals) ->
{ServerHost, LocalHint} = binary_to_term(Data), lists:map(
[#route{domain = Domain, fun({Pid, Data}) ->
pid = binary_to_term(Pid), {ServerHost, LocalHint} = binary_to_term(Data),
server_host = ServerHost, #route{domain = Domain,
local_hint = LocalHint}| pid = binary_to_term(Pid),
decode_routes(Domain, Vals)]; server_host = ServerHost,
decode_routes(_, []) -> local_hint = LocalHint}
[]. end, Vals).

103
src/ejabberd_sm_redis.erl
View File

@ -50,9 +50,12 @@ set_session(Session) ->
SIDKey = sid_to_key(Session#session.sid), SIDKey = sid_to_key(Session#session.sid),
ServKey = server_to_key(element(2, Session#session.us)), ServKey = server_to_key(element(2, Session#session.us)),
USSIDKey = us_sid_to_key(Session#session.us, Session#session.sid), USSIDKey = us_sid_to_key(Session#session.us, Session#session.sid),
case ejabberd_redis:qp([["HSET", USKey, SIDKey, T], case ejabberd_redis:multi(
["HSET", ServKey, USSIDKey, T]]) of fun() ->
[{ok, _}, {ok, _}] -> ejabberd_redis:hset(USKey, SIDKey, T),
ejabberd_redis:hset(ServKey, USSIDKey, T)
end) of
{ok, _} ->
ok; ok;
Err -> Err ->
?ERROR_MSG("failed to set session for redis: ~p", [Err]) ?ERROR_MSG("failed to set session for redis: ~p", [Err])
@ -62,7 +65,7 @@ set_session(Session) ->
{ok, #session{}} | {error, notfound}. {ok, #session{}} | {error, notfound}.
delete_session(LUser, LServer, _LResource, SID) -> delete_session(LUser, LServer, _LResource, SID) ->
USKey = us_to_key({LUser, LServer}), USKey = us_to_key({LUser, LServer}),
case ejabberd_redis:q(["HGETALL", USKey]) of case ejabberd_redis:hgetall(USKey) of
{ok, Vals} -> {ok, Vals} ->
Ss = decode_session_list(Vals), Ss = decode_session_list(Vals),
case lists:keyfind(SID, #session.sid, Ss) of case lists:keyfind(SID, #session.sid, Ss) of
@ -72,8 +75,16 @@ delete_session(LUser, LServer, _LResource, SID) ->
SIDKey = sid_to_key(SID), SIDKey = sid_to_key(SID),
ServKey = server_to_key(element(2, Session#session.us)), ServKey = server_to_key(element(2, Session#session.us)),
USSIDKey = us_sid_to_key(Session#session.us, SID), USSIDKey = us_sid_to_key(Session#session.us, SID),
ejabberd_redis:qp([["HDEL", USKey, SIDKey], case ejabberd_redis:multi(
["HDEL", ServKey, USSIDKey]]), fun() ->
ejabberd_redis:hdel(USKey, [SIDKey]),
ejabberd_redis:hdel(ServKey, [USSIDKey])
end) of
{ok, _} ->
ok;
Err ->
?ERROR_MSG("failed to delete session from redis: ~p", [Err])
end,
{ok, Session} {ok, Session}
end; end;
Err -> Err ->
@ -91,7 +102,7 @@ get_sessions() ->
-spec get_sessions(binary()) -> [#session{}]. -spec get_sessions(binary()) -> [#session{}].
get_sessions(LServer) -> get_sessions(LServer) ->
ServKey = server_to_key(LServer), ServKey = server_to_key(LServer),
case ejabberd_redis:q(["HGETALL", ServKey]) of case ejabberd_redis:hgetall(ServKey) of
{ok, Vals} -> {ok, Vals} ->
decode_session_list(Vals); decode_session_list(Vals);
Err -> Err ->
@ -102,8 +113,8 @@ get_sessions(LServer) ->
-spec get_sessions(binary(), binary()) -> [#session{}]. -spec get_sessions(binary(), binary()) -> [#session{}].
get_sessions(LUser, LServer) -> get_sessions(LUser, LServer) ->
USKey = us_to_key({LUser, LServer}), USKey = us_to_key({LUser, LServer}),
case ejabberd_redis:q(["HGETALL", USKey]) of case ejabberd_redis:hgetall(USKey) of
{ok, Vals} when is_list(Vals) -> {ok, Vals} ->
decode_session_list(Vals); decode_session_list(Vals);
Err -> Err ->
?ERROR_MSG("failed to get sessions from redis: ~p", [Err]), ?ERROR_MSG("failed to get sessions from redis: ~p", [Err]),
@ -114,8 +125,8 @@ get_sessions(LUser, LServer) ->
[#session{}]. [#session{}].
get_sessions(LUser, LServer, LResource) -> get_sessions(LUser, LServer, LResource) ->
USKey = us_to_key({LUser, LServer}), USKey = us_to_key({LUser, LServer}),
case ejabberd_redis:q(["HGETALL", USKey]) of case ejabberd_redis:hgetall(USKey) of
{ok, Vals} when is_list(Vals) -> {ok, Vals} ->
[S || S <- decode_session_list(Vals), [S || S <- decode_session_list(Vals),
element(3, S#session.usr) == LResource]; element(3, S#session.usr) == LResource];
Err -> Err ->
@ -141,52 +152,36 @@ us_sid_to_key(US, SID) ->
sid_to_key(SID) -> sid_to_key(SID) ->
term_to_binary(SID). term_to_binary(SID).
decode_session_list([_, Val|T]) -> decode_session_list(Vals) ->
[binary_to_term(Val)|decode_session_list(T)]; [binary_to_term(Val) || {_, Val} <- Vals].
decode_session_list([]) ->
[].
clean_table() -> clean_table() ->
?INFO_MSG("Cleaning Redis SM table...", []), ?INFO_MSG("Cleaning Redis SM table...", []),
lists:foreach( try
fun(LServer) -> lists:foreach(
ServKey = server_to_key(LServer), fun(LServer) ->
case ejabberd_redis:q(["HKEYS", ServKey]) of ServKey = server_to_key(LServer),
{ok, []} -> {ok, Vals} = ejabberd_redis:hkeys(ServKey),
ok; {ok, _} =
{ok, Vals} -> ejabberd_redis:multi(
Vals1 = lists:filter( fun() ->
fun(USSIDKey) -> lists:foreach(
{_, SID} = binary_to_term(USSIDKey), fun(USSIDKey) ->
node(element(2, SID)) == node() {US, SID} = binary_to_term(USSIDKey),
end, Vals), if node(element(2, SID)) == node() ->
Q1 = case Vals1 of USKey = us_to_key(US),
[] -> []; SIDKey = sid_to_key(SID),
_ -> ["HDEL", ServKey | Vals1] ejabberd_redis:hdel(ServKey, [USSIDKey]),
end, ejabberd_redis:hdel(USKey, [SIDKey]);
Q2 = lists:map( true ->
fun(USSIDKey) -> ok
{US, SID} = binary_to_term(USSIDKey), end
USKey = us_to_key(US), end, Vals)
SIDKey = sid_to_key(SID), end)
["HDEL", USKey, SIDKey] end, ejabberd_sm:get_vh_by_backend(?MODULE))
end, Vals1), catch _:{badmatch, {error, _} = Err} ->
Res = ejabberd_redis:qp(lists:delete([], [Q1|Q2])), ?ERROR_MSG("failed to clean redis c2s sessions: ~p", [Err])
case lists:filter( end.
fun({ok, _}) -> false;
(_) -> true
end, Res) of
[] ->
ok;
Errs ->
?ERROR_MSG("failed to clean redis table for "
"server ~s: ~p", [LServer, Errs])
end;
Err ->
?ERROR_MSG("failed to clean redis table for "
"server ~s: ~p", [LServer, Err])
end
end, ejabberd_sm:get_vh_by_backend(?MODULE)).
opt_type(redis_connect_timeout) -> opt_type(redis_connect_timeout) ->
fun (I) when is_integer(I), I > 0 -> I end; fun (I) when is_integer(I), I > 0 -> I end;

46
src/mod_bosh_redis.erl
View File

@ -25,7 +25,7 @@ init() ->
open_session(SID, Pid) -> open_session(SID, Pid) ->
PidBin = term_to_binary(Pid), PidBin = term_to_binary(Pid),
case ejabberd_redis:q(["HSET", ?BOSH_KEY, SID, PidBin]) of case ejabberd_redis:hset(?BOSH_KEY, SID, PidBin) of
{ok, _} -> {ok, _} ->
ok; ok;
Err -> Err ->
@ -34,7 +34,7 @@ open_session(SID, Pid) ->
end. end.
close_session(SID) -> close_session(SID) ->
case ejabberd_redis:q(["HDEL", ?BOSH_KEY, SID]) of case ejabberd_redis:hdel(?BOSH_KEY, [SID]) of
{ok, _} -> {ok, _} ->
ok; ok;
Err -> Err ->
@ -42,9 +42,15 @@ close_session(SID) ->
end. end.
find_session(SID) -> find_session(SID) ->
case ejabberd_redis:q(["HGET", ?BOSH_KEY, SID]) of case ejabberd_redis:hget(?BOSH_KEY, SID) of
{ok, Pid} when is_binary(Pid) -> {ok, Pid} when is_binary(Pid) ->
{ok, binary_to_term(Pid)}; try
{ok, binary_to_term(Pid)}
catch _:badarg ->
?ERROR_MSG("malformed data in redis (key = '~s'): ~p",
[SID, Pid]),
error
end;
{ok, _} -> {ok, _} ->
error; error;
Err -> Err ->
@ -56,21 +62,23 @@ find_session(SID) ->
%%% Internal functions %%% Internal functions
%%%=================================================================== %%%===================================================================
clean_table() -> clean_table() ->
?INFO_MSG("Cleaning Redis BOSH table...", []), ?INFO_MSG("Cleaning Redis BOSH sessions...", []),
case ejabberd_redis:q(["HGETALL", ?BOSH_KEY]) of case ejabberd_redis:hgetall(?BOSH_KEY) of
{ok, Vals} -> {ok, Vals} ->
clean_table(Vals); case ejabberd_redis:multi(
fun() ->
lists:foreach(
fun({SID, Pid}) when node(Pid) == node() ->
ejabberd_redis:hdel(?BOSH_KEY, [SID]);
(_) ->
ok
end, Vals)
end) of
{ok, _} ->
ok;
Err ->
?ERROR_MSG("failed to clean bosh sessions in redis: ~p", [Err])
end;
Err -> Err ->
?ERROR_MSG("failed to clean bosh table in redis: ~p", [Err]) ?ERROR_MSG("failed to clean bosh sessions in redis: ~p", [Err])
end. end.
clean_table([SID, PidBin|Vals]) ->
case binary_to_term(PidBin) of
Pid when node(Pid) == node() ->
close_session(SID);
_ ->
ok
end,
clean_table(Vals);
clean_table([]) ->
ok.

57
src/mod_carboncopy_redis.erl
View File

@ -39,9 +39,12 @@ enable(LUser, LServer, LResource, NS) ->
USKey = us_key(LUser, LServer), USKey = us_key(LUser, LServer),
NodeKey = node_key(), NodeKey = node_key(),
JID = jid:encode({LUser, LServer, LResource}), JID = jid:encode({LUser, LServer, LResource}),
case ejabberd_redis:qp([["HSET", USKey, LResource, NS], case ejabberd_redis:multi(
["SADD", NodeKey, JID]]) of fun() ->
[{ok, _}, {ok, _}] -> ejabberd_redis:hset(USKey, LResource, NS),
ejabberd_redis:sadd(NodeKey, [JID])
end) of
{ok, _} ->
ok; ok;
Err -> Err ->
?ERROR_MSG("failed to write in redis: ~p", [Err]), ?ERROR_MSG("failed to write in redis: ~p", [Err]),
@ -52,9 +55,12 @@ disable(LUser, LServer, LResource) ->
USKey = us_key(LUser, LServer), USKey = us_key(LUser, LServer),
NodeKey = node_key(), NodeKey = node_key(),
JID = jid:encode({LUser, LServer, LResource}), JID = jid:encode({LUser, LServer, LResource}),
case ejabberd_redis:qp([["HDEL", USKey, LResource], case ejabberd_redis:multi(
["SREM", NodeKey, JID]]) of fun() ->
[{ok, _}, {ok, _}] -> ejabberd_redis:hdel(USKey, [LResource]),
ejabberd_redis:srem(NodeKey, [JID])
end) of
{ok, _} ->
ok; ok;
Err -> Err ->
?ERROR_MSG("failed to delete from redis: ~p", [Err]), ?ERROR_MSG("failed to delete from redis: ~p", [Err]),
@ -63,9 +69,9 @@ disable(LUser, LServer, LResource) ->
list(LUser, LServer) -> list(LUser, LServer) ->
USKey = us_key(LUser, LServer), USKey = us_key(LUser, LServer),
case ejabberd_redis:q(["HGETALL", USKey]) of case ejabberd_redis:hgetall(USKey) of
{ok, Vals} -> {ok, Vals} ->
decode_vals(Vals); Vals;
Err -> Err ->
?ERROR_MSG("failed to read from redis: ~p", [Err]), ?ERROR_MSG("failed to read from redis: ~p", [Err]),
[] []
@ -77,24 +83,26 @@ list(LUser, LServer) ->
clean_table() -> clean_table() ->
?INFO_MSG("Cleaning Redis 'carboncopy' table...", []), ?INFO_MSG("Cleaning Redis 'carboncopy' table...", []),
NodeKey = node_key(), NodeKey = node_key(),
case ejabberd_redis:q(["SMEMBERS", NodeKey]) of case ejabberd_redis:smembers(NodeKey) of
{ok, JIDs} -> {ok, JIDs} ->
lists:foreach( case ejabberd_redis:multi(
fun(JID) -> fun() ->
{U, S, R} = jid:split(jid:decode(JID)), lists:foreach(
USKey = us_key(U, S), fun(JID) ->
case ejabberd_redis:q(["HDEL", USKey, R]) of {U, S, R} = jid:split(jid:decode(JID)),
{ok, _} -> USKey = us_key(U, S),
ok; ejabberd_redis:hdel(USKey, [R])
Err -> end, JIDs)
?ERROR_MSG("failed to delete from redis: ~p", end) of
[Err]) {ok, _} ->
end ok;
end, JIDs); Err ->
?ERROR_MSG("failed to delete from redis: ~p", [Err])
end;
Err -> Err ->
?ERROR_MSG("failed to read from redis: ~p", [Err]) ?ERROR_MSG("failed to read from redis: ~p", [Err])
end, end,
case ejabberd_redis:q(["DEL", NodeKey]) of case ejabberd_redis:del([NodeKey]) of
{ok, _} -> ok; {ok, _} -> ok;
Error -> ?ERROR_MSG("failed to delete from redis: ~p", [Error]) Error -> ?ERROR_MSG("failed to delete from redis: ~p", [Error])
end. end.
@ -105,8 +113,3 @@ us_key(LUser, LServer) ->
node_key() -> node_key() ->
Node = erlang:atom_to_binary(node(), latin1), Node = erlang:atom_to_binary(node(), latin1),
<<"ejabberd:carboncopy:nodes:", Node/binary>>. <<"ejabberd:carboncopy:nodes:", Node/binary>>.
decode_vals([Resource, NS|Vals]) ->
[{Resource, NS}|decode_vals(Vals)];
decode_vals([]) ->
[].