Packages
ejabberd
17.6.0
26.4.0
26.3.0
26.2.0
26.1.0
25.10.0
25.8.0
25.7.0
25.4.0
25.3.0
24.12.0
24.10.0
24.7.0
24.6.0
24.2.6
23.10.0
23.4.0
23.1.0
22.10.0
22.5.0
21.12.0
21.7.0
21.4.0
21.1.0
20.12.0
20.7.0
20.4.0
20.3.0
20.2.0
20.1.0
19.9.1
19.9.0
19.8.0
19.5.0
19.2.0
18.12.1
18.12.0
18.6.0
18.4.0
18.3.0
18.1.0
17.11.0
17.9.0
17.6.0
17.3.0
17.1.0
16.12.0-beta1
16.9.0
16.8.0
16.6.2
16.6.1
16.6.0
16.4.1
16.4.0
16.3.0
16.2.0
16.1.0-beta1
Robust, Ubiquitous and Massively Scalable Messaging Platform (XMPP, MQTT, SIP Server)
Current section
Files
Jump to
Current section
Files
src/mod_bosh_redis.erl
%%%-------------------------------------------------------------------
%%% @author Evgeny Khramtsov <ekhramtsov@process-one.net>
%%% @copyright (C) 2017, Evgeny Khramtsov
%%% @doc
%%%
%%% @end
%%% Created : 28 Mar 2017 by Evgeny Khramtsov <ekhramtsov@process-one.net>
%%%-------------------------------------------------------------------
-module(mod_bosh_redis).
-behaviour(mod_bosh).
-behaviour(gen_server).
%% API
-export([init/0, open_session/2, close_session/1, find_session/1,
cache_nodes/0]).
%% 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("bosh.hrl").
-record(state, {}).
-define(BOSH_KEY, <<"ejabberd:bosh">>).
%%%===================================================================
%%% API
%%%===================================================================
init() ->
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, [], []).
open_session(SID, Pid) ->
PidBin = term_to_binary(Pid),
case ejabberd_redis:multi(
fun() ->
ejabberd_redis:hset(?BOSH_KEY, SID, PidBin),
ejabberd_redis:publish(?BOSH_KEY, SID)
end) of
{ok, _} ->
ok;
{error, _} ->
{error, db_failure}
end.
close_session(SID) ->
case ejabberd_redis:multi(
fun() ->
ejabberd_redis:hdel(?BOSH_KEY, [SID]),
ejabberd_redis:publish(?BOSH_KEY, SID)
end) of
{ok, _} ->
ok;
{error, _} ->
{error, db_failure}
end.
find_session(SID) ->
case ejabberd_redis:hget(?BOSH_KEY, SID) of
{ok, undefined} ->
{error, notfound};
{ok, Pid} ->
try
{ok, binary_to_term(Pid)}
catch _:badarg ->
?ERROR_MSG("malformed data in redis (key = '~s'): ~p",
[SID, Pid]),
{error, db_failure}
end;
{error, _} ->
{error, db_failure}
end.
cache_nodes() ->
[node()].
%%%===================================================================
%%% gen_server callbacks
%%%===================================================================
init([]) ->
clean_table(),
{ok, #state{}}.
handle_call(_Request, _From, State) ->
Reply = ok,
{reply, Reply, State}.
handle_cast(_Msg, State) ->
{noreply, State}.
handle_info({redis_message, ?BOSH_KEY, SID}, State) ->
ets_cache:delete(?BOSH_CACHE, SID),
{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 BOSH sessions...", []),
case ejabberd_redis:hgetall(?BOSH_KEY) of
{ok, Vals} ->
ejabberd_redis:multi(
fun() ->
lists:foreach(
fun({SID, Pid}) when node(Pid) == node() ->
ejabberd_redis:hdel(?BOSH_KEY, [SID]);
(_) ->
ok
end, Vals)
end),
ok;
{error, _} ->
?ERROR_MSG("failed to clean bosh sessions in redis", [])
end.