Message handlers

We've already seen that workers are basically components without widgets. In this chapter, we will talk about message handlers that are even simpler: like workers but without a model.

The motivation

You might wonder why we even need message handlers. Components and workers are already some kind of message handlers, right? That's true, but components and workers do more than just handling messages: they also have a model that represents their state.

The problem with the state is that Rust doesn't like sharing mutable data. Only one mutable reference can exist at the time to prevent race conditions and other bugs. However, both components and workers can update their state in the update function while handling messages. This means that components and workers can only handle one message at the time. Otherwise, there would be multiple mutable references to the model.

Handling one message at the time is perfectly fine in most cases. However, if you, for example, want to handle a lot of HTTP requests and you send one message to a worker for each request you want to handle, that'd mean that one message is sent after another. This could cause a huge delay. Fortunately, message handlers can solve this issue.

Implementing a message handler

To keep it simple, we will create another counter app. Yet this time, every click will be delayed by one second. If a user clicks the increment button, the counter will be incremented exactly one second later.

The app we will write in this chapter is also available here. Run cargo run --example non_blocking_async from the example directory if you want to see the code in action.

The timing

Let's have a look at a simple timing diagram that shows what would happen if we used a worker for our app.

All three clicks happen in one second. But because we can only handle one message at a time in a worker, we have to wait one second, only for processing the first message. The second and the third message are then handled too late because our worker was blocking while handling the first message (see the striped parts in the diagram).

But how would our ideal timing diagram look like?

In the second diagram, there's no blocking. The second and the third message are handled instantly, so they can increment the counter exactly one second after the user clicked the button for the second and third time.

Alright, let's implement it!

The includes

In this example, the includes are a little special because we have two kinds of senders. We've already seen relm4::Sender (aka glib::Sender) several times as it's used by Relm4 to send messages to components and workers. The other one is tokio::sync::mpsc::Sender, the sender we use for the message handler. We could use any sender type we want for the message handler because we're implementing all of the message handling ourselves. Yet, because we want a sender that supports async Rust, the sender from tokio is a reasonable choice.

Since both senders are called Sender by default we rename the latter to TokioSender in the last line of the includes.

use gtk::prelude::{BoxExt, ButtonExt, GtkWindowExt, OrientableExt};
use relm4::{
    send, AppUpdate, Components, MessageHandler, Model, RelmApp, RelmMsgHandler, Sender,
    WidgetPlus, Widgets,
};

use tokio::runtime::{Builder, Runtime};
use tokio::sync::mpsc::{channel, Sender as TokioSender};

Relm4 runs updates for workers and components on the glib main event loop that's provided by GTK. Therefore, Relm4 uses relm4::Sender aka glib::Sender to send messages to workers and components.

The model

The model and the message type are the same as in our first app.

struct AppModel {
    counter: u8,
}

#[derive(Debug)]
enum AppMsg {
    Increment,
    Decrement,
}

The update function is identical, too.

impl AppUpdate for AppModel {
    fn update(
        &mut self,
        msg: AppMsg,
        _components: &AppComponents,
        _sender: Sender<AppMsg>,
    ) -> bool {
        match msg {
            AppMsg::Increment => {
                self.counter = self.counter.wrapping_add(1);
            }
            AppMsg::Decrement => {
                self.counter = self.counter.wrapping_sub(1);
            }
        }
        true
    }
}

The message handler

Our message handler needs to store our sender and also the tokio runtime we use to process our messages (to keep the runtime alive).

And of course, we need a message type as well.

struct AsyncHandler {
    _rt: Runtime,
    sender: TokioSender<AsyncHandlerMsg>,
}

#[derive(Debug)]
enum AsyncHandlerMsg {
    DelayedIncrement,
    DelayedDecrement,
}

Then we need to implement the MessageHandler trait for our message handler.

impl MessageHandler<AppModel> for AsyncHandler {
    type Msg = AsyncHandlerMsg;
    type Sender = TokioSender<AsyncHandlerMsg>;

    fn init(_parent_model: &AppModel, parent_sender: Sender<AppMsg>) -> Self {
        let (sender, mut rx) = channel::<AsyncHandlerMsg>(10);

        let rt = Builder::new_multi_thread()
            .worker_threads(8)
            .enable_time()
            .build()
            .unwrap();

        rt.spawn(async move {
            while let Some(msg) = rx.recv().await {
                let parent_sender = parent_sender.clone();
                tokio::spawn(async move {
                    tokio::time::sleep(std::time::Duration::from_secs(1)).await;
                    match msg {
                        AsyncHandlerMsg::DelayedIncrement => {
                            send!(parent_sender, AppMsg::Increment);
                        }
                        AsyncHandlerMsg::DelayedDecrement => {
                            send!(parent_sender, AppMsg::Decrement);
                        }
                    }
                });
            }
        });

        AsyncHandler { _rt: rt, sender }
    }

    fn send(&self, msg: Self::Msg) {
        self.sender.blocking_send(msg).unwrap();
    }

    fn sender(&self) -> Self::Sender {
        self.sender.clone()
    }
}

First we define the message type. Then we specify the sender type. You could, for example, use std::sync::mpsc::Sender, tokio::sync::mpsc::Sender or any other sender type you want.

The init function simply initializes the message handler. In the first part, we create a new tokio runtime that will process our messages. Then we check for messages in a loop.

            while let Some(msg) = rx.recv().await {

When using components and workers, this loop runs in the background. Here we need to define it ourselves. The important part here is the await. Because we wait for new messages here, the tokio runtime can process our messages in the meantime. Therefore, we can handle a lot of messages at the same time.

If you want to learn more about async in Rust, you can find more information here.

Inside the loop, we process the message by waiting one second and then sending a message back to the parent component.

The send method defines a convenient interface for sending messages to this message handler and the sender method provides a sender to connect events later.

The components

Next, we need to add the message handler to our components. It's very similar to adding workers.

struct AppComponents {
    async_handler: RelmMsgHandler<AsyncHandler, AppModel>,
}

impl Components<AppModel> for AppComponents {
    fn init_components(parent_model: &AppModel, parent_sender: Sender<AppMsg>) -> Self {
        AppComponents {
            async_handler: RelmMsgHandler::new(parent_model, parent_sender),
        }
    }

    fn connect_parent(&mut self, _parent_widgets: &AppWidgets) {}
}

The widgets

The last part we need is the widgets type. It should look familiar except for the events.

#[relm4_macros::widget]
impl Widgets<AppModel, ()> for AppWidgets {
    view! {
            gtk::ApplicationWindow {
                set_title: Some("Simple app"),
                set_default_width: 300,
                set_default_height: 100,
                set_child = Some(&gtk::Box) {
                    set_orientation: gtk::Orientation::Vertical,
                    set_margin_all: 5,
                    set_spacing: 5,

                    append = &gtk::Button {
                        set_label: "Increment",
                        connect_clicked[sender = components.async_handler.sender()] => move |_| {
                            sender.blocking_send(AsyncHandlerMsg::DelayedIncrement)
                                .expect("Receiver dropped");
                        },
                    },
                    append = &gtk::Button::with_label("Decrement") {
                        connect_clicked[sender = components.async_handler.sender()] => move |_| {
                            sender.blocking_send(AsyncHandlerMsg::DelayedDecrement)
                                .expect("Receiver dropped");
                        },
                    },
                    append = &gtk::Label {
                        set_margin_all: 5,
                        set_label: watch! { &format!("Counter: {}", model.counter) },
                    }
                },
            }
        }
}

We're connecting the event directly to the message handler. You could pass the message through the update function of your app and forward it to the message handler, but the macro provides a special syntax to connect events directly.

                        connect_clicked[sender = components.async_handler.sender()] => move |_| {
                            sender.blocking_send(AsyncHandlerMsg::DelayedIncrement)
                                .expect("Receiver dropped");
                        },

You'll notice that we use brackets instead of parentheses here. That tells the macro that we want to connect an event with a sender from a component. The syntax looks like this.

connect_name[sender_name = components.component_name.sender()] => move |_| { ... }

Conclusion

That's it! We've implemented our first message handler.

The complete code

Let's review our code in one piece one more time to see how all these parts work together:

use gtk::prelude::{BoxExt, ButtonExt, GtkWindowExt, OrientableExt};
use relm4::{
    send, AppUpdate, Components, MessageHandler, Model, RelmApp, RelmMsgHandler, Sender,
    WidgetPlus, Widgets,
};

use tokio::runtime::{Builder, Runtime};
use tokio::sync::mpsc::{channel, Sender as TokioSender};

struct AppModel {
    counter: u8,
}

#[derive(Debug)]
enum AppMsg {
    Increment,
    Decrement,
}

impl Model for AppModel {
    type Msg = AppMsg;
    type Widgets = AppWidgets;
    type Components = AppComponents;
}

impl AppUpdate for AppModel {
    fn update(
        &mut self,
        msg: AppMsg,
        _components: &AppComponents,
        _sender: Sender<AppMsg>,
    ) -> bool {
        match msg {
            AppMsg::Increment => {
                self.counter = self.counter.wrapping_add(1);
            }
            AppMsg::Decrement => {
                self.counter = self.counter.wrapping_sub(1);
            }
        }
        true
    }
}

struct AsyncHandler {
    _rt: Runtime,
    sender: TokioSender<AsyncHandlerMsg>,
}

#[derive(Debug)]
enum AsyncHandlerMsg {
    DelayedIncrement,
    DelayedDecrement,
}

impl MessageHandler<AppModel> for AsyncHandler {
    type Msg = AsyncHandlerMsg;
    type Sender = TokioSender<AsyncHandlerMsg>;

    fn init(_parent_model: &AppModel, parent_sender: Sender<AppMsg>) -> Self {
        let (sender, mut rx) = channel::<AsyncHandlerMsg>(10);

        let rt = Builder::new_multi_thread()
            .worker_threads(8)
            .enable_time()
            .build()
            .unwrap();

        rt.spawn(async move {
            while let Some(msg) = rx.recv().await {
                let parent_sender = parent_sender.clone();
                tokio::spawn(async move {
                    tokio::time::sleep(std::time::Duration::from_secs(1)).await;
                    match msg {
                        AsyncHandlerMsg::DelayedIncrement => {
                            send!(parent_sender, AppMsg::Increment);
                        }
                        AsyncHandlerMsg::DelayedDecrement => {
                            send!(parent_sender, AppMsg::Decrement);
                        }
                    }
                });
            }
        });

        AsyncHandler { _rt: rt, sender }
    }

    fn send(&self, msg: Self::Msg) {
        self.sender.blocking_send(msg).unwrap();
    }

    fn sender(&self) -> Self::Sender {
        self.sender.clone()
    }
}

struct AppComponents {
    async_handler: RelmMsgHandler<AsyncHandler, AppModel>,
}

impl Components<AppModel> for AppComponents {
    fn init_components(parent_model: &AppModel, parent_sender: Sender<AppMsg>) -> Self {
        AppComponents {
            async_handler: RelmMsgHandler::new(parent_model, parent_sender),
        }
    }

    fn connect_parent(&mut self, _parent_widgets: &AppWidgets) {}
}

#[relm4_macros::widget]
impl Widgets<AppModel, ()> for AppWidgets {
    view! {
            gtk::ApplicationWindow {
                set_title: Some("Simple app"),
                set_default_width: 300,
                set_default_height: 100,
                set_child = Some(&gtk::Box) {
                    set_orientation: gtk::Orientation::Vertical,
                    set_margin_all: 5,
                    set_spacing: 5,

                    append = &gtk::Button {
                        set_label: "Increment",
                        connect_clicked[sender = components.async_handler.sender()] => move |_| {
                            sender.blocking_send(AsyncHandlerMsg::DelayedIncrement)
                                .expect("Receiver dropped");
                        },
                    },
                    append = &gtk::Button::with_label("Decrement") {
                        connect_clicked[sender = components.async_handler.sender()] => move |_| {
                            sender.blocking_send(AsyncHandlerMsg::DelayedDecrement)
                                .expect("Receiver dropped");
                        },
                    },
                    append = &gtk::Label {
                        set_margin_all: 5,
                        set_label: watch! { &format!("Counter: {}", model.counter) },
                    }
                },
            }
        }
}

fn main() {
    let model = AppModel { counter: 0 };
    let app = RelmApp::new(model);
    app.run();
}