-module(distribute@messaging). -compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]). -define(FILEPATH, "src/distribute/messaging.gleam"). -export([encode_error_to_send_error/1, classify_send_error_to_string/1, is_network_error/1, classify_send_error/2, send/2, send_typed/3, send_global/2, send_global_typed/3, send_batch_typed/2, send_batch/1, send_batch_strict/1]). -export_type([send_error/0, dynamic_/0, batch_send_result/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 send_error() :: {name_not_found, binary()} | process_not_alive | {network_error, binary()} | {invalid_message, binary()} | {encode_failed, distribute@codec:encode_error()} | {send_failed, binary()}. -type dynamic_() :: any(). -type batch_send_result() :: {batch_send_result, integer(), integer(), integer(), list(send_error())}. -file("src/distribute/messaging.gleam", 64). ?DOC(" Convert encode error to send error\n"). -spec encode_error_to_send_error(distribute@codec:encode_error()) -> send_error(). encode_error_to_send_error(Error) -> {encode_failed, Error}. -file("src/distribute/messaging.gleam", 69). ?DOC(" Convert SendError to string for logging\n"). -spec classify_send_error_to_string(send_error()) -> binary(). classify_send_error_to_string(Error) -> case Error of {name_not_found, Name} -> <<"name_not_found: "/utf8, Name/binary>>; process_not_alive -> <<"process_not_alive"/utf8>>; {network_error, Name@1} -> <<"network_error: "/utf8, Name@1/binary>>; {invalid_message, Name@2} -> <<"invalid_message: "/utf8, Name@2/binary>>; {encode_failed, _} -> <<"encode_failed"/utf8>>; {send_failed, Reason} -> <<"send_failed: "/utf8, Reason/binary>> end. -file("src/distribute/messaging.gleam", 81). ?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, <<"connection"/utf8>> )) orelse gleam_stdlib:contains_string(Reason, <<"partition"/utf8>>). -file("src/distribute/messaging.gleam", 50). ?DOC(" Classify error reason into structured SendError\n"). -spec classify_send_error(binary(), binary()) -> send_error(). classify_send_error(Reason, Name) -> case Reason of <<"not_found"/utf8>> -> {name_not_found, Name}; <<"process_not_alive"/utf8>> -> process_not_alive; <<"invalid_message"/utf8>> -> {invalid_message, Name}; _ -> case is_network_error(Reason) of true -> {network_error, Name}; false -> {send_failed, Reason} end end. -file("src/distribute/messaging.gleam", 91). ?DOC( " Send a message to a process (local or remote).\n" "\n" " This function bypasses all type checking and encoding validation.\n" ). -spec send(gleam@erlang@process:pid_(), any()) -> nil. send(Pid, Msg) -> erlang:send(Pid, Msg), nil. -file("src/distribute/messaging.gleam", 102). ?DOC( " Send a typed message to a subject. The message is encoded using the\n" " provided encoder and sent as binary data to ensure type safety.\n" ). -spec send_typed( gleam@erlang@process:subject(bitstring()), ICK, fun((ICK) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}) ) -> {ok, nil} | {error, send_error()}. send_typed(Subject, Msg, Encoder) -> case distribute@codec:encode(Encoder, Msg) of {ok, Binary_msg} -> gleam@erlang@process:send(Subject, Binary_msg), {ok, nil}; {error, Encode_error} -> {error, {encode_failed, Encode_error}} end. -file("src/distribute/messaging.gleam", 121). ?DOC( " Send a message to a globally registered name.\n" " Returns Ok(Nil) if successful, Error if name not found or send failed.\n" "\n" " This function bypasses all type checking and encoding validation.\n" ). -spec send_global(binary(), any()) -> {ok, nil} | {error, send_error()}. send_global(Name, Msg) -> distribute@log:debug( <<"Sending message to global name"/utf8>>, [{<<"name"/utf8>>, Name}] ), Res = messaging_ffi:send_global(Name, Msg), case messaging_ffi:is_ok_atom(Res) of true -> distribute@log:debug( <<"Message sent successfully"/utf8>>, [{<<"name"/utf8>>, Name}] ), {ok, nil}; false -> case messaging_ffi:is_not_found(Res) of true -> Error = {name_not_found, Name}, distribute@log:warn( <<"Global name not found"/utf8>>, [{<<"name"/utf8>>, Name}, {<<"error"/utf8>>, classify_send_error_to_string(Error)}] ), {error, Error}; false -> Error@1 = classify_send_error( messaging_ffi:get_error_reason(Res), Name ), distribute@log:error( <<"Failed to send message"/utf8>>, [{<<"name"/utf8>>, Name}, {<<"error"/utf8>>, classify_send_error_to_string(Error@1)}] ), {error, Error@1} end end. -file("src/distribute/messaging.gleam", 185). ?DOC( " Send a typed message to a globally registered name. The message is\n" " encoded using the provided encoder before sending.\n" ). -spec send_global_typed( binary(), ICV, fun((ICV) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}) ) -> {ok, nil} | {error, send_error()}. send_global_typed(Name, Msg, Encoder) -> distribute@log:debug( <<"Sending typed message to global name"/utf8>>, [{<<"name"/utf8>>, Name}] ), case distribute@codec:encode(Encoder, Msg) of {ok, Binary_msg} -> Res = messaging_ffi:send_binary_global(Name, Binary_msg), case messaging_ffi:is_ok_atom(Res) of true -> distribute@log:debug( <<"Typed message sent successfully"/utf8>>, [{<<"name"/utf8>>, Name}] ), {ok, nil}; false -> case messaging_ffi:is_not_found(Res) of true -> Error = {name_not_found, Name}, distribute@log:warn( <<"Global name not found"/utf8>>, [{<<"name"/utf8>>, Name}, {<<"error"/utf8>>, classify_send_error_to_string(Error)}] ), {error, Error}; false -> Error@1 = classify_send_error( messaging_ffi:get_error_reason(Res), Name ), distribute@log:error( <<"Failed to send typed message"/utf8>>, [{<<"name"/utf8>>, Name}, {<<"error"/utf8>>, classify_send_error_to_string(Error@1)}] ), {error, Error@1} end end; {error, Encode_error} -> {error, {encode_failed, Encode_error}} end. -file("src/distribute/messaging.gleam", 225). ?DOC(" Send multiple typed messages in batch with error aggregation\n"). -spec send_batch_typed( list({binary(), ICZ}), fun((ICZ) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}) ) -> batch_send_result(). send_batch_typed(Messages, Encoder) -> distribute@log:debug( <<"Sending batch of typed messages"/utf8>>, [{<<"count"/utf8>>, gleam@string:inspect(erlang:length(Messages))}] ), Results = gleam@list:map( Messages, fun(Msg_pair) -> {Name, Msg} = Msg_pair, send_global_typed(Name, Msg, Encoder) end ), Successes = begin _pipe = gleam@list:filter(Results, fun(R) -> case R of {ok, _} -> true; {error, _} -> false end end), erlang:length(_pipe) end, Errors = gleam@list:fold(Results, [], fun(Acc, R@1) -> case R@1 of {error, E} -> [E | Acc]; {ok, _} -> Acc end end), Total = erlang:length(Messages), distribute@log:debug( <<"Batch typed send completed"/utf8>>, [{<<"total"/utf8>>, gleam@string:inspect(Total)}, {<<"successful"/utf8>>, gleam@string:inspect(Successes)}, {<<"failed"/utf8>>, gleam@string:inspect(erlang:length(Errors))}] ), {batch_send_result, Total, Successes, erlang:length(Errors), Errors}. -file("src/distribute/messaging.gleam", 164). ?DOC( " Send multiple messages in batch with error aggregation\n" " Returns detailed results including all errors encountered\n" ). -spec send_batch(list({binary(), binary()})) -> batch_send_result(). send_batch(Messages) -> send_batch_typed(Messages, distribute@codec:string_encoder()). -file("src/distribute/messaging.gleam", 169). ?DOC(" Send batch and return Ok only if all succeeded, Error with first failure otherwise\n"). -spec send_batch_strict(list({binary(), binary()})) -> {ok, nil} | {error, send_error()}. send_batch_strict(Messages) -> Result = send_batch_typed(Messages, distribute@codec:string_encoder()), case erlang:element(4, Result) of 0 -> {ok, nil}; _ -> case gleam@list:first(erlang:element(5, Result)) of {ok, E} -> {error, E}; {error, _} -> {error, {send_failed, <<"Unknown batch error"/utf8>>}} end end.