我正在使用 futures、tokio、hyper 和 serde_json 请求和反序列化一些数据，这些数据需要保留直到我的下一个请求。我的初始想法是创建一个包含hyper::Chunk和从Chunk中借用的反序列化数据的结构体，但是无法正确处理生命周期。我尝试使用rental crate，但也无法使其正常工作。也许我在声明缓冲区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");
}
}
}
}