diff options
| -rw-r--r-- | ejabberd.yml.example | 1 | ||||
| -rw-r--r-- | src/mod_push.erl | 8 | ||||
| -rw-r--r-- | src/mod_push_keepalive.erl | 236 | ||||
| -rw-r--r-- | src/mod_stream_mgmt.erl | 54 | ||||
| -rw-r--r-- | test/ejabberd_SUITE_data/ejabberd.yml | 2 | ||||
| -rw-r--r-- | test/push_tests.erl | 2 |
6 files changed, 290 insertions, 13 deletions
diff --git a/ejabberd.yml.example b/ejabberd.yml.example index 693a87f57..922400d2d 100644 --- a/ejabberd.yml.example +++ b/ejabberd.yml.example @@ -726,6 +726,7 @@ modules: - "hometree" - "pep" # pep requires mod_caps mod_push: {} + mod_push_keepalive: {} ## mod_register: ## ## Protect In-Band account registrations with CAPTCHA. diff --git a/src/mod_push.erl b/src/mod_push.erl index f82226440..55db7c0e9 100644 --- a/src/mod_push.erl +++ b/src/mod_push.erl @@ -43,6 +43,9 @@ %% ejabberd command. -export([get_commands_spec/0, delete_old_sessions/1]). +%% API (used by mod_push_keepalive). +-export([notify/1, notify/3, notify/5]). + -include("ejabberd.hrl"). -include("ejabberd_commands.hrl"). -include("logger.hrl"). @@ -393,7 +396,7 @@ remove_user(LUser, LServer) -> delete_sessions(LUser, LServer, LookupFun, Mod). %%-------------------------------------------------------------------- -%% Internal functions. +%% Generate push notifications. %%-------------------------------------------------------------------- -spec notify(c2s_state()) -> ok. notify(#{jid := #jid{luser = LUser, lserver = LServer}, sid := {TS, _}}) -> @@ -433,6 +436,9 @@ notify(LServer, PushLJID, Node, XData, HandleResponse) -> ejabberd_local:route_iq(IQ, HandleResponse), ok. +%%-------------------------------------------------------------------- +%% Internal functions. +%%-------------------------------------------------------------------- -spec store_session(binary(), binary(), timestamp(), jid(), binary(), xdata()) -> {ok, push_session()} | error. store_session(LUser, LServer, TS, PushJID, Node, XData) -> diff --git a/src/mod_push_keepalive.erl b/src/mod_push_keepalive.erl new file mode 100644 index 000000000..bde62fc67 --- /dev/null +++ b/src/mod_push_keepalive.erl @@ -0,0 +1,236 @@ +%%%---------------------------------------------------------------------- +%%% File : mod_push_keepalive.erl +%%% Author : Holger Weiss <holger@zedat.fu-berlin.de> +%%% Purpose : Keep pending XEP-0198 sessions alive with XEP-0357 +%%% Created : 15 Jul 2017 by Holger Weiss <holger@zedat.fu-berlin.de> +%%% +%%% +%%% ejabberd, Copyright (C) 2017 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_push_keepalive). +-author('holger@zedat.fu-berlin.de'). + +-behavior(gen_mod). + +%% gen_mod callbacks. +-export([start/2, stop/1, reload/3, mod_opt_type/1, depends/2]). + +%% ejabberd_hooks callbacks. +-export([c2s_session_pending/1, c2s_session_resumed/1, c2s_copy_session/2, + c2s_handle_cast/2, c2s_handle_info/2, c2s_stanza/3]). + +-include("logger.hrl"). +-include("xmpp.hrl"). + +-define(PUSH_BEFORE_TIMEOUT_SECS, 120). + +-type c2s_state() :: ejabberd_c2s:state(). + +%%-------------------------------------------------------------------- +%% gen_mod callbacks. +%%-------------------------------------------------------------------- +-spec start(binary(), gen_mod:opts()) -> ok. +start(Host, Opts) -> + case gen_mod:get_opt(wake_on_start, Opts, false) of + true -> + wake_all(Host); + false -> + ok + end, + register_hooks(Host). + +-spec stop(binary()) -> ok. +stop(Host) -> + unregister_hooks(Host). + +-spec reload(binary(), gen_mod:opts(), gen_mod:opts()) -> ok. +reload(Host, NewOpts, OldOpts) -> + case gen_mod:is_equal_opt(wake_on_start, NewOpts, OldOpts, false) of + {false, true, _} -> + wake_all(Host); + _ -> + ok + end, + ok. + +-spec depends(binary(), gen_mod:opts()) -> [{module(), hard | soft}]. +depends(_Host, _Opts) -> + [{mod_push, hard}, + {mod_client_state, soft}, + {mod_stream_mgmt, soft}]. + +-spec mod_opt_type(atom()) -> fun((term()) -> term()) | [atom()]. +mod_opt_type(resume_timeout) -> + fun(I) when is_integer(I), I >= 0 -> I; + (undefined) -> undefined + end; +mod_opt_type(wake_on_start) -> + fun (B) when is_boolean(B) -> B end; +mod_opt_type(wake_on_timeout) -> + fun (B) when is_boolean(B) -> B end; +mod_opt_type(O) when O == cache_life_time; O == cache_size -> + fun(I) when is_integer(I), I > 0 -> I; + (infinity) -> infinity + end; +mod_opt_type(O) when O == use_cache; O == cache_missed -> + fun (B) when is_boolean(B) -> B end; +mod_opt_type(_) -> + [resume_timeout, wake_on_start, wake_on_timeout, db_type, cache_life_time, + cache_size, use_cache, cache_missed, iqdisc]. + +%%-------------------------------------------------------------------- +%% Register/unregister hooks. +%%-------------------------------------------------------------------- +-spec register_hooks(binary()) -> ok. +register_hooks(Host) -> + ejabberd_hooks:add(c2s_session_pending, Host, ?MODULE, + c2s_session_pending, 50), + ejabberd_hooks:add(c2s_session_resumed, Host, ?MODULE, + c2s_session_resumed, 50), + ejabberd_hooks:add(c2s_copy_session, Host, ?MODULE, + c2s_copy_session, 50), + ejabberd_hooks:add(c2s_handle_cast, Host, ?MODULE, + c2s_handle_cast, 40), + ejabberd_hooks:add(c2s_handle_info, Host, ?MODULE, + c2s_handle_info, 50), + ejabberd_hooks:add(c2s_handle_send, Host, ?MODULE, + c2s_stanza, 50). + +-spec unregister_hooks(binary()) -> ok. +unregister_hooks(Host) -> + ejabberd_hooks:delete(disco_sm_features, Host, ?MODULE, + disco_sm_features, 50), + ejabberd_hooks:delete(c2s_session_pending, Host, ?MODULE, + c2s_session_pending, 50), + ejabberd_hooks:delete(c2s_session_resumed, Host, ?MODULE, + c2s_session_resumed, 50), + ejabberd_hooks:delete(c2s_copy_session, Host, ?MODULE, + c2s_copy_session, 50), + ejabberd_hooks:delete(c2s_handle_cast, Host, ?MODULE, + c2s_handle_cast, 40), + ejabberd_hooks:delete(c2s_handle_info, Host, ?MODULE, + c2s_handle_info, 50), + ejabberd_hooks:delete(c2s_handle_send, Host, ?MODULE, + c2s_stanza, 50). + +%%-------------------------------------------------------------------- +%% Hook callbacks. +%%-------------------------------------------------------------------- +-spec c2s_stanza(c2s_state(), xmpp_element() | xmlel(), term()) -> c2s_state(). +c2s_stanza(#{push_enabled := true, mgmt_state := pending} = State, + _Pkt, _SendResult) -> + maybe_restore_resume_timeout(State); +c2s_stanza(State, _Pkt, _SendResult) -> + State. + +-spec c2s_session_pending(c2s_state()) -> c2s_state(). +c2s_session_pending(#{push_enabled := true, mgmt_queue := Queue} = State) -> + case p1_queue:len(Queue) of + 0 -> + State1 = maybe_adjust_resume_timeout(State), + maybe_start_wakeup_timer(State1); + _ -> + State + end; +c2s_session_pending(State) -> + State. + +-spec c2s_session_resumed(c2s_state()) -> c2s_state(). +c2s_session_resumed(#{push_enabled := true} = State) -> + maybe_restore_resume_timeout(State); +c2s_session_resumed(State) -> + State. + +-spec c2s_copy_session(c2s_state(), c2s_state()) -> c2s_state(). +c2s_copy_session(State, #{push_enabled := true, + push_resume_timeout := ResumeTimeout, + push_wake_on_timeout := WakeOnTimeout}) -> + State#{push_resume_timeout => ResumeTimeout, + push_wake_on_timeout => WakeOnTimeout}; +c2s_copy_session(State, _) -> + State. + +-spec c2s_handle_cast(c2s_state(), any()) -> c2s_state(). +c2s_handle_cast(#{lserver := LServer} = State, push_enable) -> + ResumeTimeout = gen_mod:get_module_opt(LServer, ?MODULE, + resume_timeout, 86400), + WakeOnTimeout = gen_mod:get_module_opt(LServer, ?MODULE, + wake_on_timeout, true), + State#{push_resume_timeout => ResumeTimeout, + push_wake_on_timeout => WakeOnTimeout}; +c2s_handle_cast(State, push_disable) -> + State1 = maps:remove(push_resume_timeout, State), + maps:remove(push_wake_on_timeout, State1); +c2s_handle_cast(State, _Msg) -> + State. + +-spec c2s_handle_info(c2s_state(), any()) -> c2s_state() | {stop, c2s_state()}. +c2s_handle_info(#{push_enabled := true, mgmt_state := pending, + jid := JID} = State, {timeout, _, push_keepalive}) -> + ?INFO_MSG("Waking ~s before session times out", [jid:encode(JID)]), + mod_push:notify(State), + {stop, State}; +c2s_handle_info(State, _) -> + State. + +%%-------------------------------------------------------------------- +%% Internal functions. +%%-------------------------------------------------------------------- +-spec maybe_adjust_resume_timeout(c2s_state()) -> c2s_state(). +maybe_adjust_resume_timeout(#{push_resume_timeout := undefined} = State) -> + State; +maybe_adjust_resume_timeout(#{push_resume_timeout := Timeout} = State) -> + OrigTimeout = mod_stream_mgmt:get_resume_timeout(State), + ?DEBUG("Adjusting resume timeout to ~B seconds", [Timeout]), + State1 = mod_stream_mgmt:set_resume_timeout(State, Timeout), + State1#{push_resume_timeout_orig => OrigTimeout}. + +-spec maybe_restore_resume_timeout(c2s_state()) -> c2s_state(). +maybe_restore_resume_timeout(#{push_resume_timeout_orig := Timeout} = State) -> + ?DEBUG("Restoring resume timeout to ~B seconds", [Timeout]), + State1 = mod_stream_mgmt:set_resume_timeout(State, Timeout), + maps:remove(push_resume_timeout_orig, State1); +maybe_restore_resume_timeout(State) -> + State. + +-spec maybe_start_wakeup_timer(c2s_state()) -> c2s_state(). +maybe_start_wakeup_timer(#{push_wake_on_timeout := true, + push_resume_timeout := ResumeTimeout} = State) + when is_integer(ResumeTimeout), ResumeTimeout > ?PUSH_BEFORE_TIMEOUT_SECS -> + WakeTimeout = ResumeTimeout - ?PUSH_BEFORE_TIMEOUT_SECS, + ?DEBUG("Scheduling wake-up timer to fire in ~B seconds", [WakeTimeout]), + erlang:start_timer(timer:seconds(WakeTimeout), self(), push_keepalive), + State; +maybe_start_wakeup_timer(State) -> + State. + +-spec wake_all(binary()) -> ok | error. +wake_all(LServer) -> + ?INFO_MSG("Waking all push clients on ~s", [LServer]), + Mod = gen_mod:db_mod(LServer, mod_push), + case Mod:lookup_sessions(LServer) of + {ok, Sessions} -> + IgnoreResponse = fun(_) -> ok end, + lists:foreach(fun({_, PushLJID, Node, XData}) -> + mod_push:notify(LServer, PushLJID, Node, + XData, IgnoreResponse) + end, Sessions); + error -> + error + end. diff --git a/src/mod_stream_mgmt.erl b/src/mod_stream_mgmt.erl index 127eea3e8..068550906 100644 --- a/src/mod_stream_mgmt.erl +++ b/src/mod_stream_mgmt.erl @@ -33,6 +33,8 @@ c2s_unbinded_packet/2, c2s_closed/2, c2s_terminated/2, c2s_handle_send/3, c2s_handle_info/2, c2s_handle_call/3, c2s_handle_recv/3]). +%% adjust pending session timeout +-export([get_resume_timeout/1, set_resume_timeout/2]). -include("xmpp.hrl"). -include("logger.hrl"). @@ -235,8 +237,9 @@ c2s_handle_info(#{mgmt_ack_timer := TRef, jid := JID, mod := Mod} = State, [jid:encode(JID)]), State1 = Mod:close(State), {stop, transition_to_pending(State1)}; -c2s_handle_info(#{mgmt_state := pending, jid := JID, mod := Mod} = State, - {timeout, _, pending_timeout}) -> +c2s_handle_info(#{mgmt_state := pending, + mgmt_pending_timer := TRef, jid := JID, mod := Mod} = State, + {timeout, TRef, pending_timeout}) -> ?DEBUG("Timed out waiting for resumption of stream for ~s", [jid:encode(JID)]), Mod:stop(State#{mgmt_state => timeout}); @@ -283,6 +286,20 @@ c2s_terminated(State, _Reason) -> State. %%%=================================================================== +%%% Adjust pending session timeout +%%%=================================================================== +-spec get_resume_timeout(state()) -> non_neg_integer(). +get_resume_timeout(#{mgmt_timeout := Timeout}) -> + Timeout. + +-spec set_resume_timeout(state(), non_neg_integer()) -> state(). +set_resume_timeout(#{mgmt_timeout := Timeout} = State, Timeout) -> + State; +set_resume_timeout(State, Timeout) -> + State1 = restart_pending_timer(State, Timeout), + State1#{mgmt_timeout => Timeout}. + +%%%=================================================================== %%% Internal functions %%%=================================================================== -spec negotiate_stream_mgmt(xmpp_element(), state()) -> state(). @@ -408,8 +425,8 @@ transition_to_pending(#{mgmt_state := active, jid := JID, lserver := LServer, mgmt_timeout := Timeout} = State) -> State1 = cancel_ack_timer(State), ?INFO_MSG("Waiting for resumption of stream for ~s", [jid:encode(JID)]), - erlang:start_timer(timer:seconds(Timeout), self(), pending_timeout), - State2 = State1#{mgmt_state => pending}, + TRef = erlang:start_timer(timer:seconds(Timeout), self(), pending_timeout), + State2 = State1#{mgmt_state => pending, mgmt_pending_timer => TRef}, ejabberd_hooks:run_fold(c2s_session_pending, LServer, State2, []); transition_to_pending(State) -> State. @@ -648,20 +665,33 @@ add_resent_delay_info(_State, El, _Time) -> send(#{mod := Mod} = State, Pkt) -> Mod:send(State, Pkt). +-spec restart_pending_timer(state(), non_neg_integer()) -> state(). +restart_pending_timer(#{mgmt_pending_timer := TRef} = State, NewTimeout) -> + cancel_timer(TRef), + NewTRef = erlang:start_timer(timer:seconds(NewTimeout), self(), + pending_timeout), + State#{mgmt_pending_timer => NewTRef}; +restart_pending_timer(State, _NewTimeout) -> + State. + -spec cancel_ack_timer(state()) -> state(). cancel_ack_timer(#{mgmt_ack_timer := TRef} = State) -> - case erlang:cancel_timer(TRef) of - false -> - receive {timeout, TRef, _} -> ok - after 0 -> ok - end; - _ -> - ok - end, + cancel_timer(TRef), maps:remove(mgmt_ack_timer, State); cancel_ack_timer(State) -> State. +-spec cancel_timer(reference()) -> ok. +cancel_timer(TRef) -> + case erlang:cancel_timer(TRef) of + false -> + receive {timeout, TRef, _} -> ok + after 0 -> ok + end; + _ -> + ok + end. + -spec bounce_message_queue() -> ok. bounce_message_queue() -> receive {route, Pkt} -> diff --git a/test/ejabberd_SUITE_data/ejabberd.yml b/test/ejabberd_SUITE_data/ejabberd.yml index 5ee287400..74cabf584 100644 --- a/test/ejabberd_SUITE_data/ejabberd.yml +++ b/test/ejabberd_SUITE_data/ejabberd.yml @@ -232,6 +232,7 @@ Welcome to this XMPP server." mod_ping: [] mod_proxy65: [] mod_push: [] + mod_push_keepalive: [] mod_s2s_dialback: [] mod_stream_mgmt: resume_timeout: 3 @@ -294,6 +295,7 @@ Welcome to this XMPP server." mod_ping: [] mod_proxy65: [] mod_push: [] + mod_push_keepalive: [] mod_s2s_dialback: [] mod_stream_mgmt: resume_timeout: 3 diff --git a/test/push_tests.erl b/test/push_tests.erl index 535671ee1..4b49cc8fe 100644 --- a/test/push_tests.erl +++ b/test/push_tests.erl @@ -77,6 +77,8 @@ master_slave_cases() -> sm_master(Config) -> ct:comment("Waiting for the slave to close the socket"), peer_down = get_event(Config), + ct:comment("Waiting a bit in order to test the keepalive feature"), + ct:sleep(5000), % Without mod_push_keepalive, the session would time out. ct:comment("Sending message to the slave"), send_test_message(Config), ct:comment("Handling push notification"), |