-module(distribute@actor). -compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]). -define(FILEPATH, "src/distribute/actor.gleam"). -export([start_typed_actor/4, start_server/4, start/3, start_global/3, child_spec_typed_actor/3, child_spec_server/3, start_typed_actor_registered/5, start_typed_actor_started/4, child_spec_typed_actor_typed/4, start_typed_actor_supervised/4, pool_supervisor/5]). -export_type([actor_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. -type actor_error() :: {start_failed, gleam@otp@actor:start_error()} | {invalid_configuration, binary()}. -file("src/distribute/actor.gleam", 176). ?DOC( " Start a type-safe actor with global subject (RECOMMENDED).\n" "\n" " This is the recommended way to create distributed actors. It wraps\n" " `receiver.start_typed_actor` for convenience and compatibility.\n" "\n" " The returned `GlobalSubject(msg)` can be:\n" " - Registered globally with `registry.register_typed`\n" " - Used for cross-node messaging with `messaging.send_typed`\n" " - Passed to other processes safely\n" "\n" " Malformed messages are silently ignored to prevent crashes.\n" "\n" " ## Parameters\n" "\n" " - `initial_state`: The initial state value\n" " - `encoder`: Encoder for outgoing messages (used by clients)\n" " - `decoder`: Decoder for incoming messages\n" " - `handler`: Message handler returning `Next(state)`\n" "\n" " ## Returns\n" "\n" " A `GlobalSubject(msg)` that enforces type-safe messaging.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " pub type Counter {\n" " Inc\n" " Dec\n" " GetValue(Subject(Int))\n" " }\n" "\n" " let actor = actor.start_typed_actor(\n" " 0,\n" " counter_encoder(),\n" " counter_decoder(),\n" " fn(msg, count) {\n" " case msg {\n" " Inc -> receiver.Continue(count + 1)\n" " Dec -> receiver.Continue(count - 1)\n" " GetValue(reply) -> {\n" " process.send(reply, count)\n" " receiver.Continue(count)\n" " }\n" " }\n" " },\n" " )\n" " ```\n" ). -spec start_typed_actor( HQS, fun((HQT) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}), fun((bitstring()) -> {ok, HQT} | {error, distribute@codec:decode_error()}), fun((HQT, HQS) -> distribute@receiver:next(HQS)) ) -> distribute@global:global_subject(HQT). start_typed_actor(Initial_state, Encoder, Decoder, Handler) -> distribute@receiver:start_typed_actor( Initial_state, Encoder, Decoder, Handler ). -file("src/distribute/actor.gleam", 231). ?DOC( " Start a request-response server actor.\n" "\n" " This is a specialized version of `start_typed_actor` optimized for\n" " request-response patterns. It's semantically identical but provides\n" " clearer intent in the API.\n" "\n" " Use this when your actor primarily processes requests and sends responses\n" " back to callers via reply subjects embedded in the request messages.\n" "\n" " ## Parameters\n" "\n" " - `initial_state`: The initial server state\n" " - `encoder`: Encoder for request messages\n" " - `decoder`: Decoder for request messages\n" " - `handler`: Request handler that processes requests and sends replies\n" "\n" " ## Returns\n" "\n" " A `GlobalSubject(request)` for sending requests to the server.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " pub type Request {\n" " Add(Int, Int, Subject(Int))\n" " Multiply(Int, Int, Subject(Int))\n" " }\n" "\n" " let server = actor.start_server(\n" " Nil,\n" " request_encoder(),\n" " request_decoder(),\n" " fn(req, _state) {\n" " case req {\n" " Add(a, b, reply) -> {\n" " process.send(reply, a + b)\n" " receiver.Continue(Nil)\n" " }\n" " Multiply(a, b, reply) -> {\n" " process.send(reply, a * b)\n" " receiver.Continue(Nil)\n" " }\n" " }\n" " },\n" " )\n" " ```\n" ). -spec start_server( HQY, fun((HQZ) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}), fun((bitstring()) -> {ok, HQZ} | {error, distribute@codec:decode_error()}), fun((HQZ, HQY) -> distribute@receiver:next(HQY)) ) -> distribute@global:global_subject(HQZ). start_server(Initial_state, Encoder, Decoder, Handler) -> start_typed_actor(Initial_state, Encoder, Decoder, Handler). -file("src/distribute/actor.gleam", 259). ?DOC( " Start an actor that returns a raw `Subject(BitArray)`.\n" "\n" " This is a lower-level helper that wraps `receiver.start_typed_receiver`.\n" " Prefer `start_typed_actor` which returns a type-safe `GlobalSubject(msg)`.\n" "\n" " Use this only when you need direct access to the underlying `Subject(BitArray)`\n" " for manual message encoding or integration with legacy code.\n" "\n" " ## Parameters\n" "\n" " - `initial_state`: The initial state\n" " - `decoder`: Decoder for incoming BitArray messages\n" " - `handler`: Message handler\n" "\n" " ## Returns\n" "\n" " `Ok(Subject(BitArray))` on success, `Error(StartFailed(err))` on failure.\n" ). -spec start( HRE, fun((bitstring()) -> {ok, HRF} | {error, distribute@codec:decode_error()}), fun((HRF, HRE) -> distribute@receiver:next(HRE)) ) -> {ok, gleam@erlang@process:subject(bitstring())} | {error, gleam@otp@actor:start_error()}. start(Initial_state, Decoder, Handler) -> distribute@log:warn( <<"actor.start is deprecated and will be removed in v3.0.0; use actor.start_typed_actor(...) instead"/utf8>>, [{<<"replacement"/utf8>>, <<"actor.start_typed_actor"/utf8>>}] ), distribute@receiver:start_typed_receiver(Initial_state, Decoder, Handler). -file("src/distribute/actor.gleam", 289). ?DOC( " Start a global actor returning a raw `Subject(BitArray)`.\n" "\n" " Similar to `start` but uses `receiver.start_global_receiver` which\n" " creates a subject compatible with global registry operations.\n" "\n" " Prefer `start_typed_actor` which wraps this and returns a type-safe\n" " `GlobalSubject(msg)`.\n" "\n" " ## Parameters\n" "\n" " - `initial_state`: The initial state\n" " - `decoder`: Decoder for incoming messages\n" " - `handler`: Message handler\n" "\n" " ## Returns\n" "\n" " A `Subject(BitArray)` with a `Nil` tag for global use.\n" ). -spec start_global( HRL, fun((bitstring()) -> {ok, HRM} | {error, distribute@codec:decode_error()}), fun((HRM, HRL) -> distribute@receiver:next(HRL)) ) -> gleam@erlang@process:subject(bitstring()). start_global(Initial_state, Decoder, Handler) -> distribute@log:warn( <<"actor.start_global is deprecated and will be removed in v3.0.0; use actor.start_typed_actor(...) and registry.register_typed(...) instead"/utf8>>, [{<<"replacement"/utf8>>, <<"actor.start_typed_actor"/utf8>>}] ), distribute@receiver:start_global_receiver(Initial_state, Decoder, Handler). -file("src/distribute/actor.gleam", 337). ?DOC( " Create a supervision child spec for a typed actor.\n" "\n" " This function returns a `ChildSpecification` that can be added to a\n" " `gleam/otp/supervision` supervisor tree. The supervisor will automatically\n" " restart the actor if it crashes.\n" "\n" " Note: Returns `Subject(BitArray)` rather than `GlobalSubject(msg)` for\n" " OTP compatibility. Wrap with `global.from_subject` after supervision start\n" " if you need the typed wrapper.\n" "\n" " ## Parameters\n" "\n" " - `initial_state`: The initial state for the actor\n" " - `decoder`: Decoder for messages\n" " - `handler`: Message handler function\n" "\n" " ## Returns\n" "\n" " A `ChildSpecification(Subject(BitArray))` for use with supervisors.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import gleam/otp/supervision.{supervisor}\n" "\n" " pub fn start_supervised() {\n" " supervisor.start_link(fn(children) {\n" " children\n" " |> supervision.add(actor.child_spec_typed_actor(\n" " 0,\n" " my_decoder(),\n" " my_handler,\n" " ))\n" " })\n" " }\n" " ```\n" ). -spec child_spec_typed_actor( HRQ, fun((bitstring()) -> {ok, HRR} | {error, distribute@codec:decode_error()}), fun((HRR, HRQ) -> distribute@receiver:next(HRQ)) ) -> gleam@otp@supervision:child_specification(gleam@erlang@process:subject(bitstring())). child_spec_typed_actor(Initial_state, Decoder, Handler) -> gleam@otp@supervision:worker( fun() -> _pipe = gleam@otp@actor:new(Initial_state), _pipe@1 = gleam@otp@actor:on_message( _pipe, fun(State, Binary) -> case Decoder(Binary) of {ok, Message} -> case Handler(Message, State) of {continue, New_state} -> gleam@otp@actor:continue(New_state); stop -> gleam@otp@actor:stop(); {stop_abnormal, Reason} -> gleam@otp@actor:stop_abnormal(Reason) end; {error, _} -> gleam@otp@actor:continue(State) end end ), gleam@otp@actor:start(_pipe@1) end ). -file("src/distribute/actor.gleam", 377). ?DOC( " Create a supervision child spec for a request-response server.\n" "\n" " Identical to `child_spec_typed_actor` but with clearer semantic intent\n" " for server patterns.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " supervisor.start_link(fn(children) {\n" " children\n" " |> supervision.add(actor.child_spec_server(\n" " Nil,\n" " request_decoder(),\n" " handle_request,\n" " ))\n" " })\n" " ```\n" ). -spec child_spec_server( HRW, fun((bitstring()) -> {ok, HRX} | {error, distribute@codec:decode_error()}), fun((HRX, HRW) -> distribute@receiver:next(HRW)) ) -> gleam@otp@supervision:child_specification(gleam@erlang@process:subject(bitstring())). child_spec_server(Initial_state, Decoder, Handler) -> child_spec_typed_actor(Initial_state, Decoder, Handler). -file("src/distribute/actor.gleam", 393). ?DOC( " Start and register a typed actor under a global name.\n" " \n" " This helper starts a typed actor and registers its subject with\n" " `registry.register_typed(name, subject)`. Returns `Ok(GlobalSubject(msg))`\n" " on success or `Error(RegisterError)` if registration fails.\n" "\n" " Note: If registration fails, the started actor process will continue running.\n" " The caller is responsible for cleanup in error cases if needed.\n" ). -spec start_typed_actor_registered( binary(), HSC, fun((HSD) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}), fun((bitstring()) -> {ok, HSD} | {error, distribute@codec:decode_error()}), fun((HSD, HSC) -> distribute@receiver:next(HSC)) ) -> {ok, distribute@global:global_subject(HSD)} | {error, distribute@registry:register_error()}. start_typed_actor_registered(Name, Initial_state, Encoder, Decoder, Handler) -> Gs = start_typed_actor(Initial_state, Encoder, Decoder, Handler), case distribute@registry:register_typed(Name, distribute@global:subject(Gs)) of {ok, _} -> {ok, Gs}; {error, Err} -> {error, Err} end. -file("src/distribute/actor.gleam", 410). ?DOC( " Start a typed actor via the OTP actor API and return the full Started\n" " result with `GlobalSubject(msg)` as the returned data. This is useful\n" " for supervision integration where the caller needs access to the pid.\n" ). -spec start_typed_actor_started( HSK, fun((HSL) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}), fun((bitstring()) -> {ok, HSL} | {error, distribute@codec:decode_error()}), fun((HSL, HSK) -> distribute@receiver:next(HSK)) ) -> {ok, gleam@otp@actor:started(distribute@global:global_subject(HSL))} | {error, gleam@otp@actor:start_error()}. start_typed_actor_started(Initial_state, Encoder, Decoder, Handler) -> _pipe = gleam@otp@actor:new(Initial_state), _pipe@1 = gleam@otp@actor:on_message( _pipe, fun(State, Binary) -> case Decoder(Binary) of {ok, Message} -> case Handler(Message, State) of {continue, New_state} -> gleam@otp@actor:continue(New_state); stop -> gleam@otp@actor:stop(); {stop_abnormal, Reason} -> gleam@otp@actor:stop_abnormal(Reason) end; {error, _} -> gleam@otp@actor:continue(State) end end ), _pipe@2 = gleam@otp@actor:start(_pipe@1), gleam@result:map( _pipe@2, fun(Started) -> Subject = erlang:element(3, Started), Global_subject = distribute@global:from_subject( Subject, Encoder, Decoder ), {started, erlang:element(2, Started), Global_subject} end ). -file("src/distribute/actor.gleam", 440). ?DOC( " Create a typed child spec that returns `GlobalSubject(msg)` as the child data.\n" " This helper is supervision-friendly and returns a `ChildSpecification` with\n" " the typed data so supervisors can access the typed subject directly.\n" ). -spec child_spec_typed_actor_typed( HST, fun((HSU) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}), fun((bitstring()) -> {ok, HSU} | {error, distribute@codec:decode_error()}), fun((HSU, HST) -> distribute@receiver:next(HST)) ) -> gleam@otp@supervision:child_specification(distribute@global:global_subject(HSU)). child_spec_typed_actor_typed(Initial_state, Encoder, Decoder, Handler) -> gleam@otp@supervision:worker( fun() -> start_typed_actor_started(Initial_state, Encoder, Decoder, Handler) end ). -file("src/distribute/actor.gleam", 496). ?DOC( " Start a typed actor under a new supervisor in one call.\n" "\n" " This high-level helper creates a supervisor, adds the typed actor as a child,\n" " and starts the supervisor tree. Returns both the supervisor PID and the\n" " typed actor's GlobalSubject for convenient access.\n" "\n" " ## Parameters\n" "\n" " - `initial_state`: The initial state for the actor\n" " - `encoder`: Encoder for messages\n" " - `decoder`: Decoder for messages\n" " - `handler`: Message handler function\n" "\n" " ## Returns\n" "\n" " `Ok((Pid, GlobalSubject(msg)))` on success with both handles,\n" " or `Error(StartError)` if supervision setup fails.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " let result = actor.start_typed_actor_supervised(\n" " 0,\n" " counter_encoder(),\n" " counter_decoder(),\n" " fn(msg, count) {\n" " case msg {\n" " Increment -> receiver.Continue(count + 1)\n" " GetCount(reply) -> {\n" " process.send(reply, count)\n" " receiver.Continue(count)\n" " }\n" " }\n" " },\n" " )\n" "\n" " case result {\n" " Ok(#(sup_pid, actor_subject)) -> {\n" " // Use actor_subject for messaging\n" " let _ = global.send(actor_subject, Increment)\n" " // Supervisor will restart actor on crashes\n" " }\n" " Error(err) -> // handle error\n" " }\n" " ```\n" ). -spec start_typed_actor_supervised( HTA, fun((HTB) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}), fun((bitstring()) -> {ok, HTB} | {error, distribute@codec:decode_error()}), fun((HTB, HTA) -> distribute@receiver:next(HTA)) ) -> {ok, {gleam@erlang@process:pid_(), distribute@global:global_subject(HTB)}} | {error, gleam@otp@actor:start_error()}. start_typed_actor_supervised(Initial_state, Encoder, Decoder, Handler) -> Child_spec = child_spec_typed_actor_typed( Initial_state, Encoder, Decoder, Handler ), Builder = gleam@otp@static_supervisor:new(one_for_one), Builder@1 = gleam@otp@static_supervisor:add(Builder, Child_spec), case gleam@otp@static_supervisor:start(Builder@1) of {ok, Sup} -> case start_typed_actor_started( Initial_state, Encoder, Decoder, Handler ) of {ok, Started} -> {ok, {erlang:element(2, Sup), erlang:element(3, Started)}}; {error, Err} -> {error, Err} end; {error, Err@1} -> {error, Err@1} end. -file("src/distribute/actor.gleam", 568). ?DOC( " Create a worker pool with N typed actors under a supervisor.\n" "\n" " This helper creates a pool of identical worker actors, all supervised\n" " under a single supervisor. The pool can be used for load balancing,\n" " parallel processing, or handling concurrent requests.\n" "\n" " ## Parameters\n" "\n" " - `pool_size`: Number of worker actors to create\n" " - `initial_state`: Initial state for each worker (all start with same state)\n" " - `encoder`: Encoder for worker messages\n" " - `decoder`: Decoder for worker messages \n" " - `handler`: Message handler for workers\n" "\n" " ## Returns\n" "\n" " `Ok((Pid, List(GlobalSubject(msg))))` with the supervisor PID and all worker subjects,\n" " or `Error(StartError)` if pool creation fails.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " // Create a pool of 5 worker actors\n" " let result = actor.pool_supervisor(\n" " 5,\n" " 0,\n" " request_encoder(),\n" " request_decoder(),\n" " handle_request,\n" " )\n" "\n" " case result {\n" " Ok(#(sup_pid, workers)) -> {\n" " // Round-robin distribution\n" " list.index_map(tasks, fn(task, idx) {\n" " let worker = list.at(workers, idx % list.length(workers))\n" " case worker {\n" " Ok(w) -> global.send(w, task)\n" " Error(_) -> Error(Nil)\n" " }\n" " })\n" " }\n" " Error(err) -> // handle error\n" " }\n" " ```\n" ). -spec pool_supervisor( integer(), HTI, fun((HTJ) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}), fun((bitstring()) -> {ok, HTJ} | {error, distribute@codec:decode_error()}), fun((HTJ, HTI) -> distribute@receiver:next(HTI)) ) -> {ok, {gleam@erlang@process:pid_(), list(distribute@global:global_subject(HTJ))}} | {error, gleam@otp@actor:start_error()}. pool_supervisor(Pool_size, Initial_state, Encoder, Decoder, Handler) -> Child_specs = begin _pipe = gleam@list:range(1, Pool_size), gleam@list:map( _pipe, fun(_) -> child_spec_typed_actor_typed( Initial_state, Encoder, Decoder, Handler ) end ) end, Builder = gleam@otp@static_supervisor:new(one_for_one), Builder_with_children = gleam@list:fold( Child_specs, Builder, fun(Acc, Spec) -> gleam@otp@static_supervisor:add(Acc, Spec) end ), case gleam@otp@static_supervisor:start(Builder_with_children) of {ok, Sup} -> Workers_result = begin _pipe@1 = gleam@list:range(1, Pool_size), _pipe@2 = gleam@list:map( _pipe@1, fun(_) -> start_typed_actor_started( Initial_state, Encoder, Decoder, Handler ) end ), gleam@result:all(_pipe@2) end, case Workers_result of {ok, Started_workers} -> Subjects = gleam@list:map( Started_workers, fun(Started) -> erlang:element(3, Started) end ), {ok, {erlang:element(2, Sup), Subjects}}; {error, Err} -> {error, Err} end; {error, Err@1} -> {error, Err@1} end.