а, ну понятно

оно

смотри, ты спавнишь фьючу, и сразу же после этого у тебя программа завершается

spawn же не ждёт её выполнения

а, точно, что-то забыл про это

ок, спасибо

    let handle = tokio::spawn(async move {
        bot.launch().await.expect("bot failed");
    });
    handle.await;

скажем, такой вариант уже стабильно работает

я хотел дальше из канала читать, но решил затестить и не заметил, что идет выход из программы сразу

бывает)

Добро утро, коллеги! Помоги, пожалуйста, подружить tokio и sqlx. Есть такой стрим:

let stream =
        sqlx::query("SELECT content FROM trade WHERE created_at >= $1 AND created_at < $2 ORDER BY id")
            .bind(begin)
            .bind(end)
            .fetch(&pool);

    trade_streaming(stream).await;

В trade_streaming хотелось бы его обработать:

async fn trade_streaming(mut stream: BoxStream<Result<PgRow, Error>>) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        while let Some(row) = stream.try_next().await.unwrap() {
            // что-то читаем из row
        }
    })
}

Но компилятор ругается на время жизни:

error[E0726]: implicit elided lifetime not allowed here
  --> pr/src/db/reading.rs:42:38
   |
42 | async fn trade_streaming(mut stream: BoxStream<Result<PgRow, Error>>) -> tokio::task::JoinHandle<()> {
   |                                      ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ help: indicate the anonymous lifetime: `BoxStream<'_, Result<PgRow, Error>>`

Попытки завернуть stream в Arc и Mutex не помогают.

Что можно сдлеать?

Там в help написано - BoxStream<'_, Result...

И? Что дальше?)

После исправления ошибка сохранилась?

> Alex V
Конечно, сохранилась) Если бы это было так просто... Там проблема со static временем:

error[E0759]: `stream` has an anonymous lifetime `'_` but it needs to satisfy a `'static` lifetime requirement
  --> pr/src/db/reading.rs:42:26
   |
42 | async fn trade_streaming(mut stream: BoxStream<'_, Result<PgRow, Error>>) -> tokio::task::JoinHandle<()> {
   |                          ^^^^^^^^^^  ----------------------------------- this data with an anonymous lifetime `'_`...
   |                          |
   |                          ...is captured here...
43 |     tokio::spawn(async move {
   |     ------------ ...and is required to live as long as `'static` here

То есть сам BoxStream не является корректной сущностью, которую можно безопасно перемещать в tokio::spawn (как я понимаю ситуацию).

Может его как-то в канал обернуть?!

В принципе я нашел решение:

let conn = pool.try_acquire().unwrap();
trade_streaming(conn, date).await;

async fn trade_streaming(mut conn: PoolConnection<Postgres>, date: NaiveDate) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        let begin = date.and_hms(0, 0, 0);
        let end = begin.add(Duration::days(1));

        let mut stream =
            sqlx::query("SELECT content FROM trade WHERE created_at >= $1 AND created_at < $2 ORDER BY id")
                .bind(begin)
                .bind(end)
                .fetch(&mut conn);

        while let Some(row) = stream.try_next().await.unwrap() {
            // ...
        }
    })
}

То есть поместить сам поток в spawn, не идеально, конечно, но почему бы и нет.

Если у кого-то есть соображения на тему как передать BoxStream в tokio::spawn, буду весьма признателен.

P.S. BoxStream: https://docs.rs/futures-core/0.3.14/futures_core/stream/type.BoxStream.html