spawning a task referencing self with anonymous lifetime

[barkanido]

Hi, I am trying to implement a Kafka consumer (with rust-rdkafk) that produces messages into a flume channel:

use flume::Sender;
use futures::TryStreamExt;
use rdkafka::consumer::{stream_consumer::StreamConsumer, MessageStream};
use rdkafka::{admin::ConfigSource, config::ClientConfig};

pub struct Click;

pub struct ClicksConsumer {
    consumer: StreamConsumer,
    tx: Sender<Click>,
}

impl ClicksConsumer {
    pub fn new(brokers: String, group_id: String, tx: Sender<Click>) -> ClicksConsumer {
        let consumer: StreamConsumer = ClientConfig::new()
            .set("group.id", &group_id)
            .set("bootstrap.servers", &brokers)
            .set("enable.partition.eof", "false")
            .set("session.timeout.ms", "6000")
            .set("enable.auto.commit", "false")
            .create()
            .expect("Consumer creation failed");
        ClicksConsumer { consumer, tx }
    }

    pub async fn start(&self) {
        tokio::spawn(async move {
            let stream = self.consumer.start();
            stream.try_for_each(|burrwed_msg| {
                let message = burrwed_msg.detach();
                async move {
                    self.tx.send(Click {}); // this is blocking the task. is that a good idea?
                    Ok(())
                }
            })
        });
    }
}

I am getting this error from the compiler which I don't know how ti interpret:
error[E0759]: self has an anonymous lifetime '_ but it needs to satisfy a 'static lifetime requirement

  --> src/consumer.rs:33:24
   |
33 |     pub async fn start(&self) {
   |                        ^^^^^
   |                        |
   |                        this data with an anonymous lifetime `'_`...
   |                        ...is captured here...
34 |         tokio::spawn(async move {
   |         ------------ ...and is required to live as long as `'static` here

This ClicksConsumer is held by main for the lifetime of the program, but how do I think the compiler complains that the task may outlive it. Am I right?

[Darksonn]

This is due to Rust's single-ownership principle. To give the new task ownership of the ClicksConsumer, that task must be the only place that can access it, however the start method takes &self, which means that start only has borrowed access to the ClickConsumer. Since start does not have ownership, it cannot give away ownership to the new task.

One approach is to change start to take ownership of self. This looks like the following:

pub async fn start(self) {
    tokio::spawn(async move {
        let stream = self.consumer.start();
        stream.try_for_each(|burrwed_msg| {
            let message = burrwed_msg.detach();
            async move {
                self.tx.send(Click {});
                Ok(())
            }
        })
    });
}

An alternative is to move the tokio::spawn outside the start method like this:

pub async fn start(&mut self) {
    let stream = self.consumer.start();
    stream.try_for_each(|burrwed_msg| {
        let message = burrwed_msg.detach();
        async move {
            self.tx.send(Click {}); // this is blocking the task. is that a good idea?
            Ok(())
        }
    })
}

tokio::spawn(async move { consumer.start() });

As for the send method being blocking, that is indeed a very bad idea. You should never block the thread in async code. I have written a blog post about this topic: Async: What is blocking?

Luckily you are using the flume crate's Sender, which has an send_async method you can use instead.

---

Regarding start being called in main, that is irrelevant. The compiler only looks inside one function at the time when it verifies correctness, and it would be invalid to call start somewhere it doesn't live forever, hence it does not compile. Additionally, the main function is not considered special, and things that live until the end of main are not considered 'static.

---

As a word of advice, I typically recommend using a real loop rather than for_each or try_for_each. This makes some things simpler lifetime-wise.

use futures::stream::StreamExt;

let stream = self.consumer.start();
while let Some(burrwed_msg) = stream.next().await {
    let message = burrwed_msg.detach();
    self.tx.send_async(Click {}).await;
}