Trait GenServer 

pub trait GenServer:
    Sized
    + Send
    + 'static {
    type Message: Receivable;

    // Required method
    fn init(&mut self) -> impl Future<Output = Result<(), ExitReason>> + Send;

    // Provided methods
    fn start(
        self,
        options: GenServerOptions,
    ) -> impl Future<Output = Result<Pid, ExitReason>> + Send { ... }
    fn start_link(
        self,
        options: GenServerOptions,
    ) -> impl Future<Output = Result<Pid, ExitReason>> + Send { ... }
    fn stop<T: Into<Dest>>(
        server: T,
        reason: ExitReason,
        timeout: Option<Duration>,
    ) -> impl Future<Output = Result<(), ExitReason>> { ... }
    fn cast<T: Into<Dests>>(servers: T, message: Self::Message) { ... }
    fn cast_after<T: Into<Dests>>(
        servers: T,
        message: Self::Message,
        duration: Duration,
    ) -> Reference { ... }
    fn call<T: Into<Dest>>(
        server: T,
        message: Self::Message,
        timeout: Option<Duration>,
    ) -> impl Future<Output = Result<Self::Message, CallError>> + Send { ... }
    fn reply(from: From, message: Self::Message) { ... }
    fn terminate(
        &mut self,
        reason: ExitReason,
    ) -> impl Future<Output = ()> + Send { ... }
    fn handle_cast(
        &mut self,
        message: Self::Message,
    ) -> impl Future<Output = Result<(), ExitReason>> + Send { ... }
    fn handle_info(
        &mut self,
        info: Message<Self::Message>,
    ) -> impl Future<Output = Result<(), ExitReason>> + Send { ... }
    fn handle_call(
        &mut self,
        message: Self::Message,
        from: From,
    ) -> impl Future<Output = Result<Option<Self::Message>, ExitReason>> + Send { ... }
}

A trait for implementing the server of a client-server relation.

A GenServer is a process like any other hydra process and it can be used to keep state, execute code asynchronously and so on.

The advantage of using a generic server process (GenServer) implemented using this trait is that it will have a standard set of trait functions and include functionality for tracing and error reporting.

It will also fit into a supervision tree.

Example

Let’s start with a code example and then explore the available callbacks. Imagine we want to implement a service with a GenServer that works like a stack, allowing us to push and pop elements. We’ll customize a generic GenServer with our own module by implementing three callbacks.

#[derive(Debug, Serialize, Deserialize)]
enum StackMessage {
    Pop,
    PopResult(String),
    Push(String),
}

struct Stack {
    stack: Vec<String>,
}

impl Stack {
    pub fn with_entries(entries: Vec<&'static str>) -> Self {
        Self {
            stack: Vec::from_iter(entries.into_iter().map(Into::into)),
        }
    }
}

impl GenServer for Stack {
    type Message = StackMessage;

    async fn init(&mut self) -> Result<(), ExitReason> {
        Ok(())
    }

    async fn handle_call(&mut self, message: Self::Message, _from: From) -> Result<Option<Self::Message>, ExitReason> {
        match message {
            StackMessage::Pop => Ok(Some(StackMessage::PopResult(self.stack.remove(0)))),
            _ => unreachable!(),
        }
    }

    async fn handle_cast(&mut self, message: Self::Message) -> Result<(), ExitReason> {
        match message {
            StackMessage::Push(value) => self.stack.insert(0, value),
            _ => unreachable!(),
        }
        Ok(())
    }
}

We leave the process machinery of startup, message passing, and the message loop to the GenServer. We can now use the GenServer methods to interact with the service by creating a process and sending it messages:

// Start the server.
let pid = Stack::with_entries(vec![String::from("hello"), String::from("world")])
            .start_link(GenServerOptions::new())
            .await
            .expect("Failed to start stack!");

// This is the client.
Stack::call(pid, StackMessage::Pop, None)
        .await
        .expect("Stack call failed!");
// => StackMessage::PopResult("hello")

Stack::cast(pid, StackMessage::Push(String::from("rust")))

Stack::call(pid, StackMessage::Pop, None)
        .await
        .expect("Stack call failed!");
// => StackMessage::PopResult("rust")

Required Associated Types

type Message: Receivable

The message type that this server will use.

Required Methods

fn init(&mut self) -> impl Future<Output = Result<(), ExitReason>> + Send

Invoked when the server is started. start_link or start will block until it returns.

Provided Methods

fn start( self, options: GenServerOptions, ) -> impl Future<Output = Result<Pid, ExitReason>> + Send

Starts a GenServer process without links.

fn start_link( self, options: GenServerOptions, ) -> impl Future<Output = Result<Pid, ExitReason>> + Send

Starts a GenServer process linked to the current process.

Examples found in repository

examples/supervisor.rs (line 67)

    pub fn child_spec(self) -> ChildSpec {
        ChildSpec::new("MyServer")
            .start(move || MyServer::start_link(MyServer, GenServerOptions::new()))
    }

examples/application.rs (line 63)

    async fn start(&self) -> Result<Pid, ExitReason> {
        let pid = Stack::with_entries(vec!["hello", "world"])
            .start_link(GenServerOptions::new())
            .await
            .expect("Failed to start stack!");

        let result = Stack::call(pid, StackMessage::Pop, None)
            .await
            .expect("Stack call failed!");

        tracing::info!("{:?}", result);

        Stack::cast(pid, StackMessage::Push(String::from("rust")));

        let result = Stack::call(pid, StackMessage::Pop, None)
            .await
            .expect("Stack call failed!");

        tracing::info!("{:?}", result);

        // Otherwise, the application will run forever waiting for Stack to terminate.
        Stack::stop(pid, ExitReason::Normal, None).await?;

        Ok(pid)
    }

examples/registry.rs (line 67)

    async fn start(&self) -> Result<Pid, ExitReason> {
        // Spawn a registry that will take care of registering 'MySpace'.
        let children = [
            Registry::new("space-registry")
                .with_start(|key| {
                    let RegistryKey::String(id) = key else {
                        panic!()
                    };

                    MySpace::new(id).start_link(GenServerOptions::new())
                })
                .with_shutdown(Shutdown::Infinity)
                .child_spec(RegistryOptions::new())
                .id("space-registry"),
            ChildSpec::new("test-registry")
                .start(move || async { Ok(Process::spawn(test_registry())) }),
        ];

        // Restart only the terminated child.
        Supervisor::with_children(children)
            .strategy(SupervisionStrategy::OneForOne)
            .start_link(SupervisorOptions::new())
            .await
    }

fn stop<T: Into<Dest>>( server: T, reason: ExitReason, timeout: Option<Duration>, ) -> impl Future<Output = Result<(), ExitReason>>

Synchronously stops the server with the given reason.

The terminate callback of the given server will be invoked before exiting. This function returns an error if the process exits with a reason other than the given reason.

The default timeout is infinity.

Examples found in repository

examples/application.rs (line 82)

    async fn start(&self) -> Result<Pid, ExitReason> {
        let pid = Stack::with_entries(vec!["hello", "world"])
            .start_link(GenServerOptions::new())
            .await
            .expect("Failed to start stack!");

        let result = Stack::call(pid, StackMessage::Pop, None)
            .await
            .expect("Stack call failed!");

        tracing::info!("{:?}", result);

        Stack::cast(pid, StackMessage::Push(String::from("rust")));

        let result = Stack::call(pid, StackMessage::Pop, None)
            .await
            .expect("Stack call failed!");

        tracing::info!("{:?}", result);

        // Otherwise, the application will run forever waiting for Stack to terminate.
        Stack::stop(pid, ExitReason::Normal, None).await?;

        Ok(pid)
    }

fn cast<T: Into<Dests>>(servers: T, message: Self::Message)

Casts a request to the servers without waiting for a response.

It is unknown whether the destination server successfully handled the request.

See Process::send for performance trade-offs.

Examples found in repository

examples/supervisor.rs (line 89)

    async fn init(&mut self) -> Result<(), ExitReason> {
        let server = Process::current();

        Process::spawn(async move {
            // Ask for a formatted string.
            let hello_world = MyServer::hello(server, "hello")
                .await
                .expect("Failed to call server!");

            tracing::info!("Got: {:?}", hello_world);

            // Wait before crashing.
            Process::sleep(Duration::from_secs(1)).await;

            // Crash the process so the supervisor restarts it.
            MyServer::cast(server, MyMessage::Crash);
        });

        Ok(())
    }

examples/application.rs (line 73)

    async fn start(&self) -> Result<Pid, ExitReason> {
        let pid = Stack::with_entries(vec!["hello", "world"])
            .start_link(GenServerOptions::new())
            .await
            .expect("Failed to start stack!");

        let result = Stack::call(pid, StackMessage::Pop, None)
            .await
            .expect("Stack call failed!");

        tracing::info!("{:?}", result);

        Stack::cast(pid, StackMessage::Push(String::from("rust")));

        let result = Stack::call(pid, StackMessage::Pop, None)
            .await
            .expect("Stack call failed!");

        tracing::info!("{:?}", result);

        // Otherwise, the application will run forever waiting for Stack to terminate.
        Stack::stop(pid, ExitReason::Normal, None).await?;

        Ok(pid)
    }

fn cast_after<T: Into<Dests>>( servers: T, message: Self::Message, duration: Duration, ) -> Reference

Casts a request to the servers after the given duration without waiting for a response.

It is unknown whether the destination server successfully handled the request.

See Process::send for performance trade-offs.

fn call<T: Into<Dest>>( server: T, message: Self::Message, timeout: Option<Duration>, ) -> impl Future<Output = Result<Self::Message, CallError>> + Send

Makes a synchronous call to the server and waits for it’s reply.

The client sends the given message to the server and waits until a reply arrives or a timeout occurs. handle_call will be called on the server to handle the request.

The default timeout is 5000ms.

Examples found in repository

examples/supervisor.rs (line 58)

    pub async fn hello<T: Into<Dest>>(server: T, string: &str) -> Result<String, CallError> {
        use MyMessage::*;

        match MyServer::call(server, Hello(string.to_owned()), None).await? {
            HelloResponse(response) => Ok(response),
            _ => unreachable!(),
        }
    }

examples/application.rs (line 67)

    async fn start(&self) -> Result<Pid, ExitReason> {
        let pid = Stack::with_entries(vec!["hello", "world"])
            .start_link(GenServerOptions::new())
            .await
            .expect("Failed to start stack!");

        let result = Stack::call(pid, StackMessage::Pop, None)
            .await
            .expect("Stack call failed!");

        tracing::info!("{:?}", result);

        Stack::cast(pid, StackMessage::Push(String::from("rust")));

        let result = Stack::call(pid, StackMessage::Pop, None)
            .await
            .expect("Stack call failed!");

        tracing::info!("{:?}", result);

        // Otherwise, the application will run forever waiting for Stack to terminate.
        Stack::stop(pid, ExitReason::Normal, None).await?;

        Ok(pid)
    }

fn reply(from: From, message: Self::Message)

Replies to a client.

This function can be used to explicitly send a reply to a client that called call when the reply cannot be specified in the return value of handle_call.

client must be the from argument accepted by handle_call callbacks.

Note that reply can be called from any process, not just the GenServer that originally received the call (as long as the GenServer communicated the from argument somehow).

fn terminate(&mut self, reason: ExitReason) -> impl Future<Output = ()> + Send

Invoked when the server is about to exit. It should do any cleanup required.

terminate is useful for cleanup that requires access to the GenServer’s state. However, it is not guaranteed that terminate is called when a GenServer exits. Therefore, important cleanup should be done using process links and/or monitors. A monitoring process will receive the same reason that would be passed to terminate.

terminate is called if:

• The GenServer traps exits (using Process::flags) and the parent process sends an exit signal.
• A callback (except init) returns stop with a given reason.
• The stop method is called on a GenServer.

fn handle_cast( &mut self, message: Self::Message, ) -> impl Future<Output = Result<(), ExitReason>> + Send

Invoked to handle asynchronous cast messages.

fn handle_info( &mut self, info: Message<Self::Message>, ) -> impl Future<Output = Result<(), ExitReason>> + Send

Invoked to handle all other messages.

fn handle_call( &mut self, message: Self::Message, from: From, ) -> impl Future<Output = Result<Option<Self::Message>, ExitReason>> + Send

Invoked to handle synchronous call messages. call will block until a reply is received (unless the call times out or nodes are disconnected).

from is a struct containing the callers Pid and a Reference that uniquely identifies the call.

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety", so this trait is not object safe.

Implementors

impl GenServer for Registry

type Message = RegistryMessage

impl GenServer for Supervisor

type Message = SupervisorMessage