Packages

Typed distributed messaging for Gleam on the BEAM.

Retired package: Deprecated - The project needs to be redesigned around a much smaller and clearer core.

Current section

Files

Jump to
distribute src distribute@cluster.erl
Raw

src/distribute@cluster.erl

-module(distribute@cluster).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/distribute/cluster.gleam").
-export([classify_start_reason/1, start_node/2, connect_bool/1, classify_connect_reason/1, connect/1, nodes/0, self_node/0, ping/1, is_distributed/0, is_healthy/0, connected_count/0, health/0]).
-export_type([start_error/0, connect_error/0, dynamic_/0, cluster_health/0]).
-if(?OTP_RELEASE >= 27).
-define(MODULEDOC(Str), -moduledoc(Str)).
-define(DOC(Str), -doc(Str)).
-else.
-define(MODULEDOC(Str), -compile([])).
-define(DOC(Str), -compile([])).
-endif.
-type start_error() :: {invalid_node_name, binary()} |
already_started |
cookie_too_long |
{network_error, binary()} |
{system_error, binary()} |
{start_failed, binary()}.
-type connect_error() :: connect_timeout |
node_not_found |
{connect_network_error, binary()} |
connect_ignored.
-type dynamic_() :: any().
-type cluster_health() :: {cluster_health,
binary(),
boolean(),
list(binary()),
integer(),
list(binary()),
list(binary())}.
-file("src/distribute/cluster.gleam", 124).
?DOC(" Convert StartError to string for logging\n").
-spec classify_start_error_to_string(start_error()) -> binary().
classify_start_error_to_string(Error) ->
case Error of
{invalid_node_name, Msg} ->
<<"invalid_node_name: "/utf8, Msg/binary>>;
already_started ->
<<"already_started"/utf8>>;
cookie_too_long ->
<<"cookie_too_long"/utf8>>;
{network_error, Msg@1} ->
<<"network_error: "/utf8, Msg@1/binary>>;
{system_error, Msg@2} ->
<<"system_error: "/utf8, Msg@2/binary>>;
{start_failed, Msg@3} ->
<<"start_failed: "/utf8, Msg@3/binary>>
end.
-file("src/distribute/cluster.gleam", 136).
?DOC(" Check if error is network-related\n").
-spec is_network_error(binary()) -> boolean().
is_network_error(Reason) ->
(gleam_stdlib:contains_string(Reason, <<"network"/utf8>>) orelse gleam_stdlib:contains_string(
Reason,
<<"eaddrinuse"/utf8>>
))
orelse gleam_stdlib:contains_string(Reason, <<"connection"/utf8>>).
-file("src/distribute/cluster.gleam", 143).
?DOC(" Check if error is system/permission related\n").
-spec is_system_error(binary()) -> boolean().
is_system_error(Reason) ->
(gleam_stdlib:contains_string(Reason, <<"permission"/utf8>>) orelse gleam_stdlib:contains_string(
Reason,
<<"access"/utf8>>
))
orelse gleam_stdlib:contains_string(Reason, <<"enoent"/utf8>>).
-file("src/distribute/cluster.gleam", 106).
?DOC(" Classify error reason into structured StartError\n").
-spec classify_start_reason(binary()) -> start_error().
classify_start_reason(Reason) ->
case Reason of
<<"already_started"/utf8>> ->
already_started;
<<"invalid_node_name"/utf8>> ->
{invalid_node_name, <<"Node name must contain '@'"/utf8>>};
<<"cookie_too_long"/utf8>> ->
cookie_too_long;
_ ->
case is_network_error(Reason) of
true ->
{network_error, Reason};
false ->
case is_system_error(Reason) of
true ->
{system_error, Reason};
false ->
{start_failed, Reason}
end
end
end.
-file("src/distribute/cluster.gleam", 193).
?DOC(" Validate node name format\n").
-spec validate_node_name(binary()) -> {ok, nil} | {error, start_error()}.
validate_node_name(Name) ->
case gleam_stdlib:contains_string(Name, <<"@"/utf8>>) of
true ->
{ok, nil};
false ->
{error,
{invalid_node_name,
<<"Node name must contain '@' symbol"/utf8>>}}
end.
-file("src/distribute/cluster.gleam", 201).
?DOC(" Validate cookie length\n").
-spec validate_cookie(binary()) -> {ok, nil} | {error, start_error()}.
validate_cookie(Cookie) ->
case string:length(Cookie) > 255 of
true ->
{error, cookie_too_long};
false ->
{ok, nil}
end.
-file("src/distribute/cluster.gleam", 151).
?DOC(
" Start distributed node with the given name and cookie.\n"
" Returns Ok(Nil) on success, Error with specific failure reason.\n"
).
-spec start_node(binary(), binary()) -> {ok, nil} | {error, start_error()}.
start_node(Name, Cookie) ->
distribute@log:info(
<<"Starting distributed node"/utf8>>,
[{<<"node"/utf8>>, Name}]
),
case validate_node_name(Name) of
{error, E} ->
distribute@log:error(
<<"Node name validation failed"/utf8>>,
[{<<"node"/utf8>>, Name},
{<<"error"/utf8>>, classify_start_error_to_string(E)}]
),
{error, E};
{ok, _} ->
case validate_cookie(Cookie) of
{error, E@1} ->
distribute@log:error(
<<"Cookie validation failed"/utf8>>,
[{<<"node"/utf8>>, Name},
{<<"error"/utf8>>,
classify_start_error_to_string(E@1)}]
),
{error, E@1};
{ok, _} ->
Res = cluster_ffi:start_node(Name, Cookie),
case cluster_ffi:is_ok_atom(Res) of
true ->
distribute@log:info(
<<"Node started successfully"/utf8>>,
[{<<"node"/utf8>>, Name}]
),
{ok, nil};
false ->
Error = classify_start_reason(
cluster_ffi:get_error_reason(Res)
),
distribute@log:error(
<<"Failed to start node"/utf8>>,
[{<<"node"/utf8>>, Name},
{<<"error"/utf8>>,
classify_start_error_to_string(Error)}]
),
{error, Error}
end
end
end.
-file("src/distribute/cluster.gleam", 231).
?DOC(
" Connect to another distributed node (legacy API).\n"
" Returns True if connected successfully, False otherwise.\n"
).
-spec connect_bool(binary()) -> boolean().
connect_bool(Node) ->
cluster_ffi:is_true(cluster_ffi:connect(Node)).
-file("src/distribute/cluster.gleam", 237).
-spec validate_connect_node(binary()) -> {ok, nil} | {error, connect_error()}.
validate_connect_node(Node) ->
case gleam_stdlib:contains_string(Node, <<"@"/utf8>>) of
true ->
{ok, nil};
false ->
{error, node_not_found}
end.
-file("src/distribute/cluster.gleam", 247).
-spec classify_connect_reason(binary()) -> connect_error().
classify_connect_reason(Reason) ->
case gleam_stdlib:contains_string(Reason, <<"timeout"/utf8>>) of
true ->
connect_timeout;
false ->
{connect_network_error, Reason}
end.
-file("src/distribute/cluster.gleam", 254).
-spec classify_connect_error(dynamic_()) -> connect_error().
classify_connect_error(Value) ->
case cluster_ffi:is_ignored(Value) of
true ->
connect_ignored;
false ->
classify_connect_reason(cluster_ffi:get_error_reason(Value))
end.
-file("src/distribute/cluster.gleam", 210).
?DOC(
" Connect to another distributed node.\n"
" Returns Ok(Nil) if connected, Error with reason otherwise.\n"
).
-spec connect(binary()) -> {ok, nil} | {error, connect_error()}.
connect(Node) ->
case validate_connect_node(Node) of
{error, E} ->
{error, E};
{ok, _} ->
Result = cluster_ffi:connect(Node),
case cluster_ffi:is_true(Result) of
true ->
{ok, nil};
false ->
case cluster_ffi:is_ignored(Result) of
true ->
{error, connect_ignored};
false ->
{error, classify_connect_error(Result)}
end
end
end.
-file("src/distribute/cluster.gleam", 262).
?DOC(" Get the list of currently connected nodes.\n").
-spec nodes() -> list(binary()).
nodes() ->
cluster_ffi:nodes().
-file("src/distribute/cluster.gleam", 267).
?DOC(" Get the name of the current node.\n").
-spec self_node() -> binary().
self_node() ->
cluster_ffi:self_node().
-file("src/distribute/cluster.gleam", 273).
?DOC(
" Ping a remote node to check if it is reachable.\n"
" Returns True if the node responds with pong, False otherwise.\n"
).
-spec ping(binary()) -> boolean().
ping(Node) ->
cluster_ffi:ping(Node).
-file("src/distribute/cluster.gleam", 370).
?DOC(
" Check if this node is running as a distributed node.\n"
"\n"
" Returns True if the node was started with a name (e.g., `-name node@host`).\n"
).
-spec is_distributed() -> boolean().
is_distributed() ->
self_node() /= <<"nonode@nohost"/utf8>>.
-file("src/distribute/cluster.gleam", 404).
-spec do_list_length(list(any()), integer()) -> integer().
do_list_length(L, Acc) ->
case L of
[] ->
Acc;
[_ | Rest] ->
do_list_length(Rest, Acc + 1)
end.
-file("src/distribute/cluster.gleam", 400).
-spec list_length(list(any())) -> integer().
list_length(L) ->
do_list_length(L, 0).
-file("src/distribute/cluster.gleam", 363).
?DOC(
" Quick health check - returns True if node is distributed and has connections.\n"
"\n"
" Use this for simple health endpoints that just need a boolean.\n"
"\n"
" ## Example\n"
"\n"
" ```gleam\n"
" case cluster.is_healthy() {\n"
" True -> \"OK\"\n"
" False -> \"UNHEALTHY\"\n"
" }\n"
" ```\n"
).
-spec is_healthy() -> boolean().
is_healthy() ->
is_distributed() andalso (list_length(nodes()) > 0).
-file("src/distribute/cluster.gleam", 375).
?DOC(" Get the number of currently connected nodes.\n").
-spec connected_count() -> integer().
connected_count() ->
list_length(nodes()).
-file("src/distribute/cluster.gleam", 415).
-spec do_list_reverse(list(IAR), list(IAR)) -> list(IAR).
do_list_reverse(L, Acc) ->
case L of
[] ->
Acc;
[X | Rest] ->
do_list_reverse(Rest, [X | Acc])
end.
-file("src/distribute/cluster.gleam", 411).
-spec list_reverse(list(IAO)) -> list(IAO).
list_reverse(L) ->
do_list_reverse(L, []).
-file("src/distribute/cluster.gleam", 384).
-spec do_partition_by_ping(list(binary()), list(binary()), list(binary())) -> {list(binary()),
list(binary())}.
do_partition_by_ping(Remaining, Reachable, Unreachable) ->
case Remaining of
[] ->
{list_reverse(Reachable), list_reverse(Unreachable)};
[Node | Rest] ->
case ping(Node) of
true ->
do_partition_by_ping(Rest, [Node | Reachable], Unreachable);
false ->
do_partition_by_ping(Rest, Reachable, [Node | Unreachable])
end
end.
-file("src/distribute/cluster.gleam", 380).
-spec partition_by_ping(list(binary())) -> {list(binary()), list(binary())}.
partition_by_ping(Node_list) ->
do_partition_by_ping(Node_list, [], []).
-file("src/distribute/cluster.gleam", 317).
?DOC(
" Check the health of the cluster.\n"
"\n"
" Returns a `ClusterHealth` record with information about:\n"
" - Current node status\n"
" - Connected nodes\n"
" - Reachability of connected nodes via ping\n"
"\n"
" ## Example\n"
"\n"
" ```gleam\n"
" case cluster.health() {\n"
" Ok(status) -> {\n"
" io.println(\"Self: \" <> status.self_node)\n"
" io.println(\"Connected: \" <> int.to_string(status.connected_count))\n"
" }\n"
" Error(reason) -> io.println(\"Health check failed: \" <> reason)\n"
" }\n"
" ```\n"
).
-spec health() -> {ok, cluster_health()} | {error, binary()}.
health() ->
Self = self_node(),
Is_dist = is_distributed(),
case Is_dist of
false ->
{ok, {cluster_health, Self, false, [], 0, [], []}};
true ->
Connected = nodes(),
{Reachable, Unreachable} = partition_by_ping(Connected),
{ok,
{cluster_health,
Self,
true,
Connected,
list_length(Connected),
Reachable,
Unreachable}}
end.