如何使用已启用Futures的Hyper Chunk的Serde Zero-copy反序列化来存储结果？

Question

我正在使用期货,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");
            }
        }
    }
}

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>,
}

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

此时,剩下的"全部"是将其整合回期货代码.除了你重新创建之外,任何人都可以做很多工作,但我没有预见到这样做会有任何明显的问题.