我如何使用支持期货的超块的Serde Zero Zero拷贝避难所化

Question

我正在使用Futures，Tokio，Hyper和Serde_json来请求和估算一些我需要保留的数据，直到下一个请求。我最初的想法是制作一个包含hyper::Chunk的结构和从Chunk借用的供应数据，但无法正确实现生命。我尝试使用租赁板条箱，但我也无法正常工作。也许我在声明缓冲区Vec之前使用'buffer寿命，但也许我把其他东西弄乱了：

#[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");
            }
        }
    }
}

Answer 1

试图解决复杂的编程问题时，尽可能多地删除非常有用。拿代码并删除问题，直到问题消失。对您的代码进行一些调整，并继续删除，直到您再也不能。然后，将问题扭转并从最小的一块构建，然后回到错误。这样做两个都会向您展示问题所在的位置。

首先，让我们确保我们正确化：

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`

这很烦人；由于我们无法为Chunk实施该性状，因此我们只需要箱Chunk，证明它具有稳定的地址：

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

我还希望查看是否有一种方法可以从Chunk中获得Vec<u8>，但似乎不存在。那本来是另一种解决方案，分配和间接较少。

在这一点上，剩下的"所有"都是将其集成到期货代码中。除了您要重新创建它，除了您之外，这都是很多工作，但是我没有任何明显的问题。