Mercurial > templog
diff rust/src/params.rs @ 626:efcbe0d3afd6 rust
fix to work with hyper
author | Matt Johnston <matt@ucc.asn.au> |
---|---|
date | Wed, 06 Dec 2017 00:09:45 +0800 |
parents | 2710649ab71e |
children | d5075136442f |
line wrap: on
line diff
--- a/rust/src/params.rs Wed Dec 06 00:08:46 2017 +0800 +++ b/rust/src/params.rs Wed Dec 06 00:09:45 2017 +0800 @@ -17,11 +17,12 @@ use tokio_core::reactor::Interval; use tokio_core::reactor::Handle; -use tokio_curl::Session; use futures::{Stream,Future,future}; -use curl::easy::Easy; -use curl::easy; use self::rand::Rng; +use std::str::FromStr; +use hyper; +use hyper::header::{Headers, ETag, EntityTag}; +use hyper::client::Client; use types::*; use ::Config; @@ -77,7 +78,6 @@ limitlog: NotTooOften, // last_etag is used for long-polling. last_etag: RefCell<String>, - session: RefCell<Option<Session>>, epoch: String, config: Config, @@ -89,63 +89,43 @@ const TIMEOUT_MINUTES: u64 = 5; impl ParamWaiter { - fn make_request(&self) -> (Easy, Arc<Mutex<Vec<u8>>>) { - let mut req = Easy::new(); - req.get(true).unwrap(); - // supposedly req.url won't fail, checking happens later? - req.url(&self.config.SETTINGS_URL).unwrap(); - - req.timeout(Duration::new(TIMEOUT_MINUTES * 60, 0)).unwrap(); - - // store response - let max_response_size = 10000; - let buf = Arc::new(Mutex::new(Vec::new())); - let dst = buf.clone(); - req.write_function(move |data| { - let mut dst = dst.lock().unwrap(); - dst.extend_from_slice(data); - if dst.len() > max_response_size { - error!("Too large params response from server: {}", dst.len()); - Ok(0) - } else { - Ok(data.len()) - } - }).unwrap(); - - // http header - let e = self.last_etag.borrow(); - if !e.is_empty() { - let mut list = easy::List::new(); - let hd = format!("etag: {}", *e); - list.append(&hd).unwrap(); - req.http_headers(list).unwrap(); + fn make_request(&self) -> hyper::client::FutureResponse { + let uri = self.config.SETTINGS_URL.parse().expect("Bad SETTINGS_URL in config"); + let mut req = hyper::client::Request::new(hyper::Method::Get, uri); + { + let mut headers = req.headers_mut(); + headers.set(hyper::header::ETag(EntityTag::new(false, self.last_etag.borrow().clone()))); } - - (req, buf) + hyper::client::Client::new(&self.handle).request(req) + // XXX how do we do this? + // req.timeout(Duration::new(TIMEOUT_MINUTES * 60, 0)).unwrap(); } - fn handle_response(&self, req: &mut Easy, buf: &Vec<u8>) -> Result<Params, TemplogError> { - let text = String::from_utf8_lossy(buf).to_string(); - let resp = req.response_code()?; - match resp { - 200 => { + + fn handle_response(&self, buf : hyper::Chunk, status: hyper::StatusCode) -> Result<Params, TemplogError> { + let text = String::from_utf8_lossy(buf.as_ref()); + match status { + hyper::StatusCode::Ok => { // new params let r: Response = serde_json::from_str(&text)?; let mut le = self.last_etag.borrow_mut(); *le = r.etag; - if le.split('-').next().unwrap() == &self.epoch { - self.write_params(&r.params); - Ok(r.params) - } else { - Err(TemplogError::new(&format!("Bad epoch from server '{}' expected '{}'", *le, self.epoch))) + + // update params if the epoch is correct + if let Some(e) = le.split('-').next() { + if e == &self.epoch { + self.write_params(&r.params); + return Ok(r.params); + } } + Err(TemplogError::new(&format!("Bad epoch from server '{}' expected '{}'", *le, self.epoch))) } - 304 => { + hyper::StatusCode::NotModified => { // XXX this isn't really an error Err(TemplogError::new("304 unmodified (long polling timeout at the server)")) }, _ => { - Err(TemplogError::new(&format!("Wrong server response code {}: {}", resp, text))) + Err(TemplogError::new(&format!("Wrong server response code {}: {}", status.as_u16(), text))) }, } } @@ -158,31 +138,27 @@ } fn step(rself: Rc<Self>) -> Box<Future<Item=Params, Error=TemplogError>> { - - if rself.session.borrow().is_none() { - *rself.session.borrow_mut() = Some(Session::new(rself.handle.clone())); - } - - let (mut req, buf) = rself.make_request(); + let resp = rself.make_request(); - let ses = rself.session.borrow().clone().unwrap(); - let s = ses.perform(req) - .map_err(|e| TemplogError::new_io("tokio curl error", e.into_error())) - .and_then(move |mut rq| { - let result = buf.lock().unwrap(); - rself.handle_response(&mut rq, &result) - }); + let s = resp.map_err(|e| TemplogError::new_hyper("response", e)) + .and_then(move |r| { + let status = r.status(); + r.body().concat2() + .map_err(|e| TemplogError::new_hyper("body", e)) + .and_then(move |b| { + rself.handle_response(b, status) + }) + }); Box::new(s) } fn new(config: &Config, handle: &Handle) -> Self { let mut b = [0u8; 15]; // 15 bytes -> 20 characters base64 - rand::OsRng::new().unwrap().fill_bytes(&mut b); + rand::OsRng::new().expect("Opening system RNG failed").fill_bytes(&mut b); let epoch = base64::encode(&b); ParamWaiter { limitlog: NotTooOften::new(LOG_MINUTES*60), last_etag: RefCell::new(String::new()), - session: RefCell::new(None), epoch: epoch, config: config.clone(), handle: handle.clone(),