Packages
ejabberd
20.4.0
26.4.0
26.3.0
26.2.0
26.1.0
25.10.0
25.8.0
25.7.0
25.4.0
25.3.0
24.12.0
24.10.0
24.7.0
24.6.0
24.2.6
23.10.0
23.4.0
23.1.0
22.10.0
22.5.0
21.12.0
21.7.0
21.4.0
21.1.0
20.12.0
20.7.0
20.4.0
20.3.0
20.2.0
20.1.0
19.9.1
19.9.0
19.8.0
19.5.0
19.2.0
18.12.1
18.12.0
18.6.0
18.4.0
18.3.0
18.1.0
17.11.0
17.9.0
17.6.0
17.3.0
17.1.0
16.12.0-beta1
16.9.0
16.8.0
16.6.2
16.6.1
16.6.0
16.4.1
16.4.0
16.3.0
16.2.0
16.1.0-beta1
Robust, Ubiquitous and Massively Scalable Messaging Platform (XMPP, MQTT, SIP Server)
Current section
Files
Jump to
Current section
Files
src/ejabberd_router_fluux.erl
%%%-------------------------------------------------------------------
%%% Created : 11 Mar 2020 by Mickaël Rémond <mremond@process-one.net>
%%%
%%%
%%% ejabberd, Copyright (C) 2002-2020 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.
%%%
%%%-------------------------------------------------------------------
%%%
%%% To use this routing module, you can replace mnesia as the default
%%% routing back-end.
%%% Simply set the router_db_type in your ejabberd.yml file:
%%%
%%% router_db_type: fluux
%%%
%%% TODO:
%%% - Remove the need to use Mnesia ? By storing this in the process state.
-module(ejabberd_router_fluux).
-behaviour(ejabberd_router).
-behaviour(gen_server).
%% API
-export([init/0, register_route/5, unregister_route/3, find_routes/1,
get_all_routes/0, use_cache/0]).
%% gen_server callbacks
-export([init/1, handle_cast/2, handle_call/3, handle_info/2,
terminate/2, code_change/3, start_link/0]).
-include("ejabberd_router.hrl").
-include("logger.hrl").
-include_lib("stdlib/include/ms_transform.hrl").
-record(state, {}).
%%%===================================================================
%%% API
%%%===================================================================
-spec init() -> ok | {error, any()}.
init() ->
Spec = {?MODULE, {?MODULE, start_link, []},
transient, 5000, worker, [?MODULE]},
case supervisor:start_child(ejabberd_backend_sup, Spec) of
{ok, _Pid} -> ok;
Err -> Err
end.
-spec start_link() -> {ok, pid()} | {error, any()}.
start_link() ->
gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
use_cache() ->
false.
%% ServerHost is the main XMPP server domain
register_route(Domain, ServerHost, LocalHint, undefined, Pid) ->
?DEBUG("Register route 1: ~p ~p ~p ~p~n", [Domain, ServerHost, LocalHint, Pid]),
ShortDomain = shortdomain(Domain, ServerHost),
F = fun () ->
mnesia:write(#route{domain = ShortDomain,
pid = Pid,
server_host = ServerHost,
local_hint = LocalHint})
end,
transaction(F);
register_route(Domain, ServerHost, LocalHint, N, Pid) ->
?DEBUG("Register route 2: ~p ~p ~p ~p ~p~n", [Domain, ServerHost, LocalHint, N, Pid]),
ShortDomain = shortdomain(Domain, ServerHost),
F = fun () ->
case mnesia:wread({route, ShortDomain}) of
[] ->
mnesia:write(#route{domain = ShortDomain,
server_host = ServerHost,
pid = Pid,
local_hint = 1}),
lists:foreach(
fun (I) ->
mnesia:write(
#route{domain = ShortDomain,
pid = undefined,
server_host = ServerHost,
local_hint = I})
end,
lists:seq(2, N));
Rs ->
lists:any(
fun (#route{pid = undefined,
local_hint = I} = R) ->
mnesia:write(
#route{domain = ShortDomain,
pid = Pid,
server_host = ServerHost,
local_hint = I}),
mnesia:delete_object(R),
true;
(_) -> false
end,
Rs)
end
end,
transaction(F).
shortdomain(Domain) ->
[ShortDomain|Rest] = string:split(Domain, ".", trailing),
ServerDomain = case Rest of
[] -> ShortDomain;
[SD] -> SD
end,
[ShortDomain, ServerDomain].
shortdomain(Domain, ServerHost) ->
[ShortName|_] = case Domain of
ServerHost -> [<<"">>];
_ ->
string:split(Domain, [<<".">>,<<"localhost">>], trailing)
end,
ShortName.
unregister_route(Domain, undefined, Pid) ->
[ShortDomain, _] = shortdomain(Domain),
F = fun () ->
case mnesia:select(
route,
ets:fun2ms(
fun(#route{domain = D, pid = P} = R)
when D == ShortDomain, P == Pid -> R
end)) of
[R] -> mnesia:delete_object(R);
_ -> ok
end
end,
transaction(F);
unregister_route(Domain, _, Pid) ->
[ShortDomain, _] = shortdomain(Domain),
F = fun () ->
case mnesia:select(
route,
ets:fun2ms(
fun(#route{domain = D, pid = P} = R)
when D == ShortDomain, P == Pid -> R
end)) of
[R] ->
I = R#route.local_hint,
ServerHost = R#route.server_host,
mnesia:write(#route{domain = ShortDomain,
server_host = ServerHost,
pid = undefined,
local_hint = I}),
mnesia:delete_object(R);
_ -> ok
end
end,
transaction(F).
rewrite_domain(Routes, ServerDomain) ->
lists:map(fun(#route{domain = SD} = Route) when SD == ServerDomain ->
Route;
(Route) ->
Route#route{domain = unicode:characters_to_binary([<<".">>, ServerDomain], utf8)}
end, Routes).
find_routes(Domain) ->
[ShortDomain,ServerDomain] = shortdomain(Domain),
Routes = mnesia:dirty_read(route, ShortDomain),
{ok, rewrite_domain(Routes, ServerDomain)}.
get_all_routes() ->
{ok, mnesia:dirty_select(
route,
ets:fun2ms(
fun(#route{domain = Domain, server_host = ServerHost})
when Domain /= ServerHost -> Domain
end))}.
%%%===================================================================
%%% gen_server callbacks
%%%===================================================================
init([]) ->
update_tables(),
ejabberd_mnesia:create(?MODULE, route,
[{ram_copies, [node()]},
{type, bag},
{attributes, record_info(fields, route)}]),
mnesia:subscribe({table, route, simple}),
lists:foreach(
fun (Pid) -> erlang:monitor(process, Pid) end,
mnesia:dirty_select(
route,
ets:fun2ms(
fun(#route{pid = Pid}) -> Pid end))),
{ok, #state{}}.
handle_call(Request, From, State) ->
?WARNING_MSG("Unexpected call from ~p: ~p", [From, Request]),
{noreply, State}.
handle_cast(Msg, State) ->
?WARNING_MSG("Unexpected cast: ~p", [Msg]),
{noreply, State}.
handle_info({mnesia_table_event,
{write, #route{pid = Pid}, _ActivityId}}, State) ->
erlang:monitor(process, Pid),
{noreply, State};
handle_info({mnesia_table_event, _}, State) ->
{noreply, State};
handle_info({'DOWN', _Ref, _Type, Pid, _Info}, State) ->
F = fun () ->
Es = mnesia:select(
route,
ets:fun2ms(
fun(#route{pid = P} = E)
when P == Pid -> E
end)),
lists:foreach(
fun(E) ->
if is_integer(E#route.local_hint) ->
LDomain = E#route.domain,
I = E#route.local_hint,
ServerHost = E#route.server_host,
mnesia:write(#route{domain = LDomain,
server_host = ServerHost,
pid = undefined,
local_hint = I}),
mnesia:delete_object(E);
true ->
mnesia:delete_object(E)
end
end, Es)
end,
transaction(F),
{noreply, State};
handle_info(Info, State) ->
?ERROR_MSG("Unexpected info: ~p", [Info]),
{noreply, State}.
terminate(_Reason, _State) ->
ok.
code_change(_OldVsn, State, _Extra) ->
{ok, State}.
%%%===================================================================
%%% Internal functions
%%%===================================================================
transaction(F) ->
case mnesia:transaction(F) of
{atomic, _} ->
ok;
{aborted, Reason} ->
?ERROR_MSG("Mnesia transaction failed: ~p", [Reason]),
{error, db_failure}
end.
-spec update_tables() -> ok.
update_tables() ->
try
mnesia:transform_table(route, ignore, record_info(fields, route))
catch exit:{aborted, {no_exists, _}} ->
ok
end,
case lists:member(local_route, mnesia:system_info(tables)) of
true -> mnesia:delete_table(local_route);
false -> ok
end.