import gleam/erlang/atom.{Atom} import gleam/dynamic.{DecodeError, Dynamic} import gleam/otp/port import gleam/function import gleam/option.{None, Option, Some} /// A Pid (or Process identifier) is a reference to an OTP process, which is a /// lightweight thread that communicates by sending and receiving messages. /// pub external type Pid /// A reference is a special value where each new one is unique. For more /// information references see the [Erlang documentation][1]. /// /// [1]: https://erlang.org/doc/efficiency_guide/advanced.html#unique_references /// pub external type Reference /// Create a new reference. The reference is unique among the currently /// connected nodes in the Erlang cluster. /// pub external fn new_reference() -> Reference = "erlang" "make_ref" /// Get the Pid of the process that calls the function. /// pub external fn self() -> Pid = "erlang" "self" /// A sender is one end of a channel, it allows one or more processes to send /// data to the process that owns the channel. /// /// See the `send` function for sending of values using a sender, and /// `new_channel` for creation of a sender. /// pub opaque type Sender(msg) { Sender(pid: Pid, prepare: Option(fn(msg) -> Dynamic)) } /// A receiver is one end of a channel, it allows the owning process to receive /// data sent over the channel via the corresponding sender. /// /// See the `receive` for receiving values, the `map_receiver` and /// `merge_receiver` functions for combining receivers, and `new_channel` for /// creation of a receiver. /// pub external type Receiver(msg) external fn new_receiver(reference) -> Receiver(msg) = "gleam_otp_external" "new_receiver" /// Create a receiver for any system messages sent to the current process. /// /// If you are using a higher level abstraction such as `gleam/actor` system /// messages will be handled automatically for you and this function should not /// be used. If you are using long lived processes without using a higher level /// abstraction you will need to handle system messages manually. /// pub external fn system_receiver() -> Receiver(SystemMessage) = "gleam_otp_external" "system_receiver" /// Create a new channel for processes to communicate over, returning a sender /// and a receiver. /// /// The Receiver is owned by the process that calls this function and must not /// be sent to another process. Any process that attempts to receive on a /// receiver that does not belong to them will crash. /// pub fn new_channel() -> #(Sender(msg), Receiver(msg)) { let reference = new_reference() let sender = self() |> new_bare_sender |> map_sender(fn(msg) { #(reference, msg) }) let receiver = new_receiver(reference) #(sender, receiver) } // TODO: test /// Create a new channel sender that sends bare messages direct to the given /// process. /// /// The messages sent using this sender are not received on any particular /// channel and as such are not type checked. Always favour using the /// `new_channel` function when possible as this will provide a safer and more /// convenient /// /// This function may be useful when working with processes written in other BEAM /// languages as they may not use Gleam's channels to receive messages. /// pub fn new_bare_sender(pid: Pid) -> Sender(msg) { let prepare = fn(msg) { dynamic.from(msg) } Sender(pid: pid, prepare: Some(prepare)) } // TODO: test /// Close a channel, causing any future messages sent on it to be discarded. /// /// If the sender is used to send a message after the channel is closed it is /// still delivered to the receiver process but it will be discarded next time /// the process runs any receiver. /// /// If the receiver is for a monitor the monitor is removed and any associated /// messages are flushed from the message inbox. /// pub external fn close_channels(Receiver(msg)) -> Nil = "gleam_otp_external" "close_channels" /// Get the pid of the receiver process for a sender. /// pub fn pid(sender: Sender(msg)) -> Pid { sender.pid } /// Attempt to parse a pid from some dynamic data. /// /// This may be useful if you receive a pid in a message from an Erlang process. pub external fn pid_from_dynamic(Dynamic) -> Result(Pid, List(DecodeError)) = "gleam_otp_external" "pid_from_dynamic" /// Send a message over a channel. /// /// This function always succeeds, even if the receiving process has shut down /// or has closed the channel. /// pub fn send(sender: Sender(msg), message: msg) -> Sender(msg) { case sender.prepare { Some(prepare) -> { message |> prepare |> untyped_send(sender.pid, _) sender } None -> sender } } /// Create a sender that immediately discards any messages sent on it. /// /// This may be useful for wrapping Erlang processes which do not use channels, /// or other situations in which you need to return a sender but do not have /// one available. /// pub fn null_sender(pid: Pid) -> Sender(msg) { Sender(pid: pid, prepare: None) } type ProcessMonitorFlag { Process } external fn erlang_monitor_process(ProcessMonitorFlag, Pid) -> Reference = "erlang" "monitor" // TODO: test // TODO: test closing /// Start monitoring a process. When the process exits a message is sent over /// the returned channel with information about the exit. /// /// The message is only sent once, when the target process exits. If the /// process was not alive when this function is called the message will never /// be received. /// /// Closing the channel with `close_channels` demonitors the process and /// flushes any monitor message for this channel from the message inbox. /// pub fn monitor_process(pid: Pid) -> Receiver(ProcessDown) { let reference = erlang_monitor_process(Process, pid) new_receiver(#(Process, reference)) } type PortMonitorFlag { Port } external fn erlang_monitor_port(PortMonitorFlag, port.Port) -> Reference = "erlang" "monitor" // TODO: test /// Start monitoring a port. When the port exits a message is sent over /// the returned channel with information about the exit. /// /// The message is only sent once, when the target port exits. If the port was /// not alive when this function is called the message will never be received. /// /// Closing the channel with `close_channels` demonitors the port and flushes /// any monitor message for this channel from the message inbox. /// pub fn monitor_port(port: port.Port) -> Receiver(PortDown) { let reference = erlang_monitor_port(Port, port) new_receiver(#(Port, reference)) } /// A message received when a monitored process exits. /// pub type ProcessDown { ProcessDown(pid: Pid, reason: Dynamic) } /// A message received when a monitored port exits. /// pub type PortDown { PortDown(port: port.Port, reason: Dynamic) } /// Receive a message from one of the channels in a receiver. /// /// Be careful! This function does not return until there is a message to /// receive. If no message is received then the process will be stuck waiting /// forever. /// pub external fn receive_forever(Receiver(msg)) -> msg = "gleam_otp_external" "run_receiver_forever" /// Discard all messages on the channels in a given receiver. /// /// This function must be used carefully, it may cause caller processes to /// crash when they do not receive a reply as their request has been dropped. /// pub external fn flush(Receiver(msg)) -> Int = "gleam_otp_external" "flush_receiver" // TODO: test /// Merge one receiver into another, producing a receiver that contains the /// channels of both. /// /// If a channel is found in both receivers any mapping function from the /// second receiver is used in the new receiver. /// pub external fn merge_receiver(Receiver(a), Receiver(a)) -> Receiver(a) = "gleam_otp_external" "merge_receiver" /// A message received when a linked process exits when the current process is /// trapping exits. /// pub type Exit { Exit(pid: Pid, reason: Dynamic) } // TODO: test /// Receive a message that does not belong to any particular channel. /// /// This function is typically not very useful when working with Gleam but it /// useful when working with Erlang code that sends messages to your code. /// pub external fn bare_message_receiver() -> Receiver(Dynamic) = "gleam_otp_external" "bare_message_receiver" pub type ExitReason { // The process is stopping due to normal and expected reasons. This is not // considered an error. Normal // The process is stopping as the supervision tree the process belongs to is // shutting down. This is not considered an error. Shutdown // The process is stopping due to an unexpected problem. This is considered // and error and should be reported and logged appropriately. Abnormal(Dynamic) } pub type Mode { Running Suspended } pub type DebugOption { NoDebug } pub external type DebugState pub external fn debug_state(List(DebugOption)) -> DebugState = "sys" "debug_options" pub type StatusInfo { StatusInfo( mod: Atom, parent: Pid, mode: Mode, debug_state: DebugState, state: Dynamic, ) } // TODO: document // TODO: implement remaining messages pub type SystemMessage { // {replace_state, StateFn} // {change_code, Mod, Vsn, Extra} // {terminate, Reason} // {debug, {log, Flag}} // {debug, {trace, Flag}} // {debug, {log_to_file, FileName}} // {debug, {statistics, Flag}} // {debug, no_debug} // {debug, {install, {Func, FuncState}}} // {debug, {install, {FuncId, Func, FuncState}}} // {debug, {remove, FuncOrId}} GetStatus(Sender(StatusInfo)) Suspend(Sender(Nil)) Resume(Sender(Nil)) GetState(Sender(Dynamic)) } /// Check to see whether the process for a given Pid is alive. /// /// See the [Erlang documentation][erl] for more information. /// [erl]: http://erlang.org/doc/man/erlang.html#is_process_alive-1 /// pub external fn is_alive(Pid) -> Bool = "erlang" "is_process_alive" type KillFlag { Kill } external fn erlang_kill(to: Pid, because: KillFlag) -> Bool = "erlang" "exit" // TODO: test /// Send an untrappable `kill` exit signal to the target process. /// /// See the documentation for the Erlang [`erlang:exit`][1] function for more /// information. /// /// [1]: https://erlang.org/doc/man/erlang.html#exit-1 /// pub fn kill(pid: Pid) -> Nil { erlang_kill(pid, Kill) Nil } external fn erlang_send_exit(to: Pid, because: whatever) -> Bool = "erlang" "exit" // TODO: test // TODO: refine exit reason. Maybe make dedicated functions for each type to // match kill /// Sends an exit signal to a process, indicating that that process is to shut /// down. /// /// See the [Erlang documentation][erl] for more information. /// [erl]: http://erlang.org/doc/man/erlang.html#exit-2 /// pub fn send_exit(to pid: Pid, because reason: whatever) -> Nil { erlang_send_exit(pid, reason) Nil } /// Start a new process from a given function. /// /// The new process is linked to the parent process so if it crashes the parent /// will also crash. If you wish your program to tolerate crashes see the /// supervisor module. /// pub external fn start(fn() -> anything) -> Pid = "erlang" "spawn_link" /// Start a new process from a given function. /// /// The new process is not linked to the parent process so if it crashes the /// issue may be silently ignored, leaving your program in an invalid state. /// pub external fn start_unlinked(fn() -> anything) -> Pid = "erlang" "spawn" /// Receive a message from one of the channels in a given receiver, removing it /// from the process inbox and returning it. /// /// If there are no messages for this receiver in the inbox and one is not /// received within the timeout then an error is returned. /// pub external fn receive( receiver: Receiver(msg), timeout: Int, ) -> Result(msg, Nil) = "gleam_otp_external" "run_receiver" /// An error returned when making a call to a process. /// pub type CallError(msg) { /// The process being called exited before it sent a response. /// CalleeDown(reason: Dynamic) /// The process being called did not response within the permitted amount of /// time. /// CallTimeout } /// Add a transformation function to a receiver. When a message is received /// using this receiver the tranformation function is applied to the message. /// /// This function can be used to change the type of messages received and may /// be useful when combined with the `merge_receiver` function. /// pub external fn map_receiver(Receiver(a), with: fn(a) -> b) -> Receiver(b) = "gleam_otp_external" "map_receiver" /// Add a transformation function to a sender. When a message is sent using this /// sender the tranformation function is applied to the message before it is /// sent. /// /// This function can be used to change the type of messages sent and may /// be useful to change the type of a sender before giving it to another process. /// /// You may notice that this function takes a mapper function from `b` to `a` /// rather than from `a` to `b` as you would find in functions like `list.map` /// and `receiver.map`. This style of a map function may be called a /// _"contravarient"_ map. /// pub fn map_sender(sender: Sender(a), with mapper: fn(b) -> a) -> Sender(b) { let wrap = function.compose(mapper, _) let prepare = option.map(sender.prepare, wrap) Sender(prepare: prepare, pid: sender.pid) } // TODO: test error paths // This function is based off of Erlang's gen:do_call/4. /// Send a message over a channel and wait for a reply. /// /// If the receiving process exits or does not reply within the allowed amount /// of time then an error is returned. /// pub fn try_call( sender: Sender(request), make_request: fn(Sender(response)) -> request, timeout: Int, ) -> Result(response, CallError(response)) { let #(reply_sender, reply_receiver) = new_channel() // Monitor the callee process so we can tell if it goes down (meaning we // won't get a reply) let monitor = monitor_process(pid(sender)) // Send the request to the process over the channel send(sender, make_request(reply_sender)) let receiver = reply_receiver |> map_receiver(Ok) |> merge_receiver(map_receiver( monitor, fn(down: ProcessDown) { Error(CalleeDown(reason: down.reason)) }, )) // Await a reply or handle failure modes (timeout, process down, etc) let res = receive(receiver, timeout) // Demonitor the process and close the channels as we're done close_channels(receiver) // Prepare an appropriate error (if present) for the caller case res { Error(Nil) -> Error(CallTimeout) Ok(res) -> res } } // TODO: test error paths /// Send a message over a channel and wait for a reply. /// /// If the receiving process exits or does not reply within the allowed amount /// of time the calling process crashes. If you wish an error to be returned /// instead see the `try_call` function. /// pub fn call( sender: Sender(request), make_request: fn(Sender(response)) -> request, timeout: Int, ) -> response { assert Ok(resp) = try_call(sender, make_request, timeout) resp } type MessageQueueLenFlag { MessageQueueLen } external fn process_info_message_queue_length( Pid, MessageQueueLenFlag, ) -> #(Atom, Int) = "erlang" "process_info" /// Get the number of messages in the calling processes' inbox message queue. /// pub fn message_queue_size(pid: Pid) -> Int { process_info_message_queue_length(pid, MessageQueueLen).1 } /// Start trapping exits within the current process and return a receiver. /// /// When not trapping exits if a linked process crashes an `Exit` message is /// sent over the channel. This is the normal behaviour before this function is /// called. /// /// When trapping exits (after this function is called) if a linked process /// crashes an `Exit` message is sent over the channel. /// pub external fn trap_exits() -> Receiver(Exit) = "gleam_otp_external" "trap_exits" /// Stop trapping exits, causing any crashes in linked processes to also crash /// this process. /// /// See also the `trap_exits` function. /// pub external fn stop_trapping_exits() -> Nil = "gleam_otp_external" "stop_trapping_exits" pub external type Timer external fn erlang_send_after(Int, Pid, msg) -> Timer = "erlang" "send_after" external fn fake_timer() -> Timer = "erlang" "make_ref" /// Send a message over a channel after a specified timeout. /// pub fn send_after(sender: Sender(msg), delay: Int, message: msg) -> Timer { case sender.prepare { Some(prepare) -> erlang_send_after(delay, sender.pid, prepare(message)) None -> fake_timer() } } external fn erlang_cancel_timer(Timer) -> Dynamic = "erlang" "cancel_timer" /// Values returned when a timer is cancelled. /// pub type Cancelled { /// The timer could not be found. It probably has already triggered. /// TimerNotFound /// The timer was found and cancelled before it triggered. /// Cancelled(time_remaining: Int) } /// Cancel a given timer, causing it not to trigger if it has not done already. /// pub fn cancel_timer(timer: Timer) -> Cancelled { case dynamic.int(erlang_cancel_timer(timer)) { Ok(i) -> Cancelled(i) Error(_) -> TimerNotFound } } external fn erlang_link(Pid) -> Bool = "erlang" "link" // TODO: test /// Creates a link between the calling process and another process. /// /// When a process crashes any linked processes will also crash. This is useful /// to ensure that groups of processes that depend on each other all either /// succeed or fail together. /// /// See the `gleam/otp/supervisor` module and the `trap_exits` function for /// mechanisms for handling process crashes. /// pub fn link(pid: Pid) -> Nil { erlang_link(pid) Nil } external fn erlang_unlink(pid: Pid) -> Bool = "erlang" "link" // TODO: test /// Removes any existing link between the caller process and the target process. /// pub fn unlink(pid: Pid) -> Nil { erlang_unlink(pid) Nil } /// Send an untyped bare message to a process. /// /// You probably don't want to use this function! /// Bare messages are not not commonly used in Gleam as they are not typed /// checked, see the Sender and Receiver channel types for a type safe and more /// ergonomic way of sending messages to processes. This function may still be /// useful for sending messages to processes implemented in Erlang or other BEAM /// languages. /// /// Message handling is asynchronous and this function will likely return before /// the message is handled by the receiving processes. /// /// See the [Erlang documentation][erl] for more information. /// [erl]: http://erlang.org/doc/man/erlang.html#send-2 /// pub external fn untyped_send(to: Pid, msg: msg) -> msg = "erlang" "send"