--- /dev/null
+%%%----------------------------------------------------------------------
+%%% File : mod_push.erl
+%%% Author : Holger Weiss <holger@zedat.fu-berlin.de>
+%%% Purpose : Push Notifications (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).
+-author('holger@zedat.fu-berlin.de').
+-protocol({xep, 357, '0.2'}).
+
+-behavior(gen_mod).
+
+%% gen_mod callbacks.
+-export([start/2, stop/1, reload/3, mod_opt_type/1, depends/2]).
+
+%% ejabberd_hooks callbacks.
+-export([disco_sm_features/5, c2s_session_pending/1, c2s_copy_session/2,
+ c2s_handle_cast/2, c2s_stanza/3, mam_message/6, offline_message/1,
+ remove_user/2]).
+
+%% gen_iq_handler callback.
+-export([process_iq/1]).
+
+%% ejabberd command.
+-export([get_commands_spec/0, delete_old_sessions/1]).
+
+-include("ejabberd.hrl").
+-include("ejabberd_commands.hrl").
+-include("logger.hrl").
+-include("xmpp.hrl").
+
+-define(PUSH_CACHE, push_cache).
+
+-type c2s_state() :: ejabberd_c2s:state().
+-type timestamp() :: erlang:timestamp().
+-type push_session() :: {timestamp(), ljid(), binary(), xdata()}.
+
+-callback init(binary(), gen_mod:opts())
+ -> any().
+-callback store_session(binary(), binary(), timestamp(), jid(), binary(),
+ xdata())
+ -> {ok, push_session()} | error.
+-callback lookup_session(binary(), binary(), jid(), binary())
+ -> {ok, push_session()} | error.
+-callback lookup_session(binary(), binary(), timestamp())
+ -> {ok, push_session()} | error.
+-callback lookup_sessions(binary(), binary(), jid())
+ -> {ok, [push_session()]} | error.
+-callback lookup_sessions(binary(), binary())
+ -> {ok, [push_session()]} | error.
+-callback lookup_sessions(binary())
+ -> {ok, [push_session()]} | error.
+-callback delete_session(binary(), binary(), timestamp())
+ -> ok | error.
+-callback delete_old_sessions(binary() | global, erlang:timestamp())
+ -> any().
+-callback use_cache(binary())
+ -> boolean().
+-callback cache_nodes(binary())
+ -> [node()].
+
+-optional_callbacks([use_cache/1, cache_nodes/1]).
+
+%%--------------------------------------------------------------------
+%% gen_mod callbacks.
+%%--------------------------------------------------------------------
+-spec start(binary(), gen_mod:opts()) -> ok.
+start(Host, Opts) ->
+ IQDisc = gen_mod:get_opt(iqdisc, Opts, gen_iq_handler:iqdisc(Host)),
+ Mod = gen_mod:db_mod(Host, Opts, ?MODULE),
+ Mod:init(Host, Opts),
+ init_cache(Mod, Host, Opts),
+ register_iq_handlers(Host, IQDisc),
+ register_hooks(Host),
+ ejabberd_commands:register_commands(get_commands_spec()).
+
+-spec stop(binary()) -> ok.
+stop(Host) ->
+ unregister_hooks(Host),
+ unregister_iq_handlers(Host),
+ ejabberd_commands:unregister_commands(get_commands_spec()).
+
+-spec reload(binary(), gen_mod:opts(), gen_mod:opts()) -> ok.
+reload(Host, NewOpts, OldOpts) ->
+ NewMod = gen_mod:db_mod(Host, NewOpts, ?MODULE),
+ OldMod = gen_mod:db_mod(Host, OldOpts, ?MODULE),
+ if NewMod /= OldMod ->
+ NewMod:init(Host, NewOpts);
+ true ->
+ ok
+ end,
+ case gen_mod:is_equal_opt(iqdisc, NewOpts, OldOpts,
+ gen_iq_handler:iqdisc(Host)) of
+ {false, IQDisc, _} ->
+ register_iq_handlers(Host, IQDisc);
+ true ->
+ ok
+ end.
+
+-spec depends(binary(), gen_mod:opts()) -> [{module(), hard | soft}].
+depends(_Host, _Opts) ->
+ [].
+
+-spec mod_opt_type(atom()) -> fun((term()) -> term()) | [atom()].
+mod_opt_type(db_type) ->
+ fun(T) -> ejabberd_config:v_db(?MODULE, T) 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(iqdisc) ->
+ fun gen_iq_handler:check_type/1;
+mod_opt_type(_) ->
+ [db_type, cache_life_time, cache_size, use_cache, cache_missed, iqdisc].
+
+%%--------------------------------------------------------------------
+%% ejabberd command callback.
+%%--------------------------------------------------------------------
+-spec get_commands_spec() -> [ejabberd_commands()].
+get_commands_spec() ->
+ [#ejabberd_commands{name = delete_old_push_sessions, tags = [purge],
+ desc = "Remove push sessions older than DAYS",
+ module = ?MODULE, function = delete_old_sessions,
+ args = [{days, integer}],
+ result = {res, rescode}}].
+
+-spec delete_old_sessions(non_neg_integer()) -> ok | any().
+delete_old_sessions(Days) ->
+ CurrentTime = p1_time_compat:system_time(micro_seconds),
+ Diff = Days * 24 * 60 * 60 * 1000000,
+ TimeStamp = misc:usec_to_now(CurrentTime - Diff),
+ DBTypes = lists:usort(
+ lists:map(
+ fun(Host) ->
+ case gen_mod:db_type(Host, ?MODULE) of
+ sql -> {sql, Host};
+ Other -> {Other, global}
+ end
+ end, ?MYHOSTS)),
+ Results = lists:map(
+ fun({DBType, Host}) ->
+ Mod = gen_mod:db_mod(DBType, ?MODULE),
+ Mod:delete_old_sessions(Host, TimeStamp)
+ end, DBTypes),
+ ets_cache:clear(?PUSH_CACHE, ejabberd_cluster:get_nodes()),
+ case lists:filter(fun(Res) -> Res /= ok end, Results) of
+ [] ->
+ ?INFO_MSG("Deleted push sessions older than ~B days", [Days]),
+ ok;
+ [NotOk | _] ->
+ ?ERROR_MSG("Error while deleting old push sessions: ~p", [NotOk]),
+ NotOk
+ end.
+
+%%--------------------------------------------------------------------
+%% Register/unregister hooks.
+%%--------------------------------------------------------------------
+-spec register_hooks(binary()) -> ok.
+register_hooks(Host) ->
+ ejabberd_hooks:add(disco_sm_features, Host, ?MODULE,
+ disco_sm_features, 50),
+ ejabberd_hooks:add(c2s_session_pending, Host, ?MODULE,
+ c2s_session_pending, 50),
+ ejabberd_hooks:add(c2s_copy_session, Host, ?MODULE,
+ c2s_copy_session, 50),
+ ejabberd_hooks:add(c2s_handle_cast, Host, ?MODULE,
+ c2s_handle_cast, 50),
+ ejabberd_hooks:add(c2s_handle_send, Host, ?MODULE,
+ c2s_stanza, 50),
+ ejabberd_hooks:add(store_mam_message, Host, ?MODULE,
+ mam_message, 50),
+ ejabberd_hooks:add(offline_message_hook, Host, ?MODULE,
+ offline_message, 50),
+ ejabberd_hooks:add(remove_user, Host, ?MODULE,
+ remove_user, 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_copy_session, Host, ?MODULE,
+ c2s_copy_session, 50),
+ ejabberd_hooks:delete(c2s_handle_cast, Host, ?MODULE,
+ c2s_handle_cast, 50),
+ ejabberd_hooks:delete(c2s_handle_send, Host, ?MODULE,
+ c2s_stanza, 50),
+ ejabberd_hooks:delete(store_mam_message, Host, ?MODULE,
+ mam_message, 50),
+ ejabberd_hooks:delete(offline_message_hook, Host, ?MODULE,
+ offline_message, 50),
+ ejabberd_hooks:delete(remove_user, Host, ?MODULE,
+ remove_user, 50).
+
+%%--------------------------------------------------------------------
+%% Service discovery.
+%%--------------------------------------------------------------------
+-spec disco_sm_features(empty | {result, [binary()]} | {error, stanza_error()},
+ jid(), jid(), binary(), binary())
+ -> {result, [binary()]} | {error, stanza_error()}.
+disco_sm_features(empty, From, To, Node, Lang) ->
+ disco_sm_features({result, []}, From, To, Node, Lang);
+disco_sm_features({result, OtherFeatures},
+ #jid{luser = U, lserver = S},
+ #jid{luser = U, lserver = S}, <<"">>, _Lang) ->
+ {result, [?NS_PUSH_0 | OtherFeatures]};
+disco_sm_features(Acc, _From, _To, _Node, _Lang) ->
+ Acc.
+
+%%--------------------------------------------------------------------
+%% IQ handlers.
+%%--------------------------------------------------------------------
+-spec register_iq_handlers(binary(), gen_iq_handler:type()) -> ok.
+register_iq_handlers(Host, IQDisc) ->
+ gen_iq_handler:add_iq_handler(ejabberd_sm, Host, ?NS_PUSH_0,
+ ?MODULE, process_iq, IQDisc).
+
+-spec unregister_iq_handlers(binary()) -> ok.
+unregister_iq_handlers(Host) ->
+ gen_iq_handler:remove_iq_handler(ejabberd_sm, Host, ?NS_PUSH_0).
+
+-spec process_iq(iq()) -> iq().
+process_iq(#iq{type = get, lang = Lang} = IQ) ->
+ Txt = <<"Value 'get' of 'type' attribute is not allowed">>,
+ xmpp:make_error(IQ, xmpp:err_not_allowed(Txt, Lang));
+process_iq(#iq{lang = Lang, sub_els = [#push_enable{node = <<>>}]} = IQ) ->
+ Txt = <<"Enabling push without 'node' attribute is not supported">>,
+ xmpp:make_error(IQ, xmpp:err_feature_not_implemented(Txt, Lang));
+process_iq(#iq{from = #jid{lserver = LServer} = JID,
+ to = #jid{lserver = LServer},
+ sub_els = [#push_enable{jid = PushJID,
+ node = Node,
+ xdata = XData}]} = IQ) ->
+ case enable(JID, PushJID, Node, XData) of
+ ok ->
+ xmpp:make_iq_result(IQ);
+ error ->
+ xmpp:make_error(IQ, xmpp:err_internal_server_error())
+ end;
+process_iq(#iq{from = #jid{lserver = LServer} = JID,
+ to = #jid{lserver = LServer},
+ sub_els = [#push_disable{jid = PushJID,
+ node = Node}]} = IQ) ->
+ case disable(JID, PushJID, Node) of
+ ok ->
+ xmpp:make_iq_result(IQ);
+ error ->
+ xmpp:make_error(IQ, xmpp:err_item_not_found())
+ end;
+process_iq(IQ) ->
+ xmpp:make_error(IQ, xmpp:err_not_allowed()).
+
+-spec enable(jid(), jid(), binary(), xdata()) -> ok | error.
+enable(#jid{luser = LUser, lserver = LServer, lresource = LResource} = JID,
+ PushJID, Node, XData) ->
+ case ejabberd_sm:get_session_sid(LUser, LServer, LResource) of
+ {TS, PID} ->
+ case store_session(LUser, LServer, TS, PushJID, Node, XData) of
+ {ok, _} ->
+ ?INFO_MSG("Enabling push notifications for ~s",
+ [jid:encode(JID)]),
+ ejabberd_c2s:cast(PID, push_enable);
+ error ->
+ ?ERROR_MSG("Cannot enable push for ~s: database error",
+ [jid:encode(JID)]),
+ error
+ end;
+ none ->
+ ?WARNING_MSG("Cannot enable push for ~s: session not found",
+ [jid:encode(JID)]),
+ error
+ end.
+
+-spec disable(jid(), jid(), binary() | undefined) -> ok | error.
+disable(#jid{luser = LUser, lserver = LServer, lresource = LResource} = JID,
+ PushJID, Node) ->
+ case ejabberd_sm:get_session_sid(LUser, LServer, LResource) of
+ {_TS, PID} ->
+ ?INFO_MSG("Disabling push notifications for ~s",
+ [jid:encode(JID)]),
+ ejabberd_c2s:cast(PID, push_disable);
+ none ->
+ ?WARNING_MSG("Session not found while disabling push for ~s",
+ [jid:encode(JID)])
+ end,
+ if Node /= undefined ->
+ delete_session(LUser, LServer, PushJID, Node);
+ true ->
+ delete_sessions(LUser, LServer, PushJID)
+ end.
+
+%%--------------------------------------------------------------------
+%% Hook callbacks.
+%%--------------------------------------------------------------------
+-spec c2s_stanza(c2s_state(), xmpp_element() | xmlel(), term()) -> c2s_state().
+c2s_stanza(#{push_enabled := true, mgmt_state := pending} = State,
+ _Pkt, _SendResult) ->
+ notify(State),
+ State;
+c2s_stanza(State, _Pkt, _SendResult) ->
+ State.
+
+-spec mam_message(message() | drop, binary(), binary(), jid(),
+ chat | groupchat, recv | send) -> message().
+mam_message(#message{meta = #{push_notified := true}} = Pkt,
+ _LUser, _LServer, _Peer, _Type, _Dir) ->
+ Pkt;
+mam_message(#message{} = Pkt, LUser, LServer, _Peer, chat, _Dir) ->
+ case lookup_sessions(LUser, LServer) of
+ {ok, [_|_] = Clients} ->
+ case drop_online_sessions(LUser, LServer, Clients) of
+ [_|_] = Clients1 ->
+ ?DEBUG("Notifying ~s@~s of MAM message", [LUser, LServer]),
+ notify(LUser, LServer, Clients1);
+ [] ->
+ ok
+ end;
+ _ ->
+ ok
+ end,
+ xmpp:put_meta(Pkt, push_notified, true);
+mam_message(Pkt, _LUser, _LServer, _Peer, _Type, _Dir) ->
+ Pkt.
+
+-spec offline_message({any(), message()}) -> {any(), message()}.
+offline_message({_Action, #message{meta = #{push_notified := true}}} = Acc) ->
+ Acc;
+offline_message({Action, #message{to = #jid{luser = LUser,
+ lserver = LServer}} = Pkt}) ->
+ case lookup_sessions(LUser, LServer) of
+ {ok, [_|_] = Clients} ->
+ ?DEBUG("Notifying ~s@~s of offline message", [LUser, LServer]),
+ notify(LUser, LServer, Clients);
+ _ ->
+ ok
+ end,
+ {Action, xmpp:put_meta(Pkt, push_notified, true)}.
+
+-spec c2s_session_pending(c2s_state()) -> c2s_state().
+c2s_session_pending(#{push_enabled := true, mgmt_queue := Queue} = State) ->
+ case p1_queue:len(Queue) of
+ Len when Len > 0 ->
+ ?DEBUG("Notifying client of unacknowledged messages", []),
+ notify(State),
+ State;
+ 0 ->
+ State
+ end;
+c2s_session_pending(State) ->
+ State.
+
+-spec c2s_copy_session(c2s_state(), c2s_state()) -> c2s_state().
+c2s_copy_session(State, #{push_enabled := true}) ->
+ State#{push_enabled => true};
+c2s_copy_session(State, _) ->
+ State.
+
+-spec c2s_handle_cast(c2s_state(), any()) -> c2s_state() | {stop, c2s_state()}.
+c2s_handle_cast(State, push_enable) ->
+ {stop, State#{push_enabled => true}};
+c2s_handle_cast(State, push_disable) ->
+ {stop, maps:remove(push_enabled, State)};
+c2s_handle_cast(State, _Msg) ->
+ State.
+
+-spec remove_user(binary(), binary()) -> ok | error.
+remove_user(LUser, LServer) ->
+ ?INFO_MSG("Removing any push sessions of ~s@~s", [LUser, LServer]),
+ Mod = gen_mod:db_mod(LServer, ?MODULE),
+ LookupFun = fun() -> Mod:lookup_sessions(LUser, LServer) end,
+ delete_sessions(LUser, LServer, LookupFun, Mod).
+
+%%--------------------------------------------------------------------
+%% Internal functions.
+%%--------------------------------------------------------------------
+-spec notify(c2s_state()) -> ok.
+notify(#{jid := #jid{luser = LUser, lserver = LServer}, sid := {TS, _}}) ->
+ case lookup_session(LUser, LServer, TS) of
+ {ok, Client} ->
+ notify(LUser, LServer, [Client]);
+ error ->
+ ok
+ end.
+
+-spec notify(binary(), binary(), [push_session()]) -> ok.
+notify(LUser, LServer, Clients) ->
+ lists:foreach(
+ fun({TS, PushLJID, Node, XData}) ->
+ HandleResponse = fun(#iq{type = result}) ->
+ ok;
+ (#iq{type = error}) ->
+ delete_session(LUser, LServer, TS);
+ (timeout) ->
+ ok % Hmm.
+ end,
+ notify(LServer, PushLJID, Node, XData, HandleResponse)
+ end, Clients).
+
+-spec notify(binary(), ljid(), binary(), xdata(),
+ fun((iq() | timeout) -> any())) -> ok.
+notify(LServer, PushLJID, Node, XData, HandleResponse) ->
+ From = jid:make(LServer),
+ Item = #ps_item{xml_els = [xmpp:encode(#push_notification{})]},
+ PubSub = #pubsub{publish = #ps_publish{node = Node, items = [Item]},
+ publish_options = XData},
+ IQ = #iq{type = set,
+ from = From,
+ to = jid:make(PushLJID),
+ id = randoms:get_string(),
+ sub_els = [PubSub]},
+ ejabberd_local:route_iq(IQ, HandleResponse),
+ ok.
+
+-spec store_session(binary(), binary(), timestamp(), jid(), binary(), xdata())
+ -> {ok, push_session()} | error.
+store_session(LUser, LServer, TS, PushJID, Node, XData) ->
+ Mod = gen_mod:db_mod(LServer, ?MODULE),
+ delete_session(LUser, LServer, PushJID, Node),
+ case use_cache(Mod, LServer) of
+ true ->
+ ets_cache:delete(?PUSH_CACHE, {LUser, LServer},
+ cache_nodes(Mod, LServer)),
+ ets_cache:update(
+ ?PUSH_CACHE,
+ {LUser, LServer, TS}, {ok, {TS, PushJID, Node, XData}},
+ fun() ->
+ Mod:store_session(LUser, LServer, TS, PushJID, Node,
+ XData)
+ end, cache_nodes(Mod, LServer));
+ false ->
+ Mod:store_session(LUser, LServer, TS, PushJID, Node, XData)
+ end.
+
+-spec lookup_session(binary(), binary(), timestamp())
+ -> {ok, push_session()} | error.
+lookup_session(LUser, LServer, TS) ->
+ Mod = gen_mod:db_mod(LServer, ?MODULE),
+ case use_cache(Mod, LServer) of
+ true ->
+ ets_cache:lookup(
+ ?PUSH_CACHE, {LUser, LServer, TS},
+ fun() -> Mod:lookup_session(LUser, LServer, TS) end);
+ false ->
+ Mod:lookup_session(LUser, LServer, TS)
+ end.
+
+-spec lookup_sessions(binary(), binary()) -> {ok, [push_session()]} | error.
+lookup_sessions(LUser, LServer) ->
+ Mod = gen_mod:db_mod(LServer, ?MODULE),
+ case use_cache(Mod, LServer) of
+ true ->
+ ets_cache:lookup(
+ ?PUSH_CACHE, {LUser, LServer},
+ fun() -> Mod:lookup_sessions(LUser, LServer) end);
+ false ->
+ Mod:lookup_sessions(LUser, LServer)
+ end.
+
+-spec delete_session(binary(), binary(), timestamp()) -> ok | error.
+delete_session(LUser, LServer, TS) ->
+ Mod = gen_mod:db_mod(LServer, ?MODULE),
+ ok = Mod:delete_session(LUser, LServer, TS),
+ case use_cache(Mod, LServer) of
+ true ->
+ ets_cache:delete(?PUSH_CACHE, {LUser, LServer},
+ cache_nodes(Mod, LServer)),
+ ets_cache:delete(?PUSH_CACHE, {LUser, LServer, TS},
+ cache_nodes(Mod, LServer));
+ false ->
+ ok
+ end.
+
+-spec delete_session(binary(), binary(), jid(), binary()) -> ok | error.
+delete_session(LUser, LServer, PushJID, Node) ->
+ Mod = gen_mod:db_mod(LServer, ?MODULE),
+ case Mod:lookup_session(LUser, LServer, PushJID, Node) of
+ {ok, {TS, _, _, _}} ->
+ delete_session(LUser, LServer, TS);
+ error ->
+ error
+ end.
+
+-spec delete_sessions(binary(), binary(), jid()) -> ok | error.
+delete_sessions(LUser, LServer, PushJID) ->
+ Mod = gen_mod:db_mod(LServer, ?MODULE),
+ LookupFun = fun() -> Mod:lookup_sessions(LUser, LServer, PushJID) end,
+ delete_sessions(LUser, LServer, LookupFun, Mod).
+
+-spec delete_sessions(binary(), binary(), fun(() -> ok | error), module())
+ -> ok | error.
+delete_sessions(LUser, LServer, LookupFun, Mod) ->
+ case LookupFun() of
+ {ok, Clients} ->
+ case use_cache(Mod, LServer) of
+ true ->
+ ets_cache:delete(?PUSH_CACHE, {LUser, LServer},
+ cache_nodes(Mod, LServer));
+ false ->
+ ok
+ end,
+ lists:foreach(
+ fun({TS, _, _, _}) ->
+ ok = Mod:delete_session(LUser, LServer, TS),
+ case use_cache(Mod, LServer) of
+ true ->
+ ets_cache:delete(?PUSH_CACHE,
+ {LUser, LServer, TS},
+ cache_nodes(Mod, LServer));
+ false ->
+ ok
+ end
+ end, Clients);
+ error ->
+ error
+ end.
+
+-spec drop_online_sessions(binary(), binary(), [push_session()])
+ -> [push_session()].
+drop_online_sessions(LUser, LServer, Clients) ->
+ SessIDs = ejabberd_sm:get_session_sids(LUser, LServer),
+ [Client || {TS, _, _, _} = Client <- Clients,
+ not lists:keyfind(TS, 1, SessIDs)].
+
+%%--------------------------------------------------------------------
+%% Caching.
+%%--------------------------------------------------------------------
+-spec init_cache(module(), binary(), gen_mod:opts()) -> ok.
+init_cache(Mod, Host, Opts) ->
+ case use_cache(Mod, Host) of
+ true ->
+ CacheOpts = cache_opts(Host, Opts),
+ ets_cache:new(?PUSH_CACHE, CacheOpts);
+ false ->
+ ets_cache:delete(?PUSH_CACHE)
+ end.
+
+-spec cache_opts(binary(), gen_mod:opts()) -> [proplists:property()].
+cache_opts(Host, Opts) ->
+ MaxSize = gen_mod:get_opt(
+ cache_size, Opts,
+ ejabberd_config:cache_size(Host)),
+ CacheMissed = gen_mod:get_opt(
+ cache_missed, Opts,
+ ejabberd_config:cache_missed(Host)),
+ LifeTime = case gen_mod:get_opt(
+ cache_life_time, Opts,
+ ejabberd_config:cache_life_time(Host)) of
+ infinity -> infinity;
+ I -> timer:seconds(I)
+ end,
+ [{max_size, MaxSize}, {cache_missed, CacheMissed}, {life_time, LifeTime}].
+
+-spec use_cache(module(), binary()) -> boolean().
+use_cache(Mod, Host) ->
+ case erlang:function_exported(Mod, use_cache, 1) of
+ true -> Mod:use_cache(Host);
+ false ->
+ gen_mod:get_module_opt(
+ Host, ?MODULE, use_cache,
+ ejabberd_config:use_cache(Host))
+ end.
+
+-spec cache_nodes(module(), binary()) -> [node()].
+cache_nodes(Mod, Host) ->
+ case erlang:function_exported(Mod, cache_nodes, 1) of
+ true -> Mod:cache_nodes(Host);
+ false -> ejabberd_cluster:get_nodes()
+ end.
--- /dev/null
+%%%----------------------------------------------------------------------
+%%% File : mod_push_mnesia.erl
+%%% Author : Holger Weiss <holger@zedat.fu-berlin.de>
+%%% Purpose : Mnesia backend for Push Notifications (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_mnesia).
+-author('holger@zedat.fu-berlin.de').
+
+-behavior(mod_push).
+
+%% API
+-export([init/2, store_session/6, lookup_session/4, lookup_session/3,
+ lookup_sessions/3, lookup_sessions/2, lookup_sessions/1,
+ delete_session/3, delete_old_sessions/2]).
+
+-include_lib("stdlib/include/ms_transform.hrl").
+-include("logger.hrl").
+-include("xmpp.hrl").
+
+-record(push_session,
+ {us = {<<"">>, <<"">>} :: {binary(), binary()},
+ timestamp = p1_time_compat:timestamp() :: erlang:timestamp(),
+ service = {<<"">>, <<"">>, <<"">>} :: ljid(),
+ node = <<"">> :: binary(),
+ xdata = #xdata{} :: xdata()}).
+
+%%%-------------------------------------------------------------------
+%%% API
+%%%-------------------------------------------------------------------
+init(_Host, _Opts) ->
+ ejabberd_mnesia:create(?MODULE, push_session,
+ [{disc_only_copies, [node()]},
+ {type, bag},
+ {attributes, record_info(fields, push_session)}]).
+
+store_session(LUser, LServer, TS, PushJID, Node, XData) ->
+ US = {LUser, LServer},
+ PushLJID = jid:tolower(PushJID),
+ MaxSessions = ejabberd_sm:get_max_user_sessions(LUser, LServer),
+ F = fun() ->
+ if is_integer(MaxSessions) ->
+ enforce_max_sessions(US, MaxSessions - 1);
+ MaxSessions == infinity ->
+ ok
+ end,
+ mnesia:write(#push_session{us = US,
+ timestamp = TS,
+ service = PushLJID,
+ node = Node,
+ xdata = XData})
+ end,
+ case mnesia:transaction(F) of
+ {atomic, ok} ->
+ {ok, {TS, PushLJID, Node, XData}};
+ {aborted, E} ->
+ ?ERROR_MSG("Cannot store push session for ~s@~s: ~p",
+ [LUser, LServer, E]),
+ error
+ end.
+
+lookup_session(LUser, LServer, PushJID, Node) ->
+ PushLJID = jid:tolower(PushJID),
+ MatchSpec = ets:fun2ms(
+ fun(#push_session{us = {U, S}, service = P, node = N} = Rec)
+ when U == LUser,
+ S == LServer,
+ P == PushLJID,
+ N == Node ->
+ Rec
+ end),
+ case mnesia:dirty_select(push_session, MatchSpec) of
+ [#push_session{timestamp = TS, xdata = XData}] ->
+ {ok, {TS, PushLJID, Node, XData}};
+ _ ->
+ ?DEBUG("No push session found for ~s@~s (~p, ~s)",
+ [LUser, LServer, PushJID, Node]),
+ error
+ end.
+
+lookup_session(LUser, LServer, TS) ->
+ MatchSpec = ets:fun2ms(
+ fun(#push_session{us = {U, S}, timestamp = T} = Rec)
+ when U == LUser,
+ S == LServer,
+ T == TS ->
+ Rec
+ end),
+ case mnesia:dirty_select(push_session, MatchSpec) of
+ [#push_session{service = PushLJID, node = Node, xdata = XData}] ->
+ {ok, {TS, PushLJID, Node, XData}};
+ _ ->
+ ?DEBUG("No push session found for ~s@~s (~p)",
+ [LUser, LServer, TS]),
+ error
+ end.
+
+lookup_sessions(LUser, LServer, PushJID) ->
+ PushLJID = jid:tolower(PushJID),
+ MatchSpec = ets:fun2ms(
+ fun(#push_session{us = {U, S}, service = P, node = N} = Rec)
+ when U == LUser,
+ S == LServer,
+ P == PushLJID ->
+ Rec
+ end),
+ {ok, mnesia:dirty_select(push_session, MatchSpec)}.
+
+lookup_sessions(LUser, LServer) ->
+ Records = mnesia:dirty_read(push_session, {LUser, LServer}),
+ Clients = [{TS, PushLJID, Node, XData}
+ || #push_session{timestamp = TS,
+ service = PushLJID,
+ node = Node,
+ xdata = XData} <- Records],
+ {ok, Clients}.
+
+lookup_sessions(LServer) ->
+ MatchSpec = ets:fun2ms(
+ fun(#push_session{us = {_U, S},
+ timestamp = TS,
+ service = PushLJID,
+ node = Node,
+ xdata = XData})
+ when S == LServer ->
+ {TS, PushLJID, Node, XData}
+ end),
+ {ok, mnesia:dirty_select(push_session, MatchSpec)}.
+
+delete_session(LUser, LServer, TS) ->
+ MatchSpec = ets:fun2ms(
+ fun(#push_session{us = {U, S}, timestamp = T} = Rec)
+ when U == LUser,
+ S == LServer,
+ T == TS ->
+ Rec
+ end),
+ F = fun() ->
+ Recs = mnesia:select(push_session, MatchSpec),
+ lists:foreach(fun mnesia:delete_object/1, Recs)
+ end,
+ case mnesia:transaction(F) of
+ {atomic, ok} ->
+ ok;
+ {aborted, E} ->
+ ?ERROR_MSG("Cannot delete push seesion of ~s@~s: ~p",
+ [LUser, LServer, E]),
+ error
+ end.
+
+delete_old_sessions(_LServer, Time) ->
+ DelIfOld = fun(#push_session{timestamp = T} = Rec, ok) when T < Time ->
+ mnesia:delete_object(Rec);
+ (_Rec, ok) ->
+ ok
+ end,
+ F = fun() ->
+ mnesia:foldl(DelIfOld, ok, push_session)
+ end,
+ case mnesia:transaction(F) of
+ {atomic, ok} ->
+ ok;
+ {aborted, E} ->
+ ?ERROR_MSG("Cannot delete old push sessions: ~p", [E]),
+ error
+ end.
+
+%%--------------------------------------------------------------------
+%% Internal functions.
+%%--------------------------------------------------------------------
+-spec enforce_max_sessions({binary(), binary()}, non_neg_integer()) -> ok.
+enforce_max_sessions({U, S} = US, Max) ->
+ Recs = mnesia:wread({push_session, US}),
+ NumRecs = length(Recs),
+ if NumRecs > Max ->
+ NumOldRecs = NumRecs - Max,
+ Recs1 = lists:keysort(#push_session.timestamp, Recs),
+ Recs2 = lists:reverse(Recs1),
+ OldRecs = lists:sublist(Recs2, Max + 1, NumOldRecs),
+ ?INFO_MSG("Disabling ~B old push session(s) of ~s@~s",
+ [NumOldRecs, U, S]),
+ lists:foreach(fun(Rec) -> mnesia:delete_object(Rec) end, OldRecs);
+ true ->
+ ok
+ end.
--- /dev/null
+%%%-------------------------------------------------------------------
+%%% Author : Holger Weiss <holger@zedat.fu-berlin.de>
+%%% 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(push_tests).
+
+%% API
+-compile(export_all).
+-import(suite, [close_socket/1, connect/1, disconnect/1, get_event/1,
+ get_features/2, make_iq_result/1, my_jid/1, put_event/2, recv/1,
+ recv_iq/1, recv_message/1, self_presence/2, send/2, send_recv/2,
+ server_jid/1]).
+
+-include("suite.hrl").
+
+-define(PUSH_NODE, <<"d3v1c3">>).
+-define(PUSH_XDATA_FIELDS,
+ [#xdata_field{var = <<"FORM_TYPE">>,
+ values = [?NS_PUBSUB_PUBLISH_OPTIONS]},
+ #xdata_field{var = <<"secret">>,
+ values = [<<"c0nf1d3nt14l">>]}]).
+
+%%%===================================================================
+%%% API
+%%%===================================================================
+%%%===================================================================
+%%% Single user tests
+%%%===================================================================
+single_cases() ->
+ {push_single, [sequence],
+ [single_test(feature_enabled),
+ single_test(unsupported_iq)]}.
+
+feature_enabled(Config) ->
+ BareMyJID = jid:remove_resource(my_jid(Config)),
+ Features = get_features(Config, BareMyJID),
+ true = lists:member(?NS_PUSH_0, Features),
+ disconnect(Config).
+
+unsupported_iq(Config) ->
+ PushJID = my_jid(Config),
+ lists:foreach(
+ fun(SubEl) ->
+ #iq{type = error} =
+ send_recv(Config, #iq{type = get, sub_els = [SubEl]})
+ end, [#push_enable{jid = PushJID}, #push_disable{jid = PushJID}]),
+ disconnect(Config).
+
+%%%===================================================================
+%%% Master-slave tests
+%%%===================================================================
+master_slave_cases() ->
+ {push_master_slave, [sequence],
+ [master_slave_test(sm),
+ master_slave_test(offline),
+ master_slave_test(mam)]}.
+
+sm_master(Config) ->
+ ct:comment("Waiting for the slave to close the socket"),
+ peer_down = get_event(Config),
+ ct:comment("Sending message to the slave"),
+ send_test_message(Config),
+ ct:comment("Handling push notification"),
+ handle_notification(Config),
+ ct:comment("Receiving bounced message from the slave"),
+ #message{type = error} = recv_message(Config),
+ ct:comment("Closing the connection"),
+ disconnect(Config).
+
+sm_slave(Config) ->
+ ct:comment("Enabling push notifications"),
+ ok = enable_push(Config),
+ ct:comment("Enabling stream management"),
+ ok = enable_sm(Config),
+ ct:comment("Closing the socket"),
+ close_socket(Config).
+
+offline_master(Config) ->
+ ct:comment("Waiting for the slave to be ready"),
+ ready = get_event(Config),
+ ct:comment("Sending message to the slave"),
+ send_test_message(Config), % No push notification, slave is online.
+ ct:comment("Waiting for the slave to disconnect"),
+ peer_down = get_event(Config),
+ ct:comment("Sending message to offline storage"),
+ send_test_message(Config),
+ ct:comment("Handling push notification for offline message"),
+ handle_notification(Config),
+ ct:comment("Closing the connection"),
+ disconnect(Config).
+
+offline_slave(Config) ->
+ ct:comment("Re-enabling push notifications"),
+ ok = enable_push(Config),
+ ct:comment("Letting the master know that we're ready"),
+ put_event(Config, ready),
+ ct:comment("Receiving message from the master"),
+ recv_test_message(Config),
+ ct:comment("Closing the connection"),
+ disconnect(Config).
+
+mam_master(Config) ->
+ ct:comment("Waiting for the slave to be ready"),
+ ready = get_event(Config),
+ ct:comment("Sending message to the slave"),
+ send_test_message(Config),
+ ct:comment("Handling push notification for MAM message"),
+ handle_notification(Config),
+ ct:comment("Closing the connection"),
+ disconnect(Config).
+
+mam_slave(Config) ->
+ self_presence(Config, available),
+ ct:comment("Receiving message from offline storage"),
+ recv_test_message(Config),
+ ct:comment("Re-enabling push notifications"),
+ ok = enable_push(Config),
+ ct:comment("Enabling MAM"),
+ ok = enable_mam(Config),
+ ct:comment("Letting the master know that we're ready"),
+ put_event(Config, ready),
+ ct:comment("Receiving message from the master"),
+ recv_test_message(Config),
+ ct:comment("Waiting for the master to disconnect"),
+ peer_down = get_event(Config),
+ ct:comment("Disabling push notifications"),
+ ok = disable_push(Config),
+ ct:comment("Closing the connection and cleaning up"),
+ clean(disconnect(Config)).
+
+%%%===================================================================
+%%% Internal functions
+%%%===================================================================
+single_test(T) ->
+ list_to_atom("push_" ++ atom_to_list(T)).
+
+master_slave_test(T) ->
+ {list_to_atom("push_" ++ atom_to_list(T)), [parallel],
+ [list_to_atom("push_" ++ atom_to_list(T) ++ "_master"),
+ list_to_atom("push_" ++ atom_to_list(T) ++ "_slave")]}.
+
+enable_sm(Config) ->
+ send(Config, #sm_enable{xmlns = ?NS_STREAM_MGMT_3, resume = true}),
+ case recv(Config) of
+ #sm_enabled{resume = true} ->
+ ok;
+ #sm_failed{reason = Reason} ->
+ Reason
+ end.
+
+enable_mam(Config) ->
+ case send_recv(
+ Config, #iq{type = set, sub_els = [#mam_prefs{xmlns = ?NS_MAM_1,
+ default = always}]}) of
+ #iq{type = result} ->
+ ok;
+ #iq{type = error} = Err ->
+ xmpp:get_error(Err)
+ end.
+
+enable_push(Config) ->
+ %% Usually, the push JID would be a server JID (such as push.example.com).
+ %% We specify the peer's full user JID instead, so the push notifications
+ %% will be sent to the peer.
+ PushJID = ?config(peer, Config),
+ XData = #xdata{type = submit, fields = ?PUSH_XDATA_FIELDS},
+ case send_recv(
+ Config, #iq{type = set,
+ sub_els = [#push_enable{jid = PushJID,
+ node = ?PUSH_NODE,
+ xdata = XData}]}) of
+ #iq{type = result, sub_els = []} ->
+ ok;
+ #iq{type = error} = Err ->
+ xmpp:get_error(Err)
+ end.
+
+disable_push(Config) ->
+ PushJID = ?config(peer, Config),
+ case send_recv(
+ Config, #iq{type = set,
+ sub_els = [#push_disable{jid = PushJID,
+ node = ?PUSH_NODE}]}) of
+ #iq{type = result, sub_els = []} ->
+ ok;
+ #iq{type = error} = Err ->
+ xmpp:get_error(Err)
+ end.
+
+send_test_message(Config) ->
+ Peer = ?config(peer, Config),
+ Msg = #message{to = Peer, body = [#text{data = <<"test">>}]},
+ send(Config, Msg).
+
+recv_test_message(Config) ->
+ Peer = ?config(peer, Config),
+ #message{from = Peer,
+ body = [#text{data = <<"test">>}]} = recv_message(Config).
+
+handle_notification(Config) ->
+ From = server_jid(Config),
+ Item = #ps_item{xml_els = [xmpp:encode(#push_notification{})]},
+ Publish = #ps_publish{node = ?PUSH_NODE, items = [Item]},
+ XData = #xdata{type = submit, fields = ?PUSH_XDATA_FIELDS},
+ PubSub = #pubsub{publish = Publish, publish_options = XData},
+ IQ = #iq{type = set, from = From, sub_els = [PubSub]} = recv_iq(Config),
+ send(Config, make_iq_result(IQ)).
+
+clean(Config) ->
+ {U, S, _} = jid:tolower(my_jid(Config)),
+ mod_push:remove_user(U, S),
+ mod_mam:remove_user(U, S),
+ Config.
projects / ejabberd / commitdiff