4

我正在使用 futures、tokio、hyper 和 serde_json 来请求和反序列化一些我需要保留到下一个请求之前的数据。我最初的想法是制作一个结构，其中包含hyper::Chunk从 中借用的 和反序列化数据Chunk，但无法获得正确的生命周期。我尝试使用出租箱，但我也无法让它工作。也许我'buffer在声明缓冲区之前使用了生命周期Vec，但也许我搞砸了其他东西：

#[rental]
pub struct ChunkJson<T: serde::de::Deserialize<'buffer>> {
    buffer: Vec<u8>,
    json: T
}

有什么方法可以使生命周期正确，还是我应该使用DeserializeOwned并放弃零拷贝？

对于更多上下文，以下代码有效（定期从两个 URL 反序列化 JSON，保留结果，以便我们可以对它们都做一些事情）。我想更改我的X和Y类型以Cow<'a, str>用于他们的字段，从更改DeserializeOwned为Deserialize<'a>. 为此，我需要存储每个已反序列化的切片，但我不知道该怎么做。我正在寻找使用 Serde 的零拷贝反序列化并保留结果的示例，或者一些重组我的代码的想法。

#[macro_use]
extern crate serde_derive;

extern crate serde;
extern crate serde_json;
extern crate futures;
extern crate tokio_core;
extern crate tokio_periodic;
extern crate hyper;

use std::collections::HashMap;
use std::error::Error;

use futures::future;
use futures::Future;
use futures::stream::Stream;
use hyper::Client;


fn stream_json<'a, T: serde::de::DeserializeOwned + Send + 'a>
    (handle: &tokio_core::reactor::Handle,
     url: String,
     period: u64)
     -> Box<Stream<Item = T, Error = Box<Error>> + 'a> {
    let client = Client::new(handle);
    let timer = tokio_periodic::PeriodicTimer::new(handle).unwrap();
    timer
        .reset(::std::time::Duration::new(period, 0))
        .unwrap();
    Box::new(futures::Stream::zip(timer.from_err::<Box<Error>>(), futures::stream::unfold( (), move |_| {
            let uri = url.parse::<hyper::Uri>().unwrap();
            let get = client.get(uri).from_err::<Box<Error>>().and_then(|res| {
                res.body().concat().from_err::<Box<Error>>().and_then(|chunks| {
                    let p: Result<T, Box<Error>> = serde_json::from_slice::<T>(chunks.as_ref()).map_err(|e| Box::new(e) as Box<Error>);
                    match p {
                        Ok(json) => future::ok((json, ())),
                        Err(err) => future::err(err)
                    }
                })
            });
            Some(get)
        })).map(|x| { x.1 }))
}

#[derive(Serialize, Deserialize, Debug)]
pub struct X {
    foo: String,
}

#[derive(Serialize, Deserialize, Debug)]
pub struct Y {
    bar: String,
}

fn main() {

    let mut core = tokio_core::reactor::Core::new().unwrap();
    let handle = core.handle();

    let x_stream = stream_json::<HashMap<String, X>>(&handle, "http://localhost/X".to_string(), 2);
    let y_stream = stream_json::<HashMap<String, Y>>(&handle, "http://localhost/Y".to_string(), 5);
    let mut xy_stream = x_stream.merge(y_stream);

    let mut last_x = HashMap::new();
    let mut last_y = HashMap::new();

    loop {
        match core.run(futures::Stream::into_future(xy_stream)) {
            Ok((Some(item), stream)) => {
                match item {
                    futures::stream::MergedItem::First(x) => last_x = x,
                    futures::stream::MergedItem::Second(y) => last_y = y,
                    futures::stream::MergedItem::Both(x, y) => {
                        last_x = x;
                        last_y = y;
                    }
                }
                println!("\nx = {:?}", &last_x);
                println!("y = {:?}", &last_y);
                // Do more stuff with &last_x and &last_y

                xy_stream = stream;
            }
            Ok((None, stream)) => xy_stream = stream,
            Err(_) => {
                panic!("error");
            }
        }
    }
}
4

1 回答 1

7

在尝试解决复杂的编程问题时，尽可能多地删除是非常有用的。拿走您的代码并删除您可以删除的内容，直到问题消失。稍微调整一下你的代码并继续删除，直到你不能再删除为止。然后，扭转问题并从最小的部分开始构建并返回错误。做这两件事会告诉你问题出在哪里。

首先，让我们确保我们正确反序列化：

extern crate serde;
extern crate serde_json;
#[macro_use]
extern crate serde_derive;

use std::borrow::Cow;

#[derive(Debug, Deserialize)]
pub struct Example<'a> {
    #[serde(borrow)]
    name: Cow<'a, str>,
    key: bool,
}

impl<'a> Example<'a> {
    fn info(&self) {
        println!("{:?}", self);
        match self.name {
            Cow::Borrowed(_) => println!("Is borrowed"),
            Cow::Owned(_) => println!("Is owned"),
        }
    }
}

fn main() {
    let data: Vec<_> = br#"{"key": true, "name": "alice"}"#.to_vec();

    let decoded: Example = serde_json::from_slice(&data).expect("Couldn't deserialize");
    decoded.info();
}

在这里，我忘了添加#[serde(borrow)]属性，所以我很高兴我做了这个测试！

接下来，我们可以介绍一下租赁板条箱：

#[macro_use]
extern crate rental;

rental! {
    mod holding {
        use super::*;

        #[rental]
        pub struct VecHolder {
            data: Vec<u8>,
            parsed: Example<'data>,
        }
    }
}

fn main() {
    let data: Vec<_> = br#"{"key": true, "name": "alice"}"#.to_vec();

    let holder = holding::VecHolder::try_new(data, |data| {
        serde_json::from_slice(data)
    });
    let holder = match holder {
        Ok(holder) => holder,
        Err(_) => panic!("Unable to construct rental"),
    };

    holder.rent(|example| example.info());

    // Make sure we can move the data and it's still valid
    let holder2 = { holder };
    holder2.rent(|example| example.info());
}

接下来我们尝试创建一个出租Chunk：

#[rental]
pub struct ChunkHolder {
    data: Chunk,
    parsed: Example<'data>,
}

不幸的是，这失败了：

  --> src/main.rs:29:1
   |
29 | rental! {
   | ^
   |
   = help: message: Field `data` must have an angle-bracketed type parameter or be `String`.

哎呀！检查出租的文档，我们可以添加#[target_ty_hack="[u8]"]到该data字段。这将导致：

error[E0277]: the trait bound `hyper::Chunk: rental::__rental_prelude::StableDeref` is not satisfied
  --> src/main.rs:29:1
   |
29 | rental! {
   | ^ the trait `rental::__rental_prelude::StableDeref` is not implemented for `hyper::Chunk`
   |
   = note: required by `rental::__rental_prelude::static_assert_stable_deref`

这很烦人；因为我们不能实现 trait Chunk，我们只需要 box Chunk，证明它有一个稳定的地址：

#[rental]
pub struct ChunkHolder {
    data: Box<Chunk>,
    parsed: Example<'data>,
}

我还查看了是否有办法让Vec<u8>退出Chunk，但它似乎不存在。那将是另一种分配和间接较少的解决方案。

在这一点上，剩下的“所有”就是将其集成回期货代码中。除了你重新创建它之外，任何人都需要做很多工作，但我认为这样做不会有任何明显的问题。

于 2017-04-30T15:16:29.817 回答