%%%------------------------------------------------------------------- %%% @author Evgeny Khramtsov %%% Created : 30 Mar 2017 by Evgeny Khramtsov %%% %%% %%% ejabberd, Copyright (C) 2002-2018 ProcessOne %%% %%% This program is free software; you can redistribute it and/or %%% modify it under the terms of the GNU General Public License as %%% published by the Free Software Foundation; either version 2 of the %%% License, or (at your option) any later version. %%% %%% This program is distributed in the hope that it will be useful, %%% but WITHOUT ANY WARRANTY; without even the implied warranty of %%% MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU %%% General Public License for more details. %%% %%% You should have received a copy of the GNU General Public License along %%% with this program; if not, write to the Free Software Foundation, Inc., %%% 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. %%% %%%------------------------------------------------------------------- -module(mod_carboncopy_redis). -behaviour(mod_carboncopy). -behaviour(gen_server). %% API -export([init/2, enable/4, disable/3, list/2, cache_nodes/1]). %% gen_server callbacks -export([init/1, handle_cast/2, handle_call/3, handle_info/2, terminate/2, code_change/3, start_link/0]). -include("logger.hrl"). -include("mod_carboncopy.hrl"). -define(CARBONCOPY_KEY, <<"ejabberd:carboncopy">>). -record(state, {}). %%%=================================================================== %%% API %%%=================================================================== init(_Host, _Opts) -> Spec = {?MODULE, {?MODULE, start_link, []}, transient, 5000, worker, [?MODULE]}, case supervisor:start_child(ejabberd_backend_sup, Spec) of {ok, _Pid} -> ok; Err -> Err end. -spec start_link() -> {ok, pid()} | {error, any()}. start_link() -> gen_server:start_link({local, ?MODULE}, ?MODULE, [], []). cache_nodes(_LServer) -> [node()]. enable(LUser, LServer, LResource, NS) -> USKey = us_key(LUser, LServer), NodeKey = node_key(), JID = jid:encode({LUser, LServer, LResource}), Data = term_to_binary({NS, node()}), case ejabberd_redis:multi( fun() -> ejabberd_redis:hset(USKey, LResource, Data), ejabberd_redis:sadd(NodeKey, [JID]), ejabberd_redis:publish( ?CARBONCOPY_KEY, term_to_binary({delete, {LUser, LServer}})) end) of {ok, _} -> ok; {error, _} -> {error, db_failure} end. disable(LUser, LServer, LResource) -> USKey = us_key(LUser, LServer), NodeKey = node_key(), JID = jid:encode({LUser, LServer, LResource}), case ejabberd_redis:multi( fun() -> ejabberd_redis:hdel(USKey, [LResource]), ejabberd_redis:srem(NodeKey, [JID]), ejabberd_redis:publish( ?CARBONCOPY_KEY, term_to_binary({delete, {LUser, LServer}})) end) of {ok, _} -> ok; {error, _} -> {error, db_failure} end. list(LUser, LServer) -> USKey = us_key(LUser, LServer), case ejabberd_redis:hgetall(USKey) of {ok, Pairs} -> {ok, lists:flatmap( fun({Resource, Data}) -> try {NS, Node} = binary_to_term(Data), [{Resource, NS, Node}] catch _:_ -> ?ERROR_MSG("invalid term stored in Redis " "(key = ~s): ~p", [USKey, Data]), [] end end, Pairs)}; {error, _} -> {error, db_failure} end. %%%=================================================================== %%% gen_server callbacks %%%=================================================================== init([]) -> ejabberd_redis:subscribe([?CARBONCOPY_KEY]), clean_table(), {ok, #state{}}. handle_call(_Request, _From, State) -> Reply = ok, {reply, Reply, State}. handle_cast(_Msg, State) -> {noreply, State}. handle_info({redis_message, ?CARBONCOPY_KEY, Data}, State) -> case binary_to_term(Data) of {delete, Key} -> ets_cache:delete(?CARBONCOPY_CACHE, Key); Msg -> ?WARNING_MSG("unexpected redis message: ~p", [Msg]) end, {noreply, State}; handle_info(Info, State) -> ?ERROR_MSG("unexpected info: ~p", [Info]), {noreply, State}. terminate(_Reason, _State) -> ok. code_change(_OldVsn, State, _Extra) -> {ok, State}. %%%=================================================================== %%% Internal functions %%%=================================================================== clean_table() -> ?DEBUG("Cleaning Redis 'carboncopy' table...", []), NodeKey = node_key(), case ejabberd_redis:smembers(NodeKey) of {ok, JIDs} -> ejabberd_redis:multi( fun() -> lists:foreach( fun(JID) -> {U, S, R} = jid:split(jid:decode(JID)), USKey = us_key(U, S), ejabberd_redis:hdel(USKey, [R]) end, JIDs) end); {error, _} -> ok end, ejabberd_redis:del([NodeKey]), ok. us_key(LUser, LServer) -> <<"ejabberd:carboncopy:users:", LUser/binary, $@, LServer/binary>>. node_key() -> Node = erlang:atom_to_binary(node(), latin1), <<"ejabberd:carboncopy:nodes:", Node/binary>>.