Chapril
/
xmpp.chapril.org-ejabberd
mirror of https://github.com/processone/ejabberd.git
24
1
Fork 0
Code Releases Activity

Add SQL support for mod_push

This commit is contained in:
Evgeniy Khramtsov 2017-10-26 20:11:43 +03:00
parent 88928c0149
commit db57394399
9 changed files with 324 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
sql/lite.sql
View File

@ -375,3 +375,14 @@ CREATE TABLE proxy65 (
CREATE UNIQUE INDEX i_proxy65_sid ON proxy65 (sid);
CREATE INDEX i_proxy65_jid ON proxy65 (jid_i);
CREATE TABLE push_session (
username text NOT NULL,
timestamp bigint NOT NULL,
service text NOT NULL,
node text NOT NULL,
xml text NOT NULL
);
CREATE UNIQUE INDEX i_push_usn ON push_session (username, service, node);
CREATE UNIQUE INDEX i_push_ut ON push_session (username, timestamp);

14
sql/mssql.sql
View File

@ -541,3 +541,17 @@ WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, ALLOW_ROW_LOCKS = ON, ALLOW
CREATE INDEX [carboncopy_user] ON [carboncopy] (username)
WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON);
CREATE TABLE [dbo].[push_session] (
[username] [varchar] (255) NOT NULL,
[timestamp] [bigint] NOT NULL,
[service] [varchar] (255) NOT NULL,
[node] [varchar] (255) NOT NULL,
[xml] [varchar] (255) NOT NULL
);
CREATE UNIQUE CLUSTERED INDEX [i_push_usn] ON [push_session] (username, service, node)
WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON);
CREATE UNIQUE CLUSTERED INDEX [i_push_ut] ON [push_session] (username, timestamp)
WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON);

11
sql/mysql.sql
View File

@ -391,3 +391,14 @@ CREATE TABLE proxy65 (
CREATE UNIQUE INDEX i_proxy65_sid ON proxy65 (sid(191));
CREATE INDEX i_proxy65_jid ON proxy65 (jid_i(191));
CREATE TABLE push_session (
username text NOT NULL,
timestamp bigint NOT NULL,
service text NOT NULL,
node text NOT NULL,
xml text NOT NULL
);
CREATE UNIQUE INDEX i_push_usn ON push_session (username(191), service(191), node(191));
CREATE UNIQUE INDEX i_push_ut ON push_session (username(191), timestamp);

11
sql/pg.sql
View File

@ -395,3 +395,14 @@ CREATE TABLE proxy65 (
CREATE UNIQUE INDEX i_proxy65_sid ON proxy65 USING btree (sid);
CREATE INDEX i_proxy65_jid ON proxy65 USING btree (jid_i);
CREATE TABLE push_session (
username text NOT NULL,
timestamp bigint NOT NULL,
service text NOT NULL,
node text NOT NULL,
xml text NOT NULL
);
CREATE UNIQUE INDEX i_push_usn ON push_session USING btree (username, service, node);
CREATE UNIQUE INDEX i_push_ut ON push_session USING btree (username, timestamp);

98
src/mod_push.erl
View File

@ -61,21 +61,21 @@
-> any().
-callback store_session(binary(), binary(), timestamp(), jid(), binary(),
xdata())
-> {ok, push_session()} | error.
-> {ok, push_session()} | {error, any()}.
-callback lookup_session(binary(), binary(), jid(), binary())
-> {ok, push_session()} | error.
-> {ok, push_session()} | error | {error, any()}.
-callback lookup_session(binary(), binary(), timestamp())
-> {ok, push_session()} | error.
-> {ok, push_session()} | error | {error, any()}.
-callback lookup_sessions(binary(), binary(), jid())
-> {ok, [push_session()]} | error.
-> {ok, [push_session()]} | {error, any()}.
-callback lookup_sessions(binary(), binary())
-> {ok, [push_session()]} | error.
-> {ok, [push_session()]} | {error, any()}.
-callback lookup_sessions(binary())
-> {ok, [push_session()]} | error.
-> {ok, [push_session()]} | {error, any()}.
-callback delete_session(binary(), binary(), timestamp())
-> ok | error.
-> ok | {error, any()}.
-callback delete_old_sessions(binary() | global, erlang:timestamp())
-> any().
-> ok | {error, any()}.
-callback use_cache(binary())
-> boolean().
-callback cache_nodes(binary())
@ -253,29 +253,39 @@ process_iq(#iq{lang = Lang, sub_els = [#push_enable{node = <<>>}]} = IQ) ->
xmpp:make_error(IQ, xmpp:err_feature_not_implemented(Txt, Lang));
process_iq(#iq{from = #jid{lserver = LServer} = JID,
to = #jid{lserver = LServer},
lang = Lang,
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())
{error, db_failure} ->
Txt = <<"Database, failure">>,
xmpp:make_error(IQ, xmpp:err_internal_server_error(Txt, Lang));
{error, notfound} ->
Txt = <<"User session not found">>,
xmpp:make_error(IQ, xmpp:err_item_not_found(Txt, Lang))
end;
process_iq(#iq{from = #jid{lserver = LServer} = JID,
to = #jid{lserver = LServer},
lang = Lang,
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())
{error, db_failure} ->
Txt = <<"Database, failure">>,
xmpp:make_error(IQ, xmpp:err_internal_server_error(Txt, Lang));
{error, notfound} ->
Txt = <<"Push record not found">>,
xmpp:make_error(IQ, xmpp:err_item_not_found(Txt, Lang))
end;
process_iq(IQ) ->
xmpp:make_error(IQ, xmpp:err_not_allowed()).
-spec enable(jid(), jid(), binary(), xdata()) -> ok | error.
-spec enable(jid(), jid(), binary(), xdata()) -> ok | {error, notfound | db_failure}.
enable(#jid{luser = LUser, lserver = LServer, lresource = LResource} = JID,
PushJID, Node, XData) ->
case ejabberd_sm:get_session_sid(LUser, LServer, LResource) of
@ -285,18 +295,18 @@ enable(#jid{luser = LUser, lserver = LServer, lresource = LResource} = JID,
?INFO_MSG("Enabling push notifications for ~s",
[jid:encode(JID)]),
ejabberd_c2s:cast(PID, push_enable);
error ->
{error, _} ->
?ERROR_MSG("Cannot enable push for ~s: database error",
[jid:encode(JID)]),
error
{error, db_failure}
end;
none ->
?WARNING_MSG("Cannot enable push for ~s: session not found",
[jid:encode(JID)]),
error
{error, notfound}
end.
-spec disable(jid(), jid(), binary() | undefined) -> ok | error.
-spec disable(jid(), jid(), binary() | undefined) -> ok | {error, notfound | db_failure}.
disable(#jid{luser = LUser, lserver = LServer, lresource = LResource} = JID,
PushJID, Node) ->
case ejabberd_sm:get_session_sid(LUser, LServer, LResource) of
@ -308,7 +318,7 @@ disable(#jid{luser = LUser, lserver = LServer, lresource = LResource} = JID,
?WARNING_MSG("Session not found while disabling push for ~s",
[jid:encode(JID)])
end,
if Node /= undefined ->
if Node /= <<>> ->
delete_session(LUser, LServer, PushJID, Node);
true ->
delete_sessions(LUser, LServer, PushJID)
@ -388,7 +398,7 @@ c2s_handle_cast(State, push_disable) ->
c2s_handle_cast(State, _Msg) ->
State.
-spec remove_user(binary(), binary()) -> ok | error.
-spec remove_user(binary(), binary()) -> ok | {error, any()}.
remove_user(LUser, LServer) ->
?INFO_MSG("Removing any push sessions of ~s@~s", [LUser, LServer]),
Mod = gen_mod:db_mod(LServer, ?MODULE),
@ -403,7 +413,7 @@ notify(#{jid := #jid{luser = LUser, lserver = LServer}, sid := {TS, _}}) ->
case lookup_session(LUser, LServer, TS) of
{ok, Client} ->
notify(LUser, LServer, [Client]);
error ->
_Err ->
ok
end.
@ -440,7 +450,7 @@ notify(LServer, PushLJID, Node, XData, HandleResponse) ->
%% Internal functions.
%%--------------------------------------------------------------------
-spec store_session(binary(), binary(), timestamp(), jid(), binary(), xdata())
-> {ok, push_session()} | error.
-> {ok, push_session()} | {error, any()}.
store_session(LUser, LServer, TS, PushJID, Node, XData) ->
Mod = gen_mod:db_mod(LServer, ?MODULE),
delete_session(LUser, LServer, PushJID, Node),
@ -460,7 +470,7 @@ store_session(LUser, LServer, TS, PushJID, Node, XData) ->
end.
-spec lookup_session(binary(), binary(), timestamp())
-> {ok, push_session()} | error.
-> {ok, push_session()} | error | {error, any()}.
lookup_session(LUser, LServer, TS) ->
Mod = gen_mod:db_mod(LServer, ?MODULE),
case use_cache(Mod, LServer) of
@ -472,7 +482,7 @@ lookup_session(LUser, LServer, TS) ->
Mod:lookup_session(LUser, LServer, TS)
end.
-spec lookup_sessions(binary(), binary()) -> {ok, [push_session()]} | error.
-spec lookup_sessions(binary(), binary()) -> {ok, [push_session()]} | {error, any()}.
lookup_sessions(LUser, LServer) ->
Mod = gen_mod:db_mod(LServer, ?MODULE),
case use_cache(Mod, LServer) of
@ -484,40 +494,48 @@ lookup_sessions(LUser, LServer) ->
Mod:lookup_sessions(LUser, LServer)
end.
-spec delete_session(binary(), binary(), timestamp()) -> ok | error.
-spec delete_session(binary(), binary(), timestamp()) -> ok | {error, db_failure}.
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
case Mod:delete_session(LUser, LServer, TS) of
ok ->
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;
{error, _} ->
{error, db_failure}
end.
-spec delete_session(binary(), binary(), jid(), binary()) -> ok | error.
-spec delete_session(binary(), binary(), jid(), binary()) -> ok | {error, notfound | db_failure}.
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
{error, notfound};
{error, _} ->
{error, db_failure}
end.
-spec delete_sessions(binary(), binary(), jid()) -> ok | error.
-spec delete_sessions(binary(), binary(), jid()) -> ok | {error, notfound | db_failure}.
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.
-spec delete_sessions(binary(), binary(), fun(() -> any()), module())
-> ok | {error, _}.
delete_sessions(LUser, LServer, LookupFun, Mod) ->
case LookupFun() of
{ok, []} ->
{error, notfound};
{ok, Clients} ->
case use_cache(Mod, LServer) of
true ->
@ -538,8 +556,8 @@ delete_sessions(LUser, LServer, LookupFun, Mod) ->
ok
end
end, Clients);
error ->
error
{error, _} ->
{error, db_failure}
end.
-spec drop_online_sessions(binary(), binary(), [push_session()])

10
src/mod_push_mnesia.erl
View File

@ -75,7 +75,7 @@ store_session(LUser, LServer, TS, PushJID, Node, XData) ->
{aborted, E} ->
?ERROR_MSG("Cannot store push session for ~s@~s: ~p",
[LUser, LServer, E]),
error
{error, db_failure}
end.
lookup_session(LUser, LServer, PushJID, Node) ->
@ -91,7 +91,7 @@ lookup_session(LUser, LServer, PushJID, Node) ->
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
@ -108,7 +108,7 @@ lookup_session(LUser, LServer, TS) ->
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
@ -164,7 +164,7 @@ delete_session(LUser, LServer, TS) ->
{aborted, E} ->
?ERROR_MSG("Cannot delete push session of ~s@~s: ~p",
[LUser, LServer, E]),
error
{error, db_failure}
end.
delete_old_sessions(_LServer, Time) ->
@ -181,7 +181,7 @@ delete_old_sessions(_LServer, Time) ->
ok;
{aborted, E} ->
?ERROR_MSG("Cannot delete old push sessions: ~p", [E]),
error
{error, db_failure}
end.
%%--------------------------------------------------------------------

196
src/mod_push_sql.erl Normal file
View File

@ -0,0 +1,196 @@
%%%----------------------------------------------------------------------
%%% 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_sql).
-behaviour(mod_push).
-compile([{parse_transform, ejabberd_sql_pt}]).
%% 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("xmpp.hrl").
-include("logger.hrl").
-include("ejabberd_sql_pt.hrl").
%%%===================================================================
%%% API
%%%===================================================================
init(_Host, _Opts) ->
ok.
store_session(LUser, LServer, NowTS, PushJID, Node, XData) ->
XML = case XData of
undefined -> <<>>;
_ -> fxml:element_to_binary(xmpp:encode(XData))
end,
TS = misc:now_to_usec(NowTS),
PushLJID = jid:tolower(PushJID),
Service = jid:encode(PushLJID),
case ?SQL_UPSERT(LServer, "push_session",
["!username=%(LUser)s",
"!timestamp=%(TS)d",
"!service=%(Service)s",
"!node=%(Node)s",
"xml=%(XML)s"]) of
ok ->
{ok, {NowTS, PushLJID, Node, XData}};
Err ->
?ERROR_MSG("Failed to update 'push_session' table: ~p", [Err]),
{error, db_failure}
end.
lookup_session(LUser, LServer, PushJID, Node) ->
PushLJID = jid:tolower(PushJID),
Service = jid:encode(PushLJID),
case ejabberd_sql:sql_query(
LServer,
?SQL("select @(timestamp)d, @(xml)s from push_session "
"where username=%(LUser)s and service=%(Service)s "
"and node=%(Node)s")) of
{selected, [{TS, XML}]} ->
NowTS = misc:usec_to_now(TS),
XData = decode_xdata(XML, LUser, LServer),
{ok, {NowTS, PushLJID, Node, XData}};
{selected, []} ->
error;
Err ->
?ERROR_MSG("Failed to select from 'push_session' table: ~p", [Err]),
{error, db_failure}
end.
lookup_session(LUser, LServer, NowTS) ->
TS = misc:now_to_usec(NowTS),
case ejabberd_sql:sql_query(
LServer,
?SQL("select @(service)s, @(node)s, @(xml)s "
"from push_session where username=%(LUser)s "
"and timestamp=%(TS)d")) of
{selected, [{Service, Node, XML}]} ->
PushLJID = jid:tolower(jid:decode(Service)),
XData = decode_xdata(XML, LUser, LServer),
{ok, {NowTS, PushLJID, Node, XData}};
{selected, []} ->
error;
Err ->
?ERROR_MSG("Failed to select from 'push_session' table: ~p", [Err]),
{error, db_failure}
end.
lookup_sessions(LUser, LServer, PushJID) ->
PushLJID = jid:tolower(PushJID),
Service = jid:encode(PushLJID),
case ejabberd_sql:sql_query(
LServer,
?SQL("select @(timestamp)d, @(xml)s, @(node)s from push_session "
"where username=%(LUser)s and service=%(Service)s")) of
{selected, Rows} ->
{ok, lists:map(
fun({TS, XML, Node}) ->
NowTS = misc:usec_to_now(TS),
XData = decode_xdata(XML, LUser, LServer),
{NowTS, PushLJID, Node, XData}
end, Rows)};
Err ->
?ERROR_MSG("Failed to select from 'push_session' table: ~p", [Err]),
{error, db_failure}
end.
lookup_sessions(LUser, LServer) ->
case ejabberd_sql:sql_query(
LServer,
?SQL("select @(timestamp)d, @(xml)s, @(node)s, @(service)s "
"from push_session where username=%(LUser)s")) of
{selected, Rows} ->
{ok, lists:map(
fun({TS, XML, Node, Service}) ->
NowTS = misc:usec_to_now(TS),
XData = decode_xdata(XML, LUser, LServer),
PushLJID = jid:tolower(jid:decode(Service)),
{NowTS, PushLJID,Node, XData}
end, Rows)};
Err ->
?ERROR_MSG("Failed to select from 'push_session' table: ~p", [Err]),
{error, db_failure}
end.
lookup_sessions(LServer) ->
case ejabberd_sql:sql_query(
LServer,
?SQL("select @(username)s, @(timestamp)d, @(xml)s, "
"@(node)s, @(service)s from push_session")) of
{selected, Rows} ->
{ok, lists:map(
fun({LUser, TS, XML, Node, Service}) ->
NowTS = misc:usec_to_now(TS),
XData = decode_xdata(XML, LUser, LServer),
PushLJID = jid:tolower(jid:decode(Service)),
{NowTS, PushLJID, Node, XData}
end, Rows)};
Err ->
?ERROR_MSG("Failed to select from 'push_session' table: ~p", [Err]),
{error, db_failure}
end.
delete_session(LUser, LServer, NowTS) ->
TS = misc:now_to_usec(NowTS),
case ejabberd_sql:sql_query(
LServer,
?SQL("delete from push_session where "
"username=%(LUser)s and timestamp=%(TS)d")) of
{updated, _} ->
ok;
Err ->
?ERROR_MSG("failed to delete from 'push_session' table: ~p", [Err]),
{error, db_failure}
end.
delete_old_sessions(LServer, Time) ->
TS = misc:now_to_usec(Time),
case ejabberd_sql:sql_query(
LServer,
?SQL("delete from push_session where timestamp<%(TS)d")) of
{updated, _} ->
ok;
Err ->
?ERROR_MSG("failed to delete from 'push_session' table: ~p", [Err]),
{error, db_failure}
end.
%%%===================================================================
%%% Internal functions
%%%===================================================================
decode_xdata(<<>>, _LUser, _LServer) ->
undefined;
decode_xdata(XML, LUser, LServer) ->
case fxml_stream:parse_element(XML) of
#xmlel{} = El ->
try xmpp:decode(El)
catch _:{xmpp_codec, Why} ->
?ERROR_MSG("Failed to decode ~s for user ~s@~s "
"from table 'push_session': ~s",
[XML, LUser, LServer, xmpp:format_error(Why)]),
undefined
end;
Err ->
?ERROR_MSG("Failed to decode ~s for user ~s@~s from "
"table 'push_session': ~p",
[XML, LUser, LServer, Err]),
undefined
end.

4
test/ejabberd_SUITE.erl
View File

@ -460,6 +460,7 @@ db_tests(_) ->
muc_tests:single_cases(),
offline_tests:single_cases(),
mam_tests:single_cases(),
push_tests:single_cases(),
test_unregister]},
muc_tests:master_slave_cases(),
privacy_tests:master_slave_cases(),
@ -469,7 +470,8 @@ db_tests(_) ->
mam_tests:master_slave_cases(),
vcard_tests:master_slave_cases(),
announce_tests:master_slave_cases(),
carbons_tests:master_slave_cases()].
carbons_tests:master_slave_cases(),
push_tests:master_slave_cases()].
ldap_tests() ->
[{ldap_tests, [sequence],

15
test/ejabberd_SUITE_data/ejabberd.yml
View File

@ -52,7 +52,12 @@ host_config:
mod_disco: []
mod_ping: []
mod_proxy65: []
mod_push:
db_type: sql
mod_push_keepalive: []
mod_s2s_dialback: []
mod_stream_mgmt:
resume_timeout: 3
mod_legacy_auth: []
mod_register:
welcome_message:
@ -110,7 +115,12 @@ Welcome to this XMPP server."
mod_disco: []
mod_ping: []
mod_proxy65: []
mod_push:
db_type: sql
mod_push_keepalive: []
mod_s2s_dialback: []
mod_stream_mgmt:
resume_timeout: 3
mod_legacy_auth: []
mod_register:
welcome_message:
@ -173,7 +183,12 @@ Welcome to this XMPP server."
mod_disco: []
mod_ping: []
mod_proxy65: []
mod_push:
db_type: sql
mod_push_keepalive: []
mod_s2s_dialback: []
mod_stream_mgmt:
resume_timeout: 3
mod_legacy_auth: []
mod_register:
welcome_message: