Use cache in front of Redis/SQL RAM backends

1 files changed, 80 insertions, 8 deletions

diff --git a/src/mod_carboncopy_redis.erl b/src/mod_carboncopy_redis.erl
index 8ed33468..b72755f4 100644
--- a/src/mod_carboncopy_redis.erl
+++ b/src/mod_carboncopy_redis.erl
@@ -22,27 +22,52 @@
 %%%-------------------------------------------------------------------
 -module(mod_carboncopy_redis).
 -behaviour(mod_carboncopy).
+-behaviour(gen_server).
 
 %% API
--export([init/2, enable/4, disable/3, list/2]).
+-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("ejabberd.hrl").
 -include("logger.hrl").
+-include("mod_carboncopy.hrl").
+
+-define(CARBONCOPY_KEY, <<"ejabberd:carboncopy">>).
+
+-record(state, {}).
 
 %%%===================================================================
 %%% API
 %%%===================================================================
 init(_Host, _Opts) ->
-    clean_table().
+    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, NS),
-		   ejabberd_redis:sadd(NodeKey, [JID])
+		   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;
@@ -57,7 +82,10 @@ disable(LUser, LServer, LResource) ->
     case ejabberd_redis:multi(
 	   fun() ->
 		   ejabberd_redis:hdel(USKey, [LResource]),
-		   ejabberd_redis:srem(NodeKey, [JID])
+		   ejabberd_redis:srem(NodeKey, [JID]),
+		   ejabberd_redis:publish(
+		     ?CARBONCOPY_KEY,
+		     term_to_binary({delete, {LUser, LServer}}))
 	   end) of
 	{ok, _} ->
 	    ok;
@@ -68,13 +96,57 @@ disable(LUser, LServer, LResource) ->
 list(LUser, LServer) ->
     USKey = us_key(LUser, LServer),
     case ejabberd_redis:hgetall(USKey) of
-	{ok, Vals} ->
-	    Vals;
+	{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() ->