EventStream is not Send

[Occhioverde]

Good morning!

I'm currently trying to create an application with a client-server architecture that needs to read from an EventStream in a separate async task launched with tokio::spawn, but I noticed that the stream is not Send (the type mentioned below by the compiler is HyprlandIPCReader as it owns an instance of EventStream):

error: future cannot be sent between threads safely
   --> src/main.rs:20:18
    |
20  |     tokio::spawn(HyprlandIPCReader::setup(tx).event_loop());
    |                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ future returned by `event_loop` is not `Send`
    |
    = help: the trait `Send` is not implemented for `(dyn futures_lite::Stream<Item = Result<hyprland::event_listener::Event, HyprError>> + 'static)`, which is required by `impl futures_lite::Future<Output = Result<(), anyhow::Error>>: Send`
note: future is not `Send` as this value is used across an await
   --> src/hyprland_ipc_reader.rs:20:12
    |
16  |     pub async fn event_loop(mut self) -> anyhow::Result<()> {
    |                             -------- has type `HyprlandIPCReader` which is not `Send`
...
20  |         }).await
    |            ^^^^^ await occurs here, with `mut self` maybe used later
note: required by a bound in `tokio::spawn`
   --> /home/rsacchetto/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.41.0/src/task/spawn.rs:168:21
    |
166 |     pub fn spawn<F>(future: F) -> JoinHandle<F::Output>
    |            ----- required by a bound in this function
167 |     where
168 |         F: Future + Send + 'static,
    |                     ^^^^ required by this bound in `spawn`

Since I can't seem to understand why is it the case, do you think its implementation can be changed into something that is Send or is there a reason that makes it impossible?

For reference, here is the code that generated the error:

async fn server() -> anyhow::Result<()> {
    let (tx, rx) = broadcast::channel(100);

    log::info!("Launching Hyprland IPC Reader...");
    tokio::spawn(HyprlandIPCReader::setup(tx).event_loop());

    log::info!("Launching Client Acceptor...");
    ClientAcceptor::init(rx)?.main_loop().await
}

use futures_lite::StreamExt;
use hyprland::event_listener::{Event, EventStream};
use tokio::sync::broadcast::Sender;

pub struct HyprlandIPCReader {
    channel: Sender<Event>,
    event_stream: EventStream
}

impl HyprlandIPCReader {
    pub fn setup(channel: Sender<Event>) -> Self {
        let event_stream = EventStream::new();
        Self { channel, event_stream }
    }

    pub async fn event_loop(mut self) -> anyhow::Result<()> {
        self.event_stream.try_for_each(|event| -> anyhow::Result<()> {
            self.channel.send(event?)?;
            Ok(())
        }).await
    }
}

Thank you very much for your time!

[yavko]

It is not Send because the Stream trait object it has, is not limited to be Send

feel free to change this in a PR by making it + Send.

[Occhioverde]

Thank you very much! I thought that there was some specific reason for it to not be Send, but it's better this way.
I cloned the repo, added the + Send constraint, and it compiles just fine. I'll create a Pull Request for it as you suggested.