Mercurial > templog
view rust/src/params.rs @ 633:490e9e15b98c rust
move some bits to riker
author | Matt Johnston <matt@ucc.asn.au> |
---|---|
date | Wed, 04 Sep 2019 23:24:13 +0800 |
parents | bde302def78e |
children | a5721c02d3ee |
line wrap: on
line source
use std::time::Duration; use std::io; use std::str; use std::rc::Rc; use std::sync::{Arc,Mutex}; use std::error::Error; use std::cell::{Cell,RefCell}; use std::fs::File; use std::io::Read; use serde::{Serialize,Deserialize}; use rand::rngs::{StdRng, OsRng}; use rand::{RngCore, SeedableRng}; use std::str::FromStr; use hyper; use hyper::client::Client; use riker::actors::*; use super::types::*; use super::config::Config; #[derive(Deserialize, Serialize, Debug, Clone)] pub struct Params { pub fridge_setpoint: f32, pub fridge_difference: f32, pub overshoot_delay: u64, pub overshoot_factor: f32, pub disabled: bool, pub nowort: bool, pub fridge_range_lower: f32, pub fridge_range_upper: f32, } impl Params { pub fn defaults() -> Params { Params { fridge_setpoint: 16.0, fridge_difference: 0.2, overshoot_delay: 720, // 12 minutes overshoot_factor: 1.0, disabled: false, nowort: false, fridge_range_lower: 3.0, fridge_range_upper: 3.0, } } fn try_load(filename: &str) -> Result<Params, TemplogError> { let mut s = String::new(); File::open(filename)?.read_to_string(&mut s)?; Ok(serde_json::from_str(&s)?) } pub fn load(config: &Config) -> Params { Self::try_load(&config.PARAMS_FILE) .unwrap_or_else(|_| Params::defaults()) } } pub struct ParamWaiter { limitlog: NotTooOften, // last_etag is used for long-polling. last_etag: RefCell<String>, epoch: String, chan: ChannelRef<Params>, config: Config, } const LOG_MINUTES: u64 = 15; const MAX_RESPONSE_SIZE: usize = 10000; const TIMEOUT_MINUTES: u64 = 5; impl ParamWaiter { fn new(config: Config) -> Self { let mut b = [0u8; 15]; // 15 bytes -> 20 characters base64 OsRng.fill_bytes(&mut b); let epoch = base64::encode(&b); ParamWaiter { limitlog: NotTooOften::new(LOG_MINUTES*60), last_etag: RefCell::new(String::new()), epoch: epoch, config: config, } } async fn keep_waiting(&mut self) { loop { self.wait_updates().await; } } async fn wait_updates(&mut self) { let uri = self.config.SETTINGS_URL.parse().expect("Bad SETTINGS_URL in config"); let mut req = hyper::Request::new(hyper::Method::Get, uri); req.headers_mut().insert(hyper::header::ETAG, self.last_etag.borrow()); let resp = hyper::Client::new(&self.handle).request(req).await?; let b = resp.body().concat2().await?; let new_params = self.handle_response(b)?; self.chan.tell(Publish{msg: new_params, topic: "params".into()}, None); } fn handle_response(&self, buf : hyper::Chunk, status: hyper::StatusCode) -> Result<Params, TemplogError> { #[derive(Deserialize, Debug)] struct Response { // sent as an opaque etag: header. Has format "epoch-nonce", // responses where the epoch do not match ParamWaiter::epoch are dropped etag: String, params: Params, } 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; // 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))) } hyper::StatusCode::NotModified => { // XXX this isn't really an error. Should handle_response() return // Result<Option<Params>, TemplogError> instead? Err(TemplogError::new("304 unmodified (long polling timeout at the server)")) }, _ => { Err(TemplogError::new(&format!("Wrong server response code {}: {}", status.as_u16(), text))) }, } } fn write_params(&self, params: &Params) { let p = atomicwrites::AtomicFile::new(&self.config.PARAMS_FILE, atomicwrites::AllowOverwrite); p.write(|f| { serde_json::to_writer(f, params) }); } } impl Actor for ParamWaiter { fn post_start(&mut self, ctx: &Context<Self::Msg>) { self.chan = channel("params", &ctx.system).unwrap(); ctx.run(self.wait_updates()); } } // pub fn stream(config: Config, handle: Handle) -> Result<(), TemplogError> { // let rcself = Rc::new(ParamWaiter::new(config, handle)); // let dur = Duration::from_millis(4000); // for _ in Interval::new(dur, &rcself.handle).unwrap() { // // fetch params // // TODO - skip if inflight. // let r = await!(rcself.make_request()).map_err(|e| TemplogError::new_hyper("response", e))?; // let status = r.status(); // let b = await!(r.body().concat2()).map_err(|e| TemplogError::new_hyper("body", e))?; // if let Ok(params) = rcself.handle_response(b, status) { // stream_yield!(params); // } // } // Ok(()) // }