mirror of
https://github.com/processone/ejabberd.git
synced 2024-12-22 17:28:25 +01:00
233 lines
6.9 KiB
Erlang
233 lines
6.9 KiB
Erlang
%%%-------------------------------------------------------------------
|
|
%%% @author Evgeny Khramtsov <ekhramtsov@process-one.net>
|
|
%%% @copyright (C) 2016, Evgeny Khramtsov
|
|
%%% @doc
|
|
%%%
|
|
%%% @end
|
|
%%% Created : 15 Apr 2016 by Evgeny Khramtsov <ekhramtsov@process-one.net>
|
|
%%%-------------------------------------------------------------------
|
|
-module(mod_offline_mnesia).
|
|
|
|
-behaviour(mod_offline).
|
|
|
|
-export([init/2, store_messages/5, pop_messages/2, remove_expired_messages/1,
|
|
remove_old_messages/2, remove_user/2, read_message_headers/2,
|
|
read_message/3, remove_message/3, read_all_messages/2,
|
|
remove_all_messages/2, count_messages/2, import/2]).
|
|
|
|
-include("jlib.hrl").
|
|
-include("mod_offline.hrl").
|
|
-include("logger.hrl").
|
|
|
|
-define(OFFLINE_TABLE_LOCK_THRESHOLD, 1000).
|
|
|
|
%%%===================================================================
|
|
%%% API
|
|
%%%===================================================================
|
|
init(_Host, _Opts) ->
|
|
mnesia:create_table(offline_msg,
|
|
[{disc_only_copies, [node()]}, {type, bag},
|
|
{attributes, record_info(fields, offline_msg)}]),
|
|
update_table().
|
|
|
|
store_messages(_Host, US, Msgs, Len, MaxOfflineMsgs) ->
|
|
F = fun () ->
|
|
Count = if MaxOfflineMsgs =/= infinity ->
|
|
Len + count_mnesia_records(US);
|
|
true -> 0
|
|
end,
|
|
if Count > MaxOfflineMsgs -> discard;
|
|
true ->
|
|
if Len >= (?OFFLINE_TABLE_LOCK_THRESHOLD) ->
|
|
mnesia:write_lock_table(offline_msg);
|
|
true -> ok
|
|
end,
|
|
lists:foreach(fun (M) -> mnesia:write(M) end, Msgs)
|
|
end
|
|
end,
|
|
mnesia:transaction(F).
|
|
|
|
pop_messages(LUser, LServer) ->
|
|
US = {LUser, LServer},
|
|
F = fun () ->
|
|
Rs = mnesia:wread({offline_msg, US}),
|
|
mnesia:delete({offline_msg, US}),
|
|
Rs
|
|
end,
|
|
case mnesia:transaction(F) of
|
|
{atomic, L} ->
|
|
{ok, lists:keysort(#offline_msg.timestamp, L)};
|
|
{aborted, Reason} ->
|
|
{error, Reason}
|
|
end.
|
|
|
|
remove_expired_messages(_LServer) ->
|
|
TimeStamp = p1_time_compat:timestamp(),
|
|
F = fun () ->
|
|
mnesia:write_lock_table(offline_msg),
|
|
mnesia:foldl(fun (Rec, _Acc) ->
|
|
case Rec#offline_msg.expire of
|
|
never -> ok;
|
|
TS ->
|
|
if TS < TimeStamp ->
|
|
mnesia:delete_object(Rec);
|
|
true -> ok
|
|
end
|
|
end
|
|
end,
|
|
ok, offline_msg)
|
|
end,
|
|
mnesia:transaction(F).
|
|
|
|
remove_old_messages(Days, _LServer) ->
|
|
S = p1_time_compat:system_time(seconds) - 60 * 60 * 24 * Days,
|
|
MegaSecs1 = S div 1000000,
|
|
Secs1 = S rem 1000000,
|
|
TimeStamp = {MegaSecs1, Secs1, 0},
|
|
F = fun () ->
|
|
mnesia:write_lock_table(offline_msg),
|
|
mnesia:foldl(fun (#offline_msg{timestamp = TS} = Rec,
|
|
_Acc)
|
|
when TS < TimeStamp ->
|
|
mnesia:delete_object(Rec);
|
|
(_Rec, _Acc) -> ok
|
|
end,
|
|
ok, offline_msg)
|
|
end,
|
|
mnesia:transaction(F).
|
|
|
|
remove_user(LUser, LServer) ->
|
|
US = {LUser, LServer},
|
|
F = fun () -> mnesia:delete({offline_msg, US}) end,
|
|
mnesia:transaction(F).
|
|
|
|
read_message_headers(LUser, LServer) ->
|
|
Msgs = mnesia:dirty_read({offline_msg, {LUser, LServer}}),
|
|
Hdrs = lists:map(
|
|
fun(#offline_msg{from = From, to = To, packet = Pkt,
|
|
timestamp = TS}) ->
|
|
Seq = now_to_integer(TS),
|
|
NewPkt = jlib:add_delay_info(Pkt, LServer, TS,
|
|
<<"Offline Storage">>),
|
|
{Seq, From, To, NewPkt}
|
|
end, Msgs),
|
|
lists:keysort(1, Hdrs).
|
|
|
|
read_message(LUser, LServer, I) ->
|
|
US = {LUser, LServer},
|
|
TS = integer_to_now(I),
|
|
case mnesia:dirty_match_object(
|
|
offline_msg, #offline_msg{us = US, timestamp = TS, _ = '_'}) of
|
|
[Msg|_] ->
|
|
{ok, Msg};
|
|
_ ->
|
|
error
|
|
end.
|
|
|
|
remove_message(LUser, LServer, I) ->
|
|
US = {LUser, LServer},
|
|
TS = integer_to_now(I),
|
|
Msgs = mnesia:dirty_match_object(
|
|
offline_msg, #offline_msg{us = US, timestamp = TS, _ = '_'}),
|
|
lists:foreach(
|
|
fun(Msg) ->
|
|
mnesia:dirty_delete_object(Msg)
|
|
end, Msgs).
|
|
|
|
read_all_messages(LUser, LServer) ->
|
|
US = {LUser, LServer},
|
|
lists:keysort(#offline_msg.timestamp,
|
|
mnesia:dirty_read({offline_msg, US})).
|
|
|
|
remove_all_messages(LUser, LServer) ->
|
|
US = {LUser, LServer},
|
|
F = fun () ->
|
|
mnesia:write_lock_table(offline_msg),
|
|
lists:foreach(fun (Msg) -> mnesia:delete_object(Msg) end,
|
|
mnesia:dirty_read({offline_msg, US}))
|
|
end,
|
|
mnesia:transaction(F).
|
|
|
|
count_messages(LUser, LServer) ->
|
|
US = {LUser, LServer},
|
|
F = fun () ->
|
|
count_mnesia_records(US)
|
|
end,
|
|
case catch mnesia:async_dirty(F) of
|
|
I when is_integer(I) -> I;
|
|
_ -> 0
|
|
end.
|
|
|
|
import(_LServer, #offline_msg{} = Msg) ->
|
|
mnesia:dirty_write(Msg).
|
|
|
|
%%%===================================================================
|
|
%%% Internal functions
|
|
%%%===================================================================
|
|
%% Return the number of records matching a given match expression.
|
|
%% This function is intended to be used inside a Mnesia transaction.
|
|
%% The count has been written to use the fewest possible memory by
|
|
%% getting the record by small increment and by using continuation.
|
|
-define(BATCHSIZE, 100).
|
|
|
|
count_mnesia_records(US) ->
|
|
MatchExpression = #offline_msg{us = US, _ = '_'},
|
|
case mnesia:select(offline_msg, [{MatchExpression, [], [[]]}],
|
|
?BATCHSIZE, read) of
|
|
{Result, Cont} ->
|
|
Count = length(Result),
|
|
count_records_cont(Cont, Count);
|
|
'$end_of_table' ->
|
|
0
|
|
end.
|
|
|
|
count_records_cont(Cont, Count) ->
|
|
case mnesia:select(Cont) of
|
|
{Result, Cont} ->
|
|
NewCount = Count + length(Result),
|
|
count_records_cont(Cont, NewCount);
|
|
'$end_of_table' ->
|
|
Count
|
|
end.
|
|
|
|
jid_to_binary(#jid{user = U, server = S, resource = R,
|
|
luser = LU, lserver = LS, lresource = LR}) ->
|
|
#jid{user = iolist_to_binary(U),
|
|
server = iolist_to_binary(S),
|
|
resource = iolist_to_binary(R),
|
|
luser = iolist_to_binary(LU),
|
|
lserver = iolist_to_binary(LS),
|
|
lresource = iolist_to_binary(LR)}.
|
|
|
|
now_to_integer({MS, S, US}) ->
|
|
(MS * 1000000 + S) * 1000000 + US.
|
|
|
|
integer_to_now(Int) ->
|
|
Secs = Int div 1000000,
|
|
USec = Int rem 1000000,
|
|
MSec = Secs div 1000000,
|
|
Sec = Secs rem 1000000,
|
|
{MSec, Sec, USec}.
|
|
|
|
update_table() ->
|
|
Fields = record_info(fields, offline_msg),
|
|
case mnesia:table_info(offline_msg, attributes) of
|
|
Fields ->
|
|
ejabberd_config:convert_table_to_binary(
|
|
offline_msg, Fields, bag,
|
|
fun(#offline_msg{us = {U, _}}) -> U end,
|
|
fun(#offline_msg{us = {U, S},
|
|
from = From,
|
|
to = To,
|
|
packet = El} = R) ->
|
|
R#offline_msg{us = {iolist_to_binary(U),
|
|
iolist_to_binary(S)},
|
|
from = jid_to_binary(From),
|
|
to = jid_to_binary(To),
|
|
packet = fxml:to_xmlel(El)}
|
|
end);
|
|
_ ->
|
|
?INFO_MSG("Recreating offline_msg table", []),
|
|
mnesia:transform_table(offline_msg, ignore, Fields)
|
|
end.
|