distribute

-module(distribute@global).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/distribute/global.gleam").
-export([new/2, from_pid/3, from_subject/3, from_name/4, subject/1, owner/1, encoder/1, decoder/1, send/2, 'receive'/2, call/4, reply/3, call_error_to_string/1]).
-export_type([global_subject/1, call_error/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.

-opaque global_subject(FZR) :: {global_subject,
        gleam@erlang@process:subject(bitstring()),
        fun((FZR) -> {ok, bitstring()} |
            {error, distribute@codec:encode_error()}),
        fun((bitstring()) -> {ok, FZR} |
            {error, distribute@codec:decode_error()})}.

-type call_error() :: call_timeout |
    {call_encode_failed, distribute@codec:encode_error()} |
    {call_decode_failed, distribute@codec:decode_error()}.

-file("src/distribute/global.gleam", 44).
?DOC(" New subject owned by the current process, with a unique tag.\n").
-spec new(
    fun((FZU) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}),
    fun((bitstring()) -> {ok, FZU} | {error, distribute@codec:decode_error()})
) -> global_subject(FZU).
new(Encoder, Decoder) ->
    Subject = distribute_ffi_utils:create_subject(
        erlang:self(),
        erlang:make_ref()
    ),
    {global_subject, Subject, Encoder, Decoder}.

-file("src/distribute/global.gleam", 53).
?DOC(" Send-only subject from a remote PID. Uses a nil tag.\n").
-spec from_pid(
    gleam@erlang@process:pid_(),
    fun((FZY) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}),
    fun((bitstring()) -> {ok, FZY} | {error, distribute@codec:decode_error()})
) -> global_subject(FZY).
from_pid(Pid, Encoder, Decoder) ->
    Subject = distribute_ffi_utils:create_subject(Pid, gleam@dynamic:nil()),
    {global_subject, Subject, Encoder, Decoder}.

-file("src/distribute/global.gleam", 63).
?DOC(" Wrap an existing `Subject(BitArray)`, keeping its tag.\n").
-spec from_subject(
    gleam@erlang@process:subject(bitstring()),
    fun((GAD) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}),
    fun((bitstring()) -> {ok, GAD} | {error, distribute@codec:decode_error()})
) -> global_subject(GAD).
from_subject(Subject, Encoder, Decoder) ->
    {global_subject, Subject, Encoder, Decoder}.

-file("src/distribute/global.gleam", 73).
?DOC(
    " Subject from a name and PID. The name becomes the tag,\n"
    " allowing any node with the same name to reconstruct it.\n"
).
-spec from_name(
    binary(),
    gleam@erlang@process:pid_(),
    fun((GAH) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}),
    fun((bitstring()) -> {ok, GAH} | {error, distribute@codec:decode_error()})
) -> global_subject(GAH).
from_name(Name, Pid, Encoder, Decoder) ->
    Subject = distribute_ffi_utils:create_subject(
        Pid,
        gleam_stdlib:identity(Name)
    ),
    {global_subject, Subject, Encoder, Decoder}.

-file("src/distribute/global.gleam", 87).
-spec subject(global_subject(any())) -> gleam@erlang@process:subject(bitstring()).
subject(Global) ->
    erlang:element(2, Global).

-file("src/distribute/global.gleam", 91).
-spec owner(global_subject(any())) -> {ok, gleam@erlang@process:pid_()} |
    {error, nil}.
owner(Global) ->
    gleam@erlang@process:subject_owner(erlang:element(2, Global)).

-file("src/distribute/global.gleam", 95).
-spec encoder(global_subject(GAS)) -> fun((GAS) -> {ok, bitstring()} |
    {error, distribute@codec:encode_error()}).
encoder(Global) ->
    erlang:element(3, Global).

-file("src/distribute/global.gleam", 99).
-spec decoder(global_subject(GAV)) -> fun((bitstring()) -> {ok, GAV} |
    {error, distribute@codec:decode_error()}).
decoder(Global) ->
    erlang:element(4, Global).

-file("src/distribute/global.gleam", 108).
?DOC(" Encode and send a message.\n").
-spec send(global_subject(GAY), GAY) -> {ok, nil} |
    {error, distribute@codec:encode_error()}.
send(Global, Message) ->
    Start = distribute_ffi_utils:system_time_ms(),
    case distribute@codec:encode(erlang:element(3, Global), Message) of
        {ok, Binary} ->
            Duration = distribute_ffi_utils:system_time_ms() - Start,
            Size = erlang:byte_size(Binary),
            distribute@internal@telemetry:emit_codec_encode_stop(
                Duration,
                Size,
                true
            ),
            Start_send = distribute_ffi_utils:system_time_ms(),
            gleam@erlang@process:send(erlang:element(2, Global), Binary),
            Duration_send = distribute_ffi_utils:system_time_ms() - Start_send,
            distribute@internal@telemetry:emit_message_send_stop(
                Duration_send,
                Size,
                <<"unknown"/utf8>>,
                true
            ),
            {ok, nil};

        {error, Err} ->
            Duration@1 = distribute_ffi_utils:system_time_ms() - Start,
            distribute@internal@telemetry:emit_codec_encode_stop(
                Duration@1,
                0,
                false
            ),
            {error, Err}
    end.

-file("src/distribute/global.gleam", 137).
?DOC(" Receive and decode a message. Only works on subjects you own.\n").
-spec 'receive'(global_subject(GBC), integer()) -> {ok, GBC} |
    {error, distribute@codec:decode_error()}.
'receive'(Global, Timeout_ms) ->
    case gleam@erlang@process:'receive'(erlang:element(2, Global), Timeout_ms) of
        {ok, Binary} ->
            Start_decode = distribute_ffi_utils:system_time_ms(),
            Res = distribute@codec:decode(erlang:element(4, Global), Binary),
            Duration_decode = distribute_ffi_utils:system_time_ms() - Start_decode,
            distribute@internal@telemetry:emit_codec_decode_stop(
                Duration_decode,
                case Res of
                    {ok, _} ->
                        true;

                    _ ->
                        false
                end
            ),
            Res;

        {error, nil} ->
            {error, decode_timeout}
    end.

-file("src/distribute/global.gleam", 168).
?DOC(
    " Synchronous request/response. Creates a temporary subject,\n"
    " sends the request, waits for the reply.\n"
).
-spec call(
    global_subject(GBG),
    fun((gleam@erlang@process:subject(bitstring())) -> GBG),
    fun((bitstring()) -> {ok, GBJ} | {error, distribute@codec:decode_error()}),
    integer()
) -> {ok, GBJ} | {error, call_error()}.
call(Target, Make_request, Response_decoder, Timeout_ms) ->
    Reply_subject = gleam@erlang@process:new_subject(),
    Request = Make_request(Reply_subject),
    case send(Target, Request) of
        {error, E} ->
            {error, {call_encode_failed, E}};

        {ok, nil} ->
            case gleam@erlang@process:'receive'(Reply_subject, Timeout_ms) of
                {error, nil} ->
                    {error, call_timeout};

                {ok, Bits} ->
                    case distribute@codec:decode(Response_decoder, Bits) of
                        {ok, Value} ->
                            {ok, Value};

                        {error, E@1} ->
                            {error, {call_decode_failed, E@1}}
                    end
            end
    end.

-file("src/distribute/global.gleam", 193).
?DOC(" Send a response through a reply subject.\n").
-spec reply(
    gleam@erlang@process:subject(bitstring()),
    GBO,
    fun((GBO) -> {ok, bitstring()} | {error, distribute@codec:encode_error()})
) -> {ok, nil} | {error, distribute@codec:encode_error()}.
reply(Reply_to, Response, Encoder) ->
    case distribute@codec:encode(Encoder, Response) of
        {ok, Bits} ->
            gleam@erlang@process:send(Reply_to, Bits),
            {ok, nil};

        {error, E} ->
            {error, E}
    end.

-file("src/distribute/global.gleam", 207).
-spec call_error_to_string(call_error()) -> binary().
call_error_to_string(Error) ->
    case Error of
        call_timeout ->
            <<"Call timed out"/utf8>>;

        {call_encode_failed, E} ->
            <<"Call encode failed: "/utf8,
                (distribute@codec:encode_error_to_string(E))/binary>>;

        {call_decode_failed, E@1} ->
            <<"Call decode failed: "/utf8,
                (distribute@codec:decode_error_to_string(E@1))/binary>>
    end.