Mercurial > templog
annotate rust/src/params.rs @ 634:a5721c02d3ee rust
build succeeds
| author | Matt Johnston <matt@ucc.asn.au> |
|---|---|
| date | Sun, 22 Sep 2019 20:35:40 +0800 |
| parents | 490e9e15b98c |
| children | 4424a8b30f9c |
| rev | line source |
|---|---|
| 592 | 1 use std::time::Duration; |
| 634 | 2 |
|
611
f3e39e2107fd
still doesn't compile, improvements to TemplogError and tokio curl though
Matt Johnston <matt@ucc.asn.au>
parents:
609
diff
changeset
|
3 use std::str; |
| 634 | 4 |
| 5 | |
| 6 | |
| 7 use std::cell::{RefCell}; | |
|
624
2710649ab71e
read/write params local file. untested
Matt Johnston <matt@ucc.asn.au>
parents:
621
diff
changeset
|
8 use std::fs::File; |
|
2710649ab71e
read/write params local file. untested
Matt Johnston <matt@ucc.asn.au>
parents:
621
diff
changeset
|
9 use std::io::Read; |
| 627 | 10 |
| 633 | 11 use serde::{Serialize,Deserialize}; |
| 592 | 12 |
| 634 | 13 use rand::rngs::{OsRng}; |
| 14 use rand::{RngCore}; | |
| 15 | |
| 16 | |
| 17 use hyper; | |
| 633 | 18 |
| 634 | 19 // for try_concat() |
| 20 use futures::stream::TryStreamExt; | |
| 21 use futures::executor::block_on; | |
| 22 // for block_on().or_else | |
| 23 use futures_util::try_future::TryFutureExt; | |
|
611
f3e39e2107fd
still doesn't compile, improvements to TemplogError and tokio curl though
Matt Johnston <matt@ucc.asn.au>
parents:
609
diff
changeset
|
24 |
| 633 | 25 use riker::actors::*; |
| 26 | |
| 27 use super::types::*; | |
| 28 use super::config::Config; | |
| 592 | 29 |
| 615 | 30 #[derive(Deserialize, Serialize, Debug, Clone)] |
| 31 pub struct Params { | |
| 32 pub fridge_setpoint: f32, | |
| 33 pub fridge_difference: f32, | |
| 621 | 34 pub overshoot_delay: u64, |
| 615 | 35 pub overshoot_factor: f32, |
| 36 pub disabled: bool, | |
| 37 pub nowort: bool, | |
| 38 pub fridge_range_lower: f32, | |
| 39 pub fridge_range_upper: f32, | |
| 40 } | |
| 41 | |
| 42 impl Params { | |
| 43 pub fn defaults() -> Params { | |
| 44 Params { | |
| 45 fridge_setpoint: 16.0, | |
| 46 fridge_difference: 0.2, | |
| 47 overshoot_delay: 720, // 12 minutes | |
| 48 overshoot_factor: 1.0, | |
| 634 | 49 disabled: true, |
| 615 | 50 nowort: false, |
| 51 fridge_range_lower: 3.0, | |
| 52 fridge_range_upper: 3.0, | |
| 53 } | |
| 54 } | |
| 55 | |
|
624
2710649ab71e
read/write params local file. untested
Matt Johnston <matt@ucc.asn.au>
parents:
621
diff
changeset
|
56 fn try_load(filename: &str) -> Result<Params, TemplogError> { |
|
2710649ab71e
read/write params local file. untested
Matt Johnston <matt@ucc.asn.au>
parents:
621
diff
changeset
|
57 let mut s = String::new(); |
|
2710649ab71e
read/write params local file. untested
Matt Johnston <matt@ucc.asn.au>
parents:
621
diff
changeset
|
58 File::open(filename)?.read_to_string(&mut s)?; |
|
2710649ab71e
read/write params local file. untested
Matt Johnston <matt@ucc.asn.au>
parents:
621
diff
changeset
|
59 Ok(serde_json::from_str(&s)?) |
|
2710649ab71e
read/write params local file. untested
Matt Johnston <matt@ucc.asn.au>
parents:
621
diff
changeset
|
60 } |
|
2710649ab71e
read/write params local file. untested
Matt Johnston <matt@ucc.asn.au>
parents:
621
diff
changeset
|
61 |
| 615 | 62 pub fn load(config: &Config) -> Params { |
|
624
2710649ab71e
read/write params local file. untested
Matt Johnston <matt@ucc.asn.au>
parents:
621
diff
changeset
|
63 Self::try_load(&config.PARAMS_FILE) |
|
2710649ab71e
read/write params local file. untested
Matt Johnston <matt@ucc.asn.au>
parents:
621
diff
changeset
|
64 .unwrap_or_else(|_| Params::defaults()) |
| 615 | 65 } |
| 633 | 66 |
| 615 | 67 } |
| 68 | |
|
593
bf138339d20a
fiddling with timeouts and closures
Matt Johnston <matt@ucc.asn.au>
parents:
592
diff
changeset
|
69 pub struct ParamWaiter { |
|
609
7bda01659426
not building, paramwaiter work
Matt Johnston <matt@ucc.asn.au>
parents:
594
diff
changeset
|
70 limitlog: NotTooOften, |
| 616 | 71 // last_etag is used for long-polling. |
| 72 last_etag: RefCell<String>, | |
| 73 epoch: String, | |
| 634 | 74 chan: Option<ChannelRef<Params>>, // TODO: a way to avoid Option? |
|
614
e1bab5b36352
using some refcells for the paramwaiter
Matt Johnston <matt@ucc.asn.au>
parents:
613
diff
changeset
|
75 |
|
609
7bda01659426
not building, paramwaiter work
Matt Johnston <matt@ucc.asn.au>
parents:
594
diff
changeset
|
76 config: Config, |
| 592 | 77 } |
| 78 | |
|
611
f3e39e2107fd
still doesn't compile, improvements to TemplogError and tokio curl though
Matt Johnston <matt@ucc.asn.au>
parents:
609
diff
changeset
|
79 const LOG_MINUTES: u64 = 15; |
|
f3e39e2107fd
still doesn't compile, improvements to TemplogError and tokio curl though
Matt Johnston <matt@ucc.asn.au>
parents:
609
diff
changeset
|
80 const MAX_RESPONSE_SIZE: usize = 10000; |
|
614
e1bab5b36352
using some refcells for the paramwaiter
Matt Johnston <matt@ucc.asn.au>
parents:
613
diff
changeset
|
81 const TIMEOUT_MINUTES: u64 = 5; |
|
609
7bda01659426
not building, paramwaiter work
Matt Johnston <matt@ucc.asn.au>
parents:
594
diff
changeset
|
82 |
|
593
bf138339d20a
fiddling with timeouts and closures
Matt Johnston <matt@ucc.asn.au>
parents:
592
diff
changeset
|
83 impl ParamWaiter { |
| 633 | 84 |
| 634 | 85 pub fn new(config: Config) -> Self { |
| 633 | 86 let mut b = [0u8; 15]; // 15 bytes -> 20 characters base64 |
| 87 OsRng.fill_bytes(&mut b); | |
| 88 let epoch = base64::encode(&b); | |
| 89 | |
| 90 ParamWaiter { | |
| 91 limitlog: NotTooOften::new(LOG_MINUTES*60), | |
| 92 last_etag: RefCell::new(String::new()), | |
| 93 epoch: epoch, | |
| 634 | 94 chan: None, |
| 633 | 95 config: config, |
|
614
e1bab5b36352
using some refcells for the paramwaiter
Matt Johnston <matt@ucc.asn.au>
parents:
613
diff
changeset
|
96 } |
|
609
7bda01659426
not building, paramwaiter work
Matt Johnston <matt@ucc.asn.au>
parents:
594
diff
changeset
|
97 } |
|
7bda01659426
not building, paramwaiter work
Matt Johnston <matt@ucc.asn.au>
parents:
594
diff
changeset
|
98 |
| 634 | 99 async fn wait_updates(uri: &str, etag: &str) -> Result<(hyper::Chunk, hyper::StatusCode), TemplogError> { |
| 100 let req = hyper::Request::get(uri) | |
| 101 .header(hyper::header::ETAG, etag)//*self.last_etag.borrow()) | |
| 102 .body(hyper::Body::from("")).unwrap(); | |
| 633 | 103 |
| 634 | 104 // TODO timeout? |
| 105 let resp = hyper::Client::new().request(req).await?; | |
| 106 let status = resp.status(); | |
| 107 let chunk = resp.into_body().try_concat().await?; | |
| 633 | 108 |
| 634 | 109 Ok((chunk, status)) |
| 633 | 110 } |
| 111 | |
| 626 | 112 fn handle_response(&self, buf : hyper::Chunk, status: hyper::StatusCode) -> Result<Params, TemplogError> { |
| 634 | 113 |
| 633 | 114 #[derive(Deserialize, Debug)] |
| 115 struct Response { | |
| 116 // sent as an opaque etag: header. Has format "epoch-nonce", | |
| 117 // responses where the epoch do not match ParamWaiter::epoch are dropped | |
| 118 etag: String, | |
| 119 params: Params, | |
| 120 } | |
| 121 | |
| 626 | 122 match status { |
| 634 | 123 hyper::StatusCode::OK => { |
|
613
5c2b0d47bb83
Response has epoch_tag and params
Matt Johnston <matt@ucc.asn.au>
parents:
611
diff
changeset
|
124 // new params |
| 634 | 125 let r: Response = serde_json::from_slice(&buf)?; |
|
618
2d65a9f0bed3
params returning stream of success
Matt Johnston <matt@ucc.asn.au>
parents:
616
diff
changeset
|
126 let mut le = self.last_etag.borrow_mut(); |
|
2d65a9f0bed3
params returning stream of success
Matt Johnston <matt@ucc.asn.au>
parents:
616
diff
changeset
|
127 *le = r.etag; |
| 626 | 128 |
| 129 // update params if the epoch is correct | |
| 130 if let Some(e) = le.split('-').next() { | |
| 131 if e == &self.epoch { | |
| 132 self.write_params(&r.params); | |
| 133 return Ok(r.params); | |
| 134 } | |
|
618
2d65a9f0bed3
params returning stream of success
Matt Johnston <matt@ucc.asn.au>
parents:
616
diff
changeset
|
135 } |
| 626 | 136 Err(TemplogError::new(&format!("Bad epoch from server '{}' expected '{}'", *le, self.epoch))) |
|
613
5c2b0d47bb83
Response has epoch_tag and params
Matt Johnston <matt@ucc.asn.au>
parents:
611
diff
changeset
|
137 } |
| 634 | 138 hyper::StatusCode::NOT_MODIFIED => { |
| 629 | 139 // XXX this isn't really an error. Should handle_response() return |
| 140 // Result<Option<Params>, TemplogError> instead? | |
|
618
2d65a9f0bed3
params returning stream of success
Matt Johnston <matt@ucc.asn.au>
parents:
616
diff
changeset
|
141 Err(TemplogError::new("304 unmodified (long polling timeout at the server)")) |
|
2d65a9f0bed3
params returning stream of success
Matt Johnston <matt@ucc.asn.au>
parents:
616
diff
changeset
|
142 }, |
|
611
f3e39e2107fd
still doesn't compile, improvements to TemplogError and tokio curl though
Matt Johnston <matt@ucc.asn.au>
parents:
609
diff
changeset
|
143 _ => { |
| 634 | 144 let text = String::from_utf8_lossy(buf.as_ref()); |
| 626 | 145 Err(TemplogError::new(&format!("Wrong server response code {}: {}", status.as_u16(), text))) |
|
611
f3e39e2107fd
still doesn't compile, improvements to TemplogError and tokio curl though
Matt Johnston <matt@ucc.asn.au>
parents:
609
diff
changeset
|
146 }, |
|
f3e39e2107fd
still doesn't compile, improvements to TemplogError and tokio curl though
Matt Johnston <matt@ucc.asn.au>
parents:
609
diff
changeset
|
147 } |
|
f3e39e2107fd
still doesn't compile, improvements to TemplogError and tokio curl though
Matt Johnston <matt@ucc.asn.au>
parents:
609
diff
changeset
|
148 } |
|
f3e39e2107fd
still doesn't compile, improvements to TemplogError and tokio curl though
Matt Johnston <matt@ucc.asn.au>
parents:
609
diff
changeset
|
149 |
|
624
2710649ab71e
read/write params local file. untested
Matt Johnston <matt@ucc.asn.au>
parents:
621
diff
changeset
|
150 fn write_params(&self, params: &Params) { |
|
2710649ab71e
read/write params local file. untested
Matt Johnston <matt@ucc.asn.au>
parents:
621
diff
changeset
|
151 let p = atomicwrites::AtomicFile::new(&self.config.PARAMS_FILE, atomicwrites::AllowOverwrite); |
|
2710649ab71e
read/write params local file. untested
Matt Johnston <matt@ucc.asn.au>
parents:
621
diff
changeset
|
152 p.write(|f| { |
|
2710649ab71e
read/write params local file. untested
Matt Johnston <matt@ucc.asn.au>
parents:
621
diff
changeset
|
153 serde_json::to_writer(f, params) |
|
2710649ab71e
read/write params local file. untested
Matt Johnston <matt@ucc.asn.au>
parents:
621
diff
changeset
|
154 }); |
|
2710649ab71e
read/write params local file. untested
Matt Johnston <matt@ucc.asn.au>
parents:
621
diff
changeset
|
155 } |
| 634 | 156 |
| 157 fn do_poll(&mut self, ctx: &Context<<Self as Actor>::Msg>) -> Result<(), TemplogError> { | |
| 158 let url = self.config.SETTINGS_URL.clone(); | |
| 159 let etag = self.last_etag.borrow().clone(); | |
| 160 let h = ctx.run(async move { | |
| 161 Self::wait_updates(&url, &etag).await | |
| 162 }).expect("spawn failed"); // XXX error handling | |
| 163 let (chunk, stat) = block_on(h)?; | |
| 164 let new_params = self.handle_response(chunk, stat)?; | |
| 165 self.chan.as_ref().unwrap().tell(Publish{msg: new_params, topic: "params".into()}, None); | |
| 166 Ok(()) | |
| 167 } | |
| 633 | 168 } |
|
624
2710649ab71e
read/write params local file. untested
Matt Johnston <matt@ucc.asn.au>
parents:
621
diff
changeset
|
169 |
| 634 | 170 #[derive(Clone,Debug)] |
| 171 pub struct PollForParams; | |
| 172 | |
| 633 | 173 impl Actor for ParamWaiter { |
| 634 | 174 type Msg = PollForParams; |
|
611
f3e39e2107fd
still doesn't compile, improvements to TemplogError and tokio curl though
Matt Johnston <matt@ucc.asn.au>
parents:
609
diff
changeset
|
175 |
| 634 | 176 fn recv(&mut self, |
| 177 ctx: &Context<Self::Msg>, | |
| 178 _msg: Self::Msg, | |
| 179 _sender: Sender) { | |
| 180 // schedule a retry once this iteration finishes | |
| 181 ctx.schedule_once(Duration::from_secs(1), ctx.myself(), None, PollForParams); | |
| 182 | |
| 183 if let Err(e) = self.do_poll(ctx) { | |
| 184 warn!("Problem fetching params: {}", e); | |
| 185 } | |
| 186 } | |
| 187 | |
| 188 fn pre_start(&mut self, ctx: &Context<Self::Msg>) { | |
| 189 self.chan = Some(channel("params", &ctx.system).unwrap()); | |
| 190 ctx.schedule_once(Duration::from_secs(1), ctx.myself(), None, PollForParams); | |
| 592 | 191 } |
| 192 } | |
| 193 | |
| 633 | 194 // pub fn stream(config: Config, handle: Handle) -> Result<(), TemplogError> { |
| 195 // let rcself = Rc::new(ParamWaiter::new(config, handle)); | |
| 196 | |
| 197 // let dur = Duration::from_millis(4000); | |
| 198 // for _ in Interval::new(dur, &rcself.handle).unwrap() { | |
| 199 // // fetch params | |
| 200 // // TODO - skip if inflight. | |
| 201 // let r = await!(rcself.make_request()).map_err(|e| TemplogError::new_hyper("response", e))?; | |
| 202 // let status = r.status(); | |
| 203 // let b = await!(r.body().concat2()).map_err(|e| TemplogError::new_hyper("body", e))?; | |
| 204 // if let Ok(params) = rcself.handle_response(b, status) { | |
| 205 // stream_yield!(params); | |
| 206 // } | |
| 207 // } | |
| 208 // Ok(()) | |
| 209 // } | |
| 210 |