Current section
Files
Jump to
Current section
Files
src/gleam@otp@actor.erl
-module(gleam@otp@actor).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/gleam/otp/actor.gleam").
-export([continue/1, stop/0, stop_abnormal/1, with_selector/2, initialised/1, selecting/2, returning/2, new/1, new_with_initialiser/2, on_message/2, named/2, start/1, send/2, call/3]).
-export_type([message/1, next/2, self/2, started/1, initialised/3, builder/3, start_error/0, start_init_message/1]).
-if(?OTP_RELEASE >= 27).
-define(MODULEDOC(Str), -moduledoc(Str)).
-define(DOC(Str), -doc(Str)).
-else.
-define(MODULEDOC(Str), -compile([])).
-define(DOC(Str), -compile([])).
-endif.
?MODULEDOC(
" This module provides the _Actor_ abstraction, one of the most common\n"
" building blocks of Gleam OTP programs.\n"
"\n"
" An Actor is a process like any other BEAM process and can be used to hold\n"
" state, execute code, and communicate with other processes by sending and\n"
" receiving messages. The advantage of using the actor abstraction over a bare\n"
" process is that it provides a single interface for commonly needed\n"
" functionality, including support for the [tracing and debugging\n"
" features in OTP][erlang-sys].\n"
"\n"
" Gleam's Actor is similar to Erlang's `gen_server` and Elixir's `GenServer`\n"
" but differs in that it offers a fully typed interface. This different API is\n"
" why Gleam uses the name \"Actor\" rather than some variation of\n"
" \"generic-server\".\n"
"\n"
" [erlang-sys]: https://www.erlang.org/doc/man/sys.html\n"
"\n"
" ## Example\n"
"\n"
" An Actor can be used to create a client-server interaction between an Actor\n"
" (the server) and other processes (the clients). In this example we have an\n"
" Actor that works as a stack, allowing clients to push and pop elements.\n"
"\n"
" ```gleam\n"
" pub fn main() {\n"
" // Start the actor with initial state of an empty list, and the\n"
" // `handle_message` callback function (defined below).\n"
" // We assert that it starts successfully.\n"
" // \n"
" // In real-world Gleam OTP programs we would likely write a wrapper functions\n"
" // called `start`, `push` `pop`, `shutdown` to start and interact with the\n"
" // Actor. We are not doing that here for the sake of showing how the Actor \n"
" // API works.\n"
" let assert Ok(actor) =\n"
" actor.new([]) |> actor.on_message(handle_message) |> actor.start\n"
" let subject = actor.data\n"
"\n"
" // We can send a message to the actor to push elements onto the stack.\n"
" process.send(subject, Push(\"Joe\"))\n"
" process.send(subject, Push(\"Mike\"))\n"
" process.send(subject, Push(\"Robert\"))\n"
"\n"
" // The `Push` message expects no response, these messages are sent purely for\n"
" // the side effect of mutating the state held by the actor.\n"
" //\n"
" // We can also send the `Pop` message to take a value off of the actor's\n"
" // stack. This message expects a response, so we use `process.call` to send a\n"
" // message and wait until a reply is received.\n"
" //\n"
" // In this instance we are giving the actor 10 milliseconds to reply, if the\n"
" // `call` function doesn't get a reply within this time it will panic and\n"
" // crash the client process.\n"
" let assert Ok(\"Robert\") = process.call(subject, 10, Pop)\n"
" let assert Ok(\"Mike\") = process.call(subject, 10, Pop)\n"
" let assert Ok(\"Joe\") = process.call(subject, 10, Pop)\n"
"\n"
" // The stack is now empty, so if we pop again the actor replies with an error.\n"
" let assert Error(Nil) = process.call(subject, 10, Pop)\n"
"\n"
" // Lastly, we can send a message to the actor asking it to shut down.\n"
" process.send(subject, Shutdown)\n"
" }\n"
" ```\n"
"\n"
" Here is the code that is used to implement this actor:\n"
"\n"
" ```gleam\n"
" // First step of implementing the stack Actor is to define the message type that\n"
" // it can receive.\n"
" //\n"
" // The type of the elements in the stack is not fixed so a type parameter\n"
" // is used for it instead of a concrete type such as `String` or `Int`.\n"
" pub type Message(element) {\n"
" // The `Shutdown` message is used to tell the actor to stop.\n"
" // It is the simplest message type, it contains no data.\n"
" //\n"
" // Most the time we don't define an API to shut down an actor, but in this\n"
" // example we do to show how it can be done.\n"
" Shutdown\n"
" \n"
" // The `Push` message is used to add a new element to the stack.\n"
" // It contains the item to add, the type of which is the `element`\n"
" // parameterised type.\n"
" Push(push: element)\n"
" \n"
" // The `Pop` message is used to remove an element from the stack.\n"
" // It contains a `Subject`, which is used to send the response back to the\n"
" // message sender. In this case the reply is of type `Result(element, Nil)`.\n"
" Pop(reply_with: Subject(Result(element, Nil)))\n"
" }\n"
" \n"
" // The last part is to implement the `handle_message` callback function.\n"
" //\n"
" // This function is called by the Actor for each message it receives.\n"
" // Actors are single threaded only doing one thing at a time, so they handle\n"
" // messages sequentially one at a time, in the order they are received.\n"
" //\n"
" // The function takes the current state and a message, and returns a data\n"
" // structure that indicates what to do next, along with the new state.\n"
" fn handle_message(\n"
" stack: List(e),\n"
" message: Message(e),\n"
" ) -> actor.Next(List(e), Message(e)) {\n"
" case message {\n"
" // For the `Shutdown` message we return the `actor.stop` value, which causes\n"
" // the actor to discard any remaining messages and stop.\n"
" // We may chose to do some clean-up work here, but this actor doesn't need\n"
" // to do this.\n"
" Shutdown -> actor.stop()\n"
" \n"
" // For the `Push` message we add the new element to the stack and return\n"
" // `actor.continue` with this new stack, causing the actor to process any\n"
" // queued messages or wait for more.\n"
" Push(value) -> {\n"
" let new_state = [value, ..stack]\n"
" actor.continue(new_state)\n"
" }\n"
" \n"
" // For the `Pop` message we attempt to remove an element from the stack,\n"
" // sending it or an error back to the caller, before continuing.\n"
" Pop(client) -> {\n"
" case stack {\n"
" [] -> {\n"
" // When the stack is empty we can't pop an element, so we send an\n"
" // error back.\n"
" process.send(client, Error(Nil))\n"
" actor.continue([])\n"
" }\n"
" \n"
" [first, ..rest] -> {\n"
" // Otherwise we send the first element back and use the remaining\n"
" // elements as the new state.\n"
" process.send(client, Ok(first))\n"
" actor.continue(rest)\n"
" }\n"
" }\n"
" }\n"
" }\n"
" }\n"
" ```\n"
).
-type message(EFI) :: {message, EFI} |
{system, gleam@otp@system:system_message()} |
{unexpected, gleam@dynamic:dynamic_()}.
-opaque next(EFJ, EFK) :: {continue,
EFJ,
gleam@option:option(gleam@erlang@process:selector(EFK))} |
{stop, gleam@erlang@process:exit_reason()}.
-type self(EFL, EFM) :: {self,
gleam@otp@system:mode(),
gleam@erlang@process:pid_(),
EFL,
gleam@erlang@process:selector(message(EFM)),
gleam@otp@system:debug_state(),
fun((EFL, EFM) -> next(EFL, EFM))}.
-type started(EFN) :: {started, gleam@erlang@process:pid_(), EFN}.
-opaque initialised(EFO, EFP, EFQ) :: {initialised,
EFO,
gleam@option:option(gleam@erlang@process:selector(EFP)),
EFQ}.
-opaque builder(EFR, EFS, EFT) :: {builder,
fun((gleam@erlang@process:subject(EFS)) -> {ok,
initialised(EFR, EFS, EFT)} |
{error, binary()}),
integer(),
fun((EFR, EFS) -> next(EFR, EFS)),
gleam@option:option(gleam@erlang@process:name(EFS))}.
-type start_error() :: init_timeout |
{init_failed, binary()} |
{init_exited, gleam@erlang@process:exit_reason()}.
-type start_init_message(EFU) :: {ack, {ok, EFU} | {error, binary()}} |
{mon, gleam@erlang@process:down()}.
-file("src/gleam/otp/actor.gleam", 185).
?DOC(" Indicate the actor should continue, processing any waiting or future messages.\n").
-spec continue(EFZ) -> next(EFZ, any()).
continue(State) ->
{continue, State, none}.
-file("src/gleam/otp/actor.gleam", 193).
?DOC(
" Indicate the actor should stop and shut-down, handling no futher messages.\n"
"\n"
" The reason for exiting is `Normal`.\n"
).
-spec stop() -> next(any(), any()).
stop() ->
{stop, normal}.
-file("src/gleam/otp/actor.gleam", 202).
?DOC(
" Indicate the actor is in a bad state and should shut down. It will not\n"
" handle any new messages, and any linked processes will also exit abnormally.\n"
"\n"
" The provided reason will be given and propagated.\n"
).
-spec stop_abnormal(binary()) -> next(any(), any()).
stop_abnormal(Reason) ->
{stop, {abnormal, gleam_stdlib:identity(Reason)}}.
-file("src/gleam/otp/actor.gleam", 210).
?DOC(
" Provide a selector to change the messages that the actor is handling\n"
" going forward. This replaces any selector that was previously given\n"
" in the actor's `init` callback, or in any previous `Next` value.\n"
).
-spec with_selector(next(EGL, EGM), gleam@erlang@process:selector(EGM)) -> next(EGL, EGM).
with_selector(Value, Selector) ->
case Value of
{continue, State, _} ->
{continue, State, {some, Selector}};
{stop, _} ->
Value
end.
-file("src/gleam/otp/actor.gleam", 268).
?DOC(
" Takes the post-initialisation state of the actor. This state will be passed\n"
" to the `on_message` callback each time a message is received.\n"
).
-spec initialised(EGS) -> initialised(EGS, any(), nil).
initialised(State) ->
{initialised, State, none, nil}.
-file("src/gleam/otp/actor.gleam", 277).
?DOC(
" Add a selector for the actor to receive messages with.\n"
"\n"
" If a message is received by the actor but not selected for with the\n"
" selector then the actor will discard it and log a warning.\n"
).
-spec selecting(
initialised(EGX, any(), EGZ),
gleam@erlang@process:selector(EHD)
) -> initialised(EGX, EHD, EGZ).
selecting(Initialised, Selector) ->
{initialised,
erlang:element(2, Initialised),
{some, Selector},
erlang:element(4, Initialised)}.
-file("src/gleam/otp/actor.gleam", 287).
?DOC(
" Add the data to return to the parent process. This might be a subject that\n"
" the actor will receive messages over.\n"
).
-spec returning(initialised(EHI, EHJ, any()), EHO) -> initialised(EHI, EHJ, EHO).
returning(Initialised, Return) ->
{initialised,
erlang:element(2, Initialised),
erlang:element(3, Initialised),
Return}.
-file("src/gleam/otp/actor.gleam", 329).
?DOC(
" Create a builder for an actor without a custom initialiser. The actor\n"
" returns a subject to the parent that can be used to send messages to the\n"
" actor.\n"
"\n"
" If the actor has been given a name with the `named` function then the\n"
" subject is a named subject.\n"
"\n"
" If you wish to create an actor with some other initialisation logic that\n"
" runs before it starts handling messages, see `new_with_initialiser`.\n"
).
-spec new(EHS) -> builder(EHS, EHT, gleam@erlang@process:subject(EHT)).
new(State) ->
Initialise = fun(Subject) -> _pipe = initialised(State),
_pipe@1 = returning(_pipe, Subject),
{ok, _pipe@1} end,
{builder, Initialise, 1000, fun(State@1, _) -> continue(State@1) end, none}.
-file("src/gleam/otp/actor.gleam", 358).
?DOC(
" Create a builder for an actor with a custom initialiser that runs before\n"
" the start function returns to the parent, and before the actor starts\n"
" handling messages.\n"
"\n"
" The first argument is a number of milliseconds that the initialiser\n"
" function is expected to return within. If it takes longer the initialiser\n"
" is considered to have failed and the actor will be killed, and an error\n"
" will be returned to the parent.\n"
"\n"
" The actor's default subject is passed to the initialiser function. You can\n"
" chose to return it to the parent with `returning`, use it in some other\n"
" way, or ignore it completely.\n"
"\n"
" If a custom selector is given using the `selecting` function then this\n"
" overwrites the default selector, which selects for the default subject, so\n"
" you will need to add the subject to the custom selector yourself.\n"
).
-spec new_with_initialiser(
integer(),
fun((gleam@erlang@process:subject(EHY)) -> {ok, initialised(EIA, EHY, EIB)} |
{error, binary()})
) -> builder(EIA, EHY, EIB).
new_with_initialiser(Timeout, Initialise) ->
{builder, Initialise, Timeout, fun(State, _) -> continue(State) end, none}.
-file("src/gleam/otp/actor.gleam", 376).
?DOC(
" Set the message handler for the actor. This callback function will be\n"
" called each time the actor receives a message.\n"
"\n"
" Actors handle messages sequentially, later messages being handled after the\n"
" previous one has been handled.\n"
).
-spec on_message(builder(EIK, EIL, EIM), fun((EIK, EIL) -> next(EIK, EIL))) -> builder(EIK, EIL, EIM).
on_message(Builder, Handler) ->
{builder,
erlang:element(2, Builder),
erlang:element(3, Builder),
Handler,
erlang:element(5, Builder)}.
-file("src/gleam/otp/actor.gleam", 395).
?DOC(
" Provide a name for the actor to be registered with when started, enabling\n"
" it to receive messages via a named subject. This is useful for making\n"
" processes that can take over from an older one that has exited due to a\n"
" failure, or to avoid passing subjects from receiver processes to sender\n"
" processes.\n"
"\n"
" If the name is already registered to another process then the actor will\n"
" fail to start.\n"
"\n"
" When this function is used the actor's default subject will be a named\n"
" subject using this name.\n"
).
-spec named(builder(EIV, EIW, EIX), gleam@erlang@process:name(EIW)) -> builder(EIV, EIW, EIX).
named(Builder, Name) ->
{builder,
erlang:element(2, Builder),
erlang:element(3, Builder),
erlang:element(4, Builder),
{some, Name}}.
-file("src/gleam/otp/actor.gleam", 402).
-spec exit_process(gleam@erlang@process:exit_reason()) -> gleam@erlang@process:exit_reason().
exit_process(Reason) ->
case Reason of
{abnormal, Reason@1} ->
gleam@erlang@process:send_abnormal_exit(erlang:self(), Reason@1);
killed ->
gleam@erlang@process:kill(erlang:self());
_ ->
nil
end,
Reason.
-file("src/gleam/otp/actor.gleam", 443).
-spec select_system_messages(gleam@erlang@process:selector(message(EJK))) -> gleam@erlang@process:selector(message(EJK)).
select_system_messages(Selector) ->
_pipe = Selector,
gleam@erlang@process:select_record(
_pipe,
erlang:binary_to_atom(<<"system"/utf8>>),
2,
fun gleam_otp_external:convert_system_message/1
).
-file("src/gleam/otp/actor.gleam", 412).
-spec receive_message(self(any(), EJG)) -> message(EJG).
receive_message(Self) ->
Selector = case erlang:element(2, Self) of
suspended ->
_pipe = gleam_erlang_ffi:new_selector(),
select_system_messages(_pipe);
running ->
_pipe@1 = gleam_erlang_ffi:new_selector(),
_pipe@2 = gleam@erlang@process:select_other(
_pipe@1,
fun(Field@0) -> {unexpected, Field@0} end
),
_pipe@3 = gleam_erlang_ffi:merge_selector(
_pipe@2,
erlang:element(5, Self)
),
select_system_messages(_pipe@3)
end,
gleam_erlang_ffi:select(Selector).
-file("src/gleam/otp/actor.gleam", 453).
-spec process_status_info(self(any(), any())) -> gleam@otp@system:status_info().
process_status_info(Self) ->
{status_info,
erlang:binary_to_atom(<<"gleam@otp@actor"/utf8>>),
erlang:element(3, Self),
erlang:element(2, Self),
erlang:element(6, Self),
gleam_otp_external:identity(erlang:element(4, Self))}.
-file("src/gleam/otp/actor.gleam", 463).
-spec loop(self(any(), any())) -> gleam@erlang@process:exit_reason().
loop(Self) ->
case receive_message(Self) of
{system, System} ->
case System of
{get_state, Callback} ->
Callback(
gleam_otp_external:identity(erlang:element(4, Self))
),
loop(Self);
{resume, Callback@1} ->
Callback@1(),
loop(
{self,
running,
erlang:element(3, Self),
erlang:element(4, Self),
erlang:element(5, Self),
erlang:element(6, Self),
erlang:element(7, Self)}
);
{suspend, Callback@2} ->
Callback@2(),
loop(
{self,
suspended,
erlang:element(3, Self),
erlang:element(4, Self),
erlang:element(5, Self),
erlang:element(6, Self),
erlang:element(7, Self)}
);
{get_status, Callback@3} ->
Callback@3(process_status_info(Self)),
loop(Self)
end;
{unexpected, Message} ->
logger:warning(
unicode:characters_to_list(
<<"Actor discarding unexpected message: ~s"/utf8>>
),
[unicode:characters_to_list(gleam@string:inspect(Message))]
),
loop(Self);
{message, Msg} ->
case (erlang:element(7, Self))(erlang:element(4, Self), Msg) of
{stop, Reason} ->
exit_process(Reason);
{continue, State, New_selector} ->
Selector = case New_selector of
none ->
erlang:element(5, Self);
{some, S} ->
gleam_erlang_ffi:map_selector(
S,
fun(Field@0) -> {message, Field@0} end
)
end,
loop(
{self,
erlang:element(2, Self),
erlang:element(3, Self),
State,
Selector,
erlang:element(6, Self),
erlang:element(7, Self)}
)
end
end.
-file("src/gleam/otp/actor.gleam", 573).
-spec try_register_self(gleam@erlang@process:name(any())) -> {ok, nil} |
{error, binary()}.
try_register_self(Name) ->
case gleam_erlang_ffi:register_process(erlang:self(), Name) of
{ok, nil} ->
{ok, nil};
{error, _} ->
{error, <<"name already registered"/utf8>>}
end.
-file("src/gleam/otp/actor.gleam", 522).
-spec initialise_actor(
builder(any(), any(), EKD),
gleam@erlang@process:pid_(),
gleam@erlang@process:subject({ok, EKD} | {error, binary()})
) -> gleam@erlang@process:exit_reason().
initialise_actor(Builder, Parent, Ack) ->
Result@1 = begin
gleam@result:'try'(case erlang:element(5, Builder) of
none ->
{ok, gleam@erlang@process:new_subject()};
{some, Name} ->
gleam@result:'try'(
try_register_self(Name),
fun(_) ->
{ok, gleam@erlang@process:named_subject(Name)}
end
)
end, fun(Subject) ->
gleam@result:'try'(
(erlang:element(2, Builder))(Subject),
fun(Result) -> {ok, {Subject, Result}} end
)
end)
end,
case Result@1 of
{ok, {Subject@1, {initialised, State, Selector, Return}}} ->
Selector@2 = case Selector of
{some, Selector@1} ->
Selector@1;
none ->
_pipe = gleam_erlang_ffi:new_selector(),
gleam@erlang@process:select(_pipe, Subject@1)
end,
Selector@3 = gleam_erlang_ffi:map_selector(
Selector@2,
fun(Field@0) -> {message, Field@0} end
),
gleam@erlang@process:send(Ack, {ok, Return}),
Self = {self,
running,
Parent,
State,
Selector@3,
sys:debug_options([]),
erlang:element(4, Builder)},
loop(Self);
{error, Reason} ->
gleam@erlang@process:send(Ack, {error, Reason}),
exit_process(normal)
end.
-file("src/gleam/otp/actor.gleam", 598).
?DOC(
" Start an actor from a given specification. If the actor's `init` function\n"
" returns an error or does not return within `init_timeout` then an error is\n"
" returned.\n"
"\n"
" If you do not need to specify the initialisation behaviour of your actor\n"
" consider using the `start` function.\n"
).
-spec start(builder(any(), any(), EKQ)) -> {ok, started(EKQ)} |
{error, start_error()}.
start(Builder) ->
Timeout = erlang:element(3, Builder),
Ack_subject = gleam@erlang@process:new_subject(),
Self = erlang:self(),
Child = proc_lib:spawn_link(
fun() -> initialise_actor(Builder, Self, Ack_subject) end
),
Monitor = gleam@erlang@process:monitor(Child),
Selector = begin
_pipe = gleam_erlang_ffi:new_selector(),
_pipe@1 = gleam@erlang@process:select_map(
_pipe,
Ack_subject,
fun(Field@0) -> {ack, Field@0} end
),
gleam@erlang@process:select_specific_monitor(
_pipe@1,
Monitor,
fun(Field@0) -> {mon, Field@0} end
)
end,
Result = case gleam_erlang_ffi:select(Selector, Timeout) of
{ok, {ack, {ok, Subject}}} ->
{ok, Subject};
{ok, {ack, {error, Reason}}} ->
{error, {init_failed, Reason}};
{ok, {mon, Down}} ->
{error, {init_exited, erlang:element(4, Down)}};
{error, nil} ->
gleam@erlang@process:unlink(Child),
gleam@erlang@process:kill(Child),
{error, init_timeout}
end,
gleam@erlang@process:demonitor_process(Monitor),
case Result of
{ok, Data} ->
{ok, {started, Child, Data}};
{error, Error} ->
{error, Error}
end.
-file("src/gleam/otp/actor.gleam", 648).
?DOC(
" Send a message over a given channel.\n"
"\n"
" This is a re-export of `process.send`, for the sake of convenience.\n"
).
-spec send(gleam@erlang@process:subject(EKX), EKX) -> nil.
send(Subject, Msg) ->
gleam@erlang@process:send(Subject, Msg).
-file("src/gleam/otp/actor.gleam", 660).
?DOC(
" Send a synchronous message and wait for a response from the receiving\n"
" process.\n"
"\n"
" If a reply is not received within the given timeout then the sender process\n"
" crashes rather than leaving the processes in an invalid state. \n"
"\n"
" This is a re-export of `process.call`, for the sake of convenience.\n"
).
-spec call(
gleam@erlang@process:subject(EKZ),
integer(),
fun((gleam@erlang@process:subject(ELB)) -> EKZ)
) -> ELB.
call(Subject, Timeout, Make_message) ->
gleam@erlang@process:call(Subject, Timeout, Make_message).