stream - 在多个闭包中表达变量对的生命周期

Question

我正在努力以一种取悦借用检查员的方式来表达我的代码。

我有一个函数 create_task这创造了一些数据库操作的 future 。有一个值流，其中每个元素都需要插入到事务中的数据库中。问题是在多个闭包之间共享事务，因为它也可变地借用了连接对象。

#![feature(conservative_impl_trait)]

extern crate futures;
extern crate rusqlite;

use futures::prelude::*;
use futures::{future, stream};
use rusqlite::Connection;

fn main() {
    let task = create_task();
    task.wait().unwrap();
}

fn create_task() -> impl Future<Item = (), Error = ()> {
    let mut conn = Connection::open("temp.db").unwrap();
    conn.execute("CREATE TABLE IF NOT EXISTS temp (val INTEGER)", &[]).unwrap();

    // tx takes a mut ref to conn!
    let tx = conn.transaction().unwrap();

    stream::iter_ok::<_, ()>(vec![1, 2, 3])
        .for_each(|val| {
            // tx borrowed here!
            tx.execute("INSERT INTO temp (val) VALUES (?1)", &[&val]).unwrap();
            future::ok(())
        })
        .map(|_| {
            // tx moved/consumed here!
            tx.commit().unwrap();
        })
}

代码存在多个问题:

• conn活得不够长。它也需要移动到闭包。也许作为 Rc<Connection>因为两次关闭？
• conn不能简单地共享为 Rc因为可变性要求。也许Rc<RefCell<Connection>>是更合适的类型？
• 借阅检查员不知道借到tx在第一个 for_each 之后结束关闭，因此它不能移动到第二个 map关闭。再次将其移动为 Rc<Transaction>两个关闭可能是合理的？

我一直在摆弄这些想法，并且知道所需的生命周期是可能的并且有意义，但无法以正确的方式向编译器表达我的代码。

Answer 1

我相信您的第一个问题是您还没有完全理解 futures 是多么懒惰。您正在 create_task 中创建一个 Connection，引用它，将该引用放入流/ future ，然后尝试返回该 future 。 此时甚至还没有执行任何闭包。

你 cannot return a reference to a value created in a function .不要尝试 store the transaction and the connection in the same struct, either .

相反，接受对 Connection 的引用并返回包含该生命周期的 Future。

下一个问题是编译器不知道如何调用闭包或以什么顺序调用。与其试图关闭交易，不如让它从一个交易“流动”到另一个交易，让所有权系统确保交易始终在正确的位置。

#![feature(conservative_impl_trait)]

extern crate futures;
extern crate rusqlite;

use futures::prelude::*;
use futures::{future, stream};
use rusqlite::Connection;

fn main() {
    let mut conn = Connection::open("temp.db").unwrap();
    conn.execute("CREATE TABLE IF NOT EXISTS temp (val INTEGER)", &[]).unwrap();

    let task = create_task(&mut conn);
    task.wait().unwrap();
}

fn create_task<'a>(conn: &'a mut rusqlite::Connection) -> impl Future<Item = (), Error = ()> + 'a {
    let tx = conn.transaction().unwrap();
    stream::iter_ok::<_, ()>(vec![1, 2, 3])
        .fold(tx, |tx, val| {
            tx.execute("INSERT INTO temp (val) VALUES (?1)", &[&val]).unwrap();
            future::ok(tx)
        })
        .map(move |tx| {
            tx.commit().unwrap();
        })
}

一个巨大的警告:如果 execute 不是异步的，你真的不应该在这样的 future 中使用它。任何阻塞操作都会导致你所有的 future 停滞不前。您可能应该在单独的线程/线程池上运行同步工作负载。

另见:

• Is there any way to return a reference to a variable created in a function?
• How to store a SqliteConnection and SqliteStatement objects in the same struct in Rust?
• Why can't I store a value and a reference to that value in the same struct?