How to save the result by deserializing Serde Zero-copy futures Hyper Chunk?

I am using futures, tokio, hyper and serde_json to query and deserialize some data that I need to save before my next request. My initial thought was to create a structure containing hyper::Chunk

, and deserialized data that was borrowed from Chunk

but couldn't use the lifetime correctly. I tried using a rental box , but I can't seem to get this to work. I may be using the lifetime 'buffer

before I declare the buffer Vec

, but perhaps I am confusing something else:

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

      

        
        
        
      

    

Is there a way to make the rights of life correct or should I just use DeserializeOwned

and discard the null copy?

For more context, the following code works (periodically deserializing the JSON from two urls, saving the results so we can do something with them). I would like to change the types X

and Y

to use Cow<'a, str>

for my fields, changing from DeserializeOwned

to Deserialize<'a>

. For this to work, I need to save the slice that has been deserialized for each, but I don't know how. I'm looking for examples that use Serde null copy deserialization and keep the result, or some ideas for restructuring my code to work.

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