mirror of
https://github.com/processone/ejabberd.git
synced 2024-12-22 17:28:25 +01:00
214 lines
6.7 KiB
Erlang
214 lines
6.7 KiB
Erlang
%%%-------------------------------------------------------------------
|
|
%%% @author Evgeny Khramtsov <ekhramtsov@process-one.net>
|
|
%%% Created : 5 Jul 2017 by Evgeny Khramtsov <ekhramtsov@process-one.net>
|
|
%%%
|
|
%%%
|
|
%%% ejabberd, Copyright (C) 2002-2018 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(ejabberd_cluster).
|
|
-behaviour(ejabberd_config).
|
|
-behaviour(gen_server).
|
|
|
|
%% API
|
|
-export([start_link/0, call/4, multicall/3, multicall/4, eval_everywhere/3,
|
|
eval_everywhere/4]).
|
|
%% Backend dependent API
|
|
-export([get_nodes/0, get_known_nodes/0, join/1, leave/1, subscribe/0,
|
|
subscribe/1, node_id/0, get_node_by_id/1, send/2, wait_for_sync/1]).
|
|
%% gen_server callbacks
|
|
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
|
|
terminate/2, code_change/3]).
|
|
-export([opt_type/1]).
|
|
|
|
-include("logger.hrl").
|
|
|
|
-type dst() :: pid() | atom() | {atom(), node()}.
|
|
|
|
-callback init() -> ok | {error, any()}.
|
|
-callback get_nodes() -> [node()].
|
|
-callback get_known_nodes() -> [node()].
|
|
-callback join(node()) -> ok | {error, any()}.
|
|
-callback leave(node()) -> ok | {error, any()}.
|
|
-callback node_id() -> binary().
|
|
-callback get_node_by_id(binary()) -> node().
|
|
-callback send({atom(), node()}, term()) -> boolean().
|
|
-callback wait_for_sync(timeout()) -> ok | {error, any()}.
|
|
-callback subscribe(dst()) -> ok.
|
|
|
|
-record(state, {}).
|
|
|
|
%%%===================================================================
|
|
%%% API
|
|
%%%===================================================================
|
|
start_link() ->
|
|
gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
|
|
|
|
-spec call(node(), module(), atom(), [any()]) -> any().
|
|
call(Node, Module, Function, Args) ->
|
|
rpc:call(Node, Module, Function, Args, rpc_timeout()).
|
|
|
|
-spec multicall(module(), atom(), [any()]) -> {list(), [node()]}.
|
|
multicall(Module, Function, Args) ->
|
|
multicall(get_nodes(), Module, Function, Args).
|
|
|
|
-spec multicall([node()], module(), atom(), list()) -> {list(), [node()]}.
|
|
multicall(Nodes, Module, Function, Args) ->
|
|
rpc:multicall(Nodes, Module, Function, Args, rpc_timeout()).
|
|
|
|
-spec eval_everywhere(module(), atom(), [any()]) -> ok.
|
|
eval_everywhere(Module, Function, Args) ->
|
|
eval_everywhere(get_nodes(), Module, Function, Args),
|
|
ok.
|
|
|
|
-spec eval_everywhere([node()], module(), atom(), [any()]) -> ok.
|
|
eval_everywhere(Nodes, Module, Function, Args) ->
|
|
rpc:eval_everywhere(Nodes, Module, Function, Args),
|
|
ok.
|
|
|
|
%%%===================================================================
|
|
%%% Backend dependent API
|
|
%%%===================================================================
|
|
-spec get_nodes() -> [node()].
|
|
get_nodes() ->
|
|
Mod = get_mod(),
|
|
Mod:get_nodes().
|
|
|
|
-spec get_known_nodes() -> [node()].
|
|
get_known_nodes() ->
|
|
Mod = get_mod(),
|
|
Mod:get_known_nodes().
|
|
|
|
-spec join(node()) -> ok | {error, any()}.
|
|
join(Node) ->
|
|
Mod = get_mod(),
|
|
Mod:join(Node).
|
|
|
|
-spec leave(node()) -> ok | {error, any()}.
|
|
leave(Node) ->
|
|
Mod = get_mod(),
|
|
Mod:leave(Node).
|
|
|
|
-spec node_id() -> binary().
|
|
node_id() ->
|
|
Mod = get_mod(),
|
|
Mod:node_id().
|
|
|
|
-spec get_node_by_id(binary()) -> node().
|
|
get_node_by_id(ID) ->
|
|
Mod = get_mod(),
|
|
Mod:get_node_by_id(ID).
|
|
|
|
%% Note that false positive returns are possible, while false negatives are not.
|
|
%% In other words: positive return value (i.e. 'true') doesn't guarantee
|
|
%% successful delivery, while negative return value ('false') means
|
|
%% the delivery has definitely failed.
|
|
-spec send(dst(), term()) -> boolean().
|
|
send({Name, Node}, Msg) when Node == node() ->
|
|
send(Name, Msg);
|
|
send(undefined, _Msg) ->
|
|
false;
|
|
send(Name, Msg) when is_atom(Name) ->
|
|
send(whereis(Name), Msg);
|
|
send(Pid, Msg) when is_pid(Pid) andalso node(Pid) == node() ->
|
|
case erlang:is_process_alive(Pid) of
|
|
true ->
|
|
erlang:send(Pid, Msg),
|
|
true;
|
|
false ->
|
|
false
|
|
end;
|
|
send(Dst, Msg) ->
|
|
Mod = get_mod(),
|
|
Mod:send(Dst, Msg).
|
|
|
|
-spec wait_for_sync(timeout()) -> ok | {error, any()}.
|
|
wait_for_sync(Timeout) ->
|
|
Mod = get_mod(),
|
|
Mod:wait_for_sync(Timeout).
|
|
|
|
-spec subscribe() -> ok.
|
|
subscribe() ->
|
|
subscribe(self()).
|
|
|
|
-spec subscribe(dst()) -> ok.
|
|
subscribe(Proc) ->
|
|
Mod = get_mod(),
|
|
Mod:subscribe(Proc).
|
|
|
|
%%%===================================================================
|
|
%%% gen_server API
|
|
%%%===================================================================
|
|
init([]) ->
|
|
Ticktime = ejabberd_config:get_option(net_ticktime, 60),
|
|
Nodes = ejabberd_config:get_option(cluster_nodes, []),
|
|
net_kernel:set_net_ticktime(Ticktime),
|
|
lists:foreach(fun(Node) ->
|
|
net_kernel:connect_node(Node)
|
|
end, Nodes),
|
|
Mod = get_mod(),
|
|
case Mod:init() of
|
|
ok ->
|
|
Mod:subscribe(?MODULE),
|
|
{ok, #state{}};
|
|
{error, Reason} ->
|
|
{stop, Reason}
|
|
end.
|
|
|
|
handle_call(_Request, _From, State) ->
|
|
Reply = ok,
|
|
{reply, Reply, State}.
|
|
|
|
handle_cast(_Msg, State) ->
|
|
{noreply, State}.
|
|
|
|
handle_info({node_up, Node}, State) ->
|
|
?INFO_MSG("Node ~s has joined", [Node]),
|
|
{noreply, State};
|
|
handle_info({node_down, Node}, State) ->
|
|
?INFO_MSG("Node ~s has left", [Node]),
|
|
{noreply, State};
|
|
handle_info(_Info, State) ->
|
|
{noreply, State}.
|
|
|
|
terminate(_Reason, _State) ->
|
|
ok.
|
|
|
|
code_change(_OldVsn, State, _Extra) ->
|
|
{ok, State}.
|
|
|
|
%%%===================================================================
|
|
%%% Internal functions
|
|
%%%===================================================================
|
|
get_mod() ->
|
|
Backend = ejabberd_config:get_option(cluster_backend, mnesia),
|
|
list_to_atom("ejabberd_cluster_" ++ atom_to_list(Backend)).
|
|
|
|
rpc_timeout() ->
|
|
timer:seconds(ejabberd_config:get_option(rpc_timeout, 5)).
|
|
|
|
opt_type(net_ticktime) ->
|
|
fun (P) when is_integer(P), P > 0 -> P end;
|
|
opt_type(cluster_nodes) ->
|
|
fun (Ns) -> true = lists:all(fun is_atom/1, Ns), Ns end;
|
|
opt_type(rpc_timeout) ->
|
|
fun (T) when is_integer(T), T > 0 -> T end;
|
|
opt_type(cluster_backend) ->
|
|
fun (T) -> ejabberd_config:v_db(?MODULE, T) end;
|
|
opt_type(_) ->
|
|
[rpc_timeout, cluster_backend, cluster_nodes, net_ticktime].
|