Packages

Self-healing clusters for Gleam applications on the BEAM!

Current section

Files

Jump to
barnacle src barnacle.erl
Raw

src/barnacle.erl

-module(barnacle).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]).
-export([with_name/2, with_poll_interval/2, with_listener/2, with_connect_nodes_function/2, with_disconnect_nodes_function/2, with_list_nodes_function/2, refresh/2, pause/2, shutdown/2, run_once/1, start/1, child_spec/2, new_strategy/1, custom/1, local_epmd/0, epmd/1, dns/2, get_node_basename/1]).
-export_type([barnacle/1, strategy/1, node_disconnect_error/0, message/1, barnacle_response/1, state/1, refresh_error/1]).
-opaque barnacle(HPL) :: {barnacle,
gleam@option:option(gleam@erlang@atom:atom_()),
strategy(HPL),
integer(),
gleam@option:option(gleam@erlang@process:subject(barnacle_response(HPL)))}.
-opaque strategy(HPM) :: {strategy,
fun(() -> {ok, list(gleam@erlang@atom:atom_())} | {error, HPM}),
fun((list(gleam@erlang@atom:atom_())) -> {ok,
list(gleam@erlang@atom:atom_())} |
{error,
list({gleam@erlang@atom:atom_(),
gleam@erlang@node:connect_error()})}),
fun((list(gleam@erlang@atom:atom_())) -> {ok,
list(gleam@erlang@atom:atom_())} |
{error, list({gleam@erlang@atom:atom_(), node_disconnect_error()})}),
fun(() -> {ok, list(gleam@erlang@atom:atom_())} | {error, HPM})}.
-type node_disconnect_error() :: failed_to_disconnect | local_node_is_not_alive.
-opaque message(HPN) :: {refresh,
gleam@option:option(gleam@erlang@process:subject({ok,
list(gleam@erlang@atom:atom_())} |
{error, refresh_error(HPN)}))} |
{pause, gleam@option:option(gleam@erlang@process:subject(nil))} |
{shutdown, gleam@option:option(gleam@erlang@process:subject(nil))}.
-type barnacle_response(HPO) :: {refresh_response,
{ok, list(gleam@erlang@atom:atom_())} | {error, refresh_error(HPO)}} |
{pause_response, nil} |
{shutdown_response, nil}.
-type state(HPP) :: {state,
gleam@erlang@process:subject(message(HPP)),
barnacle(HPP),
gleam@option:option(gleam@erlang@process:timer())}.
-type refresh_error(HPQ) :: {strategy_error, HPQ} |
{connect_error,
list({gleam@erlang@atom:atom_(), gleam@erlang@node:connect_error()})} |
{disconnect_error,
list({gleam@erlang@atom:atom_(), node_disconnect_error()})}.
-spec with_name(barnacle(HPY), binary()) -> barnacle(HPY).
with_name(Barnacle, Name) ->
erlang:setelement(
2,
Barnacle,
{some,
begin
_pipe = Name,
erlang:binary_to_atom(_pipe)
end}
).
-spec with_poll_interval(barnacle(HQB), integer()) -> barnacle(HQB).
with_poll_interval(Barnacle, Poll_interval) ->
erlang:setelement(4, Barnacle, Poll_interval).
-spec with_listener(
barnacle(HQE),
gleam@erlang@process:subject(barnacle_response(HQE))
) -> barnacle(HQE).
with_listener(Barnacle, Listener) ->
erlang:setelement(5, Barnacle, {some, Listener}).
-spec with_connect_nodes_function(
strategy(HQQ),
fun((list(gleam@erlang@atom:atom_())) -> {ok,
list(gleam@erlang@atom:atom_())} |
{error,
list({gleam@erlang@atom:atom_(), gleam@erlang@node:connect_error()})})
) -> strategy(HQQ).
with_connect_nodes_function(Strategy, Connect_nodes) ->
erlang:setelement(3, Strategy, Connect_nodes).
-spec with_disconnect_nodes_function(
strategy(HQY),
fun((list(gleam@erlang@atom:atom_())) -> {ok,
list(gleam@erlang@atom:atom_())} |
{error, list({gleam@erlang@atom:atom_(), node_disconnect_error()})})
) -> strategy(HQY).
with_disconnect_nodes_function(Strategy, Disconnect_nodes) ->
erlang:setelement(4, Strategy, Disconnect_nodes).
-spec with_list_nodes_function(
strategy(HRG),
fun(() -> {ok, list(gleam@erlang@atom:atom_())} | {error, HRG})
) -> strategy(HRG).
with_list_nodes_function(Strategy, List_nodes) ->
erlang:setelement(5, Strategy, List_nodes).
-spec list_nodes() -> {ok, list(gleam@erlang@atom:atom_())} | {error, any()}.
list_nodes() ->
_pipe = [erlang:node() | erlang:nodes()],
_pipe@1 = gleam@list:map(_pipe, fun gleam_erlang_ffi:identity/1),
{ok, _pipe@1}.
-spec refresh(gleam@erlang@process:subject(message(HSP)), integer()) -> {ok,
list(gleam@erlang@atom:atom_())} |
{error, refresh_error(HSP)}.
refresh(Subject, Timeout) ->
gleam@otp@actor:call(
Subject,
fun(Subj) -> {refresh, {some, Subj}} end,
Timeout
).
-spec pause(gleam@erlang@process:subject(message(any())), integer()) -> nil.
pause(Subject, Timeout) ->
gleam@otp@actor:call(
Subject,
fun(Subj) -> {pause, {some, Subj}} end,
Timeout
).
-spec shutdown(gleam@erlang@process:subject(message(any())), integer()) -> nil.
shutdown(Subject, Timeout) ->
gleam@otp@actor:call(
Subject,
fun(Subj) -> {shutdown, {some, Subj}} end,
Timeout
).
-spec send_response(gleam@option:option(gleam@erlang@process:subject(HTM)), HTM) -> nil.
send_response(Maybe_client, Response) ->
gleam@option:map(
Maybe_client,
fun(_capture) -> gleam@erlang@process:send(_capture, Response) end
),
nil.
-spec cancel_timer(gleam@option:option(gleam@erlang@process:timer())) -> nil.
cancel_timer(Timer) ->
gleam@option:map(Timer, fun gleam@erlang@process:cancel_timer/1),
nil.
-spec refresh_nodes(barnacle(HTQ)) -> {ok, list(gleam@erlang@atom:atom_())} |
{error, refresh_error(HTQ)}.
refresh_nodes(Barnacle) ->
gleam@result:'try'(
begin
_pipe = (erlang:element(2, erlang:element(3, Barnacle)))(),
gleam@result:map_error(
_pipe,
fun(Field@0) -> {strategy_error, Field@0} end
)
end,
fun(Available_nodes_list) ->
gleam@result:'try'(
begin
_pipe@1 = (erlang:element(5, erlang:element(3, Barnacle)))(),
gleam@result:map_error(
_pipe@1,
fun(Field@0) -> {strategy_error, Field@0} end
)
end,
fun(Current_nodes_list) ->
Self = begin
_pipe@2 = erlang:node(),
gleam_erlang_ffi:identity(_pipe@2)
end,
Available_nodes = begin
_pipe@3 = Available_nodes_list,
_pipe@4 = gleam@set:from_list(_pipe@3),
gleam@set:delete(_pipe@4, Self)
end,
Current_nodes = begin
_pipe@5 = Current_nodes_list,
_pipe@6 = gleam@set:from_list(_pipe@5),
gleam@set:delete(_pipe@6, Self)
end,
Nodes_to_add = gleam@set:difference(
Available_nodes,
Current_nodes
),
Nodes_to_remove = gleam@set:difference(
Current_nodes,
Available_nodes
),
Connect_results = (erlang:element(
3,
erlang:element(3, Barnacle)
))(
begin
_pipe@7 = Nodes_to_add,
gleam@set:to_list(_pipe@7)
end
),
gleam@result:'try'(
begin
_pipe@8 = Connect_results,
gleam@result:map_error(
_pipe@8,
fun(Field@0) -> {connect_error, Field@0} end
)
end,
fun(_) ->
Disconnect_results = (erlang:element(
4,
erlang:element(3, Barnacle)
))(
begin
_pipe@9 = Nodes_to_remove,
gleam@set:to_list(_pipe@9)
end
),
gleam@result:'try'(
begin
_pipe@10 = Disconnect_results,
gleam@result:map_error(
_pipe@10,
fun(Field@0) -> {disconnect_error, Field@0} end
)
end,
fun(_) ->
_pipe@11 = (erlang:element(
5,
erlang:element(3, Barnacle)
))(),
gleam@result:map_error(
_pipe@11,
fun(Field@0) -> {strategy_error, Field@0} end
)
end
)
end
)
end
)
end
).
-spec run_once(barnacle(HSM)) -> {ok, list(gleam@erlang@atom:atom_())} |
{error, refresh_error(HSM)}.
run_once(Barnacle) ->
_pipe = Barnacle,
refresh_nodes(_pipe).
-spec handle_message(message(HTB), state(HTB)) -> gleam@otp@actor:next(any(), state(HTB)).
handle_message(Message, State) ->
{state, Self, Barnacle, Timer} = State,
case Message of
{refresh, Return} ->
Refresh_result = refresh_nodes(Barnacle),
cancel_timer(Timer),
Timer@1 = gleam@erlang@process:send_after(
Self,
erlang:element(4, Barnacle),
{refresh, none}
),
send_response(Return, Refresh_result),
send_response(
erlang:element(5, Barnacle),
{refresh_response, Refresh_result}
),
gleam@otp@actor:continue(
erlang:setelement(4, State, {some, Timer@1})
);
{pause, Return@1} ->
cancel_timer(Timer),
send_response(Return@1, nil),
send_response(erlang:element(5, Barnacle), {pause_response, nil}),
gleam@otp@actor:continue(erlang:setelement(4, State, none));
{shutdown, Return@2} ->
cancel_timer(Timer),
send_response(Return@2, nil),
send_response(erlang:element(5, Barnacle), {shutdown_response, nil}),
{stop, normal}
end.
-spec spec(
barnacle(HTF),
gleam@option:option(gleam@erlang@process:subject(gleam@erlang@process:subject(message(HTF))))
) -> gleam@otp@actor:spec(state(HTF), message(HTF)).
spec(Barnacle, Parent) ->
{spec,
fun() ->
gleam@option:map(
erlang:element(2, Barnacle),
fun(_capture) ->
gleam_erlang_ffi:register_process(erlang:self(), _capture)
end
),
Self = gleam@erlang@process:new_subject(),
Selector = begin
_pipe = gleam_erlang_ffi:new_selector(),
gleam@erlang@process:selecting(
_pipe,
Self,
fun gleam@function:identity/1
)
end,
gleam@option:map(
Parent,
fun(_capture@1) ->
gleam@erlang@process:send(_capture@1, Self)
end
),
Timer = gleam@erlang@process:send_after(
Self,
erlang:element(4, Barnacle),
{refresh, none}
),
{ready, {state, Self, Barnacle, {some, Timer}}, Selector}
end,
10000,
fun handle_message/2}.
-spec start(barnacle(HSD)) -> {ok, gleam@erlang@process:subject(message(HSD))} |
{error, gleam@otp@actor:start_error()}.
start(Barnacle) ->
_pipe = Barnacle,
_pipe@1 = spec(_pipe, none),
gleam@otp@actor:start_spec(_pipe@1).
-spec child_spec(
barnacle(HSG),
gleam@erlang@process:subject(gleam@erlang@process:subject(message(HSG)))
) -> gleam@otp@supervisor:child_spec(message(HSG), IDI, IDI).
child_spec(Barnacle, Parent) ->
gleam@otp@supervisor:worker(fun(_) -> _pipe = Barnacle,
_pipe@1 = spec(_pipe, {some, Parent}),
gleam@otp@actor:start_spec(_pipe@1) end).
-spec result_apply(list({ok, HTT} | {error, HTU})) -> {ok, list(HTT)} |
{error, list(HTU)}.
result_apply(Results) ->
case gleam@result:partition(Results) of
{Vals, []} ->
{ok, Vals};
{_, Errs} ->
{error, Errs}
end.
-spec connect_nodes(list(gleam@erlang@atom:atom_())) -> {ok,
list(gleam@erlang@atom:atom_())} |
{error,
list({gleam@erlang@atom:atom_(), gleam@erlang@node:connect_error()})}.
connect_nodes(Nodes) ->
_pipe = Nodes,
_pipe@1 = gleam@list:map(
_pipe,
fun(Node) -> case gleam_erlang_ffi:connect_node(Node) of
{ok, _} ->
{ok, Node};
{error, Err} ->
{error, {Node, Err}}
end end
),
result_apply(_pipe@1).
-spec disconnect_nodes(list(gleam@erlang@atom:atom_())) -> {ok,
list(gleam@erlang@atom:atom_())} |
{error, list({gleam@erlang@atom:atom_(), node_disconnect_error()})}.
disconnect_nodes(Nodes) ->
_pipe = Nodes,
_pipe@1 = gleam@list:map(
_pipe,
fun(Node) -> case barnacle_ffi:disconnect_from_node(Node) of
{ok, _} ->
{ok, Node};
{error, Err} ->
{error, {Node, Err}}
end end
),
result_apply(_pipe@1).
-spec default_strategy() -> strategy(any()).
default_strategy() ->
{strategy,
fun() -> {ok, []} end,
fun connect_nodes/1,
fun disconnect_nodes/1,
fun list_nodes/0}.
-spec default_barnacle() -> barnacle(any()).
default_barnacle() ->
{barnacle, none, default_strategy(), 5000, none}.
-spec new_strategy(
fun(() -> {ok, list(gleam@erlang@atom:atom_())} | {error, HQM})
) -> strategy(HQM).
new_strategy(Discover_nodes) ->
erlang:setelement(2, default_strategy(), Discover_nodes).
-spec custom(strategy(HRW)) -> barnacle(HRW).
custom(Strategy) ->
erlang:setelement(3, default_barnacle(), Strategy).
-spec local_epmd() -> barnacle(nil).
local_epmd() ->
erlang:setelement(
3,
default_barnacle(),
erlang:setelement(
2,
default_strategy(),
fun barnacle@internal@local_epmd:discover_nodes/0
)
).
-spec epmd(list(gleam@erlang@atom:atom_())) -> barnacle(nil).
epmd(Nodes) ->
erlang:setelement(
3,
default_barnacle(),
erlang:setelement(2, default_strategy(), fun() -> {ok, Nodes} end)
).
-spec dns(binary(), binary()) -> barnacle(nil).
dns(Basename, Hostname_query) ->
erlang:setelement(
3,
default_barnacle(),
erlang:setelement(
2,
default_strategy(),
fun() ->
barnacle@internal@dns:discover_nodes(Basename, Hostname_query)
end
)
).
-spec get_node_basename(gleam@erlang@node:node_()) -> {ok, binary()} |
{error, nil}.
get_node_basename(Node) ->
_pipe = Node,
_pipe@1 = gleam_erlang_ffi:identity(_pipe),
_pipe@2 = erlang:atom_to_binary(_pipe@1),
_pipe@3 = gleam@string:split_once(_pipe@2, <<"@"/utf8>>),
gleam@result:map(_pipe@3, fun(Tuple) -> erlang:element(1, Tuple) end).