| 1 |
|
-module(mod_stream_management_cets). |
| 2 |
|
-behaviour(mod_stream_management_backend). |
| 3 |
|
|
| 4 |
|
-include("mongoose.hrl"). |
| 5 |
|
-include("jlib.hrl"). |
| 6 |
|
-include_lib("stdlib/include/ms_transform.hrl"). |
| 7 |
|
|
| 8 |
|
-export([init/2, |
| 9 |
|
stop/1, |
| 10 |
|
register_smid/3, |
| 11 |
|
unregister_smid/2, |
| 12 |
|
get_sid/2]). |
| 13 |
|
|
| 14 |
|
-export([read_stale_h/2, |
| 15 |
|
write_stale_h/3, |
| 16 |
|
delete_stale_h/2]). |
| 17 |
|
|
| 18 |
|
-export([clear_table/2]). |
| 19 |
|
|
| 20 |
|
-ignore_xref([start_link/1]). |
| 21 |
|
|
| 22 |
|
-define(TABLE, cets_stream_management_session). |
| 23 |
|
-define(TABLE_H, cets_stream_management_stale_h). |
| 24 |
|
|
| 25 |
|
init(HostType, #{stale_h := StaleOpts}) -> |
| 26 |
202 |
cets:start(?TABLE, #{}), |
| 27 |
202 |
cets_discovery:add_table(mongoose_cets_discovery, ?TABLE), |
| 28 |
202 |
maybe_init_stale_h(HostType, StaleOpts), |
| 29 |
202 |
ok. |
| 30 |
|
|
| 31 |
|
stop(HostType) -> |
| 32 |
189 |
stop_cleaner(HostType). |
| 33 |
|
|
| 34 |
|
maybe_init_stale_h(HostType, StaleOpts = #{enabled := true}) -> |
| 35 |
3 |
cets:start(?TABLE_H, #{}), |
| 36 |
3 |
cets_discovery:add_table(mongoose_cets_discovery, ?TABLE_H), |
| 37 |
3 |
start_cleaner(HostType, StaleOpts); |
| 38 |
199 |
maybe_init_stale_h(_, _) -> ok. |
| 39 |
|
|
| 40 |
|
-spec register_smid(HostType, SMID, SID) -> ok when |
| 41 |
|
HostType :: mongooseim:host_type(), |
| 42 |
|
SMID :: mod_stream_management:smid(), |
| 43 |
|
SID :: ejabberd_sm:sid(). |
| 44 |
|
register_smid(_HostType, SMID, SID) -> |
| 45 |
57 |
cets:insert_many(?TABLE, [{{sid, SID}, SMID}, {{smid, SMID}, SID}]). |
| 46 |
|
|
| 47 |
|
-spec unregister_smid(mongooseim:host_type(), ejabberd_sm:sid()) -> |
| 48 |
|
{ok, SMID :: mod_stream_management:smid()} | {error, smid_not_found}. |
| 49 |
|
unregister_smid(_HostType, SID) -> |
| 50 |
149 |
case ets:lookup(?TABLE, {sid, SID}) of |
| 51 |
|
[] -> |
| 52 |
93 |
{error, smid_not_found}; |
| 53 |
|
[{_, SMID}] -> |
| 54 |
56 |
cets:delete_many(?TABLE, [{sid, SID}, {smid, SMID}]), |
| 55 |
56 |
{ok, SMID} |
| 56 |
|
end. |
| 57 |
|
|
| 58 |
|
-spec get_sid(mongooseim:host_type(), mod_stream_management:smid()) -> |
| 59 |
|
{sid, ejabberd_sm:sid()} | {error, smid_not_found}. |
| 60 |
|
get_sid(_HostType, SMID) -> |
| 61 |
24 |
case ets:lookup(?TABLE, {smid, SMID}) of |
| 62 |
|
[] -> |
| 63 |
9 |
{error, smid_not_found}; |
| 64 |
|
[{_, SID}] -> |
| 65 |
15 |
{sid, SID} |
| 66 |
|
end. |
| 67 |
|
|
| 68 |
|
%% stale_h functions |
| 69 |
|
|
| 70 |
|
-spec read_stale_h(HostType, SMID) -> |
| 71 |
|
{stale_h, non_neg_integer()} | {error, smid_not_found} when |
| 72 |
|
HostType :: mongooseim:host_type(), |
| 73 |
|
SMID :: mod_stream_management:smid(). |
| 74 |
|
read_stale_h(_HostType, SMID) -> |
| 75 |
25 |
case ets:lookup(?TABLE_H, SMID) of |
| 76 |
|
[] -> |
| 77 |
1 |
{error, smid_not_found}; |
| 78 |
|
[{_, H, _}] -> |
| 79 |
24 |
{stale_h, H} |
| 80 |
|
end. |
| 81 |
|
|
| 82 |
|
-spec write_stale_h(HostType, SMID, H) -> ok when |
| 83 |
|
HostType :: mongooseim:host_type(), |
| 84 |
|
SMID :: mod_stream_management:smid(), |
| 85 |
|
H :: non_neg_integer(). |
| 86 |
|
write_stale_h(_HostType, SMID, H) -> |
| 87 |
8 |
Stamp = erlang:monotonic_time(second), |
| 88 |
8 |
cets:insert(?TABLE_H, {SMID, H, Stamp}). |
| 89 |
|
|
| 90 |
|
-spec delete_stale_h(HostType, SMID) -> ok when |
| 91 |
|
HostType :: mongooseim:host_type(), |
| 92 |
|
SMID :: mod_stream_management:smid(). |
| 93 |
|
delete_stale_h(_HostType, SMID) -> |
| 94 |
:-( |
cets:delete(?TABLE_H, SMID). |
| 95 |
|
|
| 96 |
|
%% stale_h cleaning logic |
| 97 |
|
|
| 98 |
|
start_cleaner(HostType, #{repeat_after := Interval, geriatric := TTL}) -> |
| 99 |
|
%% TODO cleaner should be a service |
| 100 |
3 |
WOpts = #{host_type => HostType, action => fun ?MODULE:clear_table/2, |
| 101 |
|
opts => TTL, interval => timer:seconds(Interval)}, |
| 102 |
3 |
mongoose_collector:start_common(?MODULE, HostType, WOpts). |
| 103 |
|
|
| 104 |
|
stop_cleaner(HostType) -> |
| 105 |
189 |
mongoose_collector:stop_common(?MODULE, HostType). |
| 106 |
|
|
| 107 |
|
clear_table(_HostType, GeriatricAge) -> |
| 108 |
2 |
TimeToDie = erlang:monotonic_time(second) - GeriatricAge, |
| 109 |
2 |
MS = ets:fun2ms(fun({_, _, S}) when S < TimeToDie -> true end), |
| 110 |
2 |
ets:select_delete(?TABLE_H, MS). |