diff --git a/src/main.rs b/src/main.rs index c888a90..2f45d09 100644 --- a/src/main.rs +++ b/src/main.rs @@ -73,21 +73,76 @@ impl Resources { queue.pop_front() } + fn worker_fetch(&self, cal_id: &str, cal_opts: &CalendarOptions, etag: Option, last_modified: Option) + -> Result<(Option, Option, HashMap), Error> + { + let mut req = self.http_client.get(&cal_opts.url) + .header(USER_AGENT, "Ticker/0.0.0"); + match etag { + Some(etag) => req = req.header(IF_NONE_MATCH, etag), + None => (), + } + match last_modified { + Some(last_modified) => req = req.header(IF_MODIFIED_SINCE, last_modified), + None => (), + } + let mut res = req.send()?; + + if res.status() != 200 { + let msg = format!("HTTP {}", res.status()); + return Err(Error::Message(msg)); + } + + let etag = res.headers().get(ETAG) + .and_then(|v| v.to_str().ok()) + .map(|s| s.to_owned()); + let last_modified = res.headers().get(LAST_MODIFIED) + .and_then(|v| v.to_str().ok()) + .map(|s| s.to_owned()); + + let mut p = Parser::new(); + let mut buf = [0; 1024]; + let mut events = HashMap::new(); + loop { + match res.read(&mut buf)? { + len if len > 0 => { + let data = &buf[..len]; + p.feed(data, |obj| { + let mut objs = vec![]; + extract_vevent_objs(&mut objs, obj); + for obj in objs { + if let Some(event) = obj_to_event(cal_id.to_owned(), &obj) { + events.insert(event.id.clone(), event); + } else { + let dtstart: Option<&str> = obj.get("DTSTART"); + let summary: Option<&str> = obj.get("SUMMARY"); + println!("cannot convert {} {:?} {:?}", + obj.name, dtstart, summary); + } + } + }); + } + _ => break, + } + } + Ok((etag, last_modified, events)) + } + fn worker_job(&self, db: &PgConnection) -> Result { - let (id, cal_opts) = match self.next() { + let (cal_id, cal_opts) = match self.next() { Some(next) => next, None => return Ok(true), }; let (etag, last_modified) = db.transaction::<_, Error, _>(|| { let cals = calendars - .filter(schema::calendars::dsl::id.eq(id.clone())) + .filter(schema::calendars::dsl::id.eq(cal_id.clone())) .limit(1) .load::(db)?; match cals.first() { None => { let cal = NewCalendar { - id: id.clone(), + id: cal_id.clone(), url: cal_opts.url.clone(), last_fetch: Some(Utc::now().naive_utc()), }; @@ -98,7 +153,7 @@ impl Resources { } Some(cal) => { diesel::update(calendars) - .filter(schema::calendars::dsl::id.eq(id.clone())) + .filter(schema::calendars::dsl::id.eq(cal_id.clone())) .set(schema::calendars::dsl::last_fetch.eq(Utc::now().naive_utc())) .execute(db)?; let result = @@ -113,74 +168,31 @@ impl Resources { } })?; - let mut req = self.http_client.get(&cal_opts.url) - .header(USER_AGENT, "Ticker/0.0.0"); - match etag { - Some(etag) => req = req.header(IF_NONE_MATCH, etag), - None => (), - } - match last_modified { - Some(last_modified) => req = req.header(IF_MODIFIED_SINCE, last_modified), - None => (), - } - let mut res = req.send()?; + let (etag, last_modified, events) = + self.worker_fetch(&cal_id, &cal_opts, etag, last_modified) + .map_err(|e| { + let msg = format!("{:?}", e); + println!("[{}] {}", cal_id, msg); + let _ = diesel::update(calendars) + .filter(schema::calendars::dsl::id.eq(cal_id.clone())) + .set(schema::calendars::dsl::error_message.eq(msg)) + .execute(db); + e + })?; - println!("[{}] HTTP {}", id, res.status()); - if res.status() != 200 { - let msg = format!("HTTP {}", res.status()); - diesel::update(calendars) - .filter(schema::calendars::dsl::id.eq(id.clone())) - .set((schema::calendars::dsl::last_success.eq(Utc::now().naive_utc()), - schema::calendars::dsl::error_message.eq(msg))) - .execute(db)?; - return Ok(false) - } - - let etag = res.headers().get(ETAG) - .and_then(|v| v.to_str().ok()) - .map(|s| s.to_owned()); - let last_modified = res.headers().get(LAST_MODIFIED) - .and_then(|v| v.to_str().ok()) - .map(|s| s.to_owned()); + println!("[{}] {} events", cal_id, events.len()); db.transaction::<_, Error, _>(|| { diesel::update(calendars) - .filter(schema::calendars::dsl::id.eq(id.clone())) + .filter(schema::calendars::dsl::id.eq(cal_id.clone())) .set((schema::calendars::dsl::last_success.eq(Utc::now().naive_utc()), schema::calendars::dsl::etag.eq(etag), - schema::calendars::dsl::last_modified.eq(last_modified))) + schema::calendars::dsl::last_modified.eq(last_modified), + schema::calendars::dsl::error_message.eq(None as Option))) .execute(db)?; - let mut p = Parser::new(); - let mut buf = [0; 1024]; - let mut events = HashMap::new(); - loop { - match res.read(&mut buf)? { - len if len > 0 => { - let data = &buf[..len]; - p.feed(data, |obj| { - let mut objs = vec![]; - extract_vevent_objs(&mut objs, obj); - for obj in objs { - if let Some(event) = obj_to_event(id.clone(), &obj) { - events.insert(event.id.clone(), event); - } else { - let dtstart: Option<&str> = obj.get("DTSTART"); - let summary: Option<&str> = obj.get("SUMMARY"); - println!("cannot convert {} {:?} {:?}", - obj.name, dtstart, summary); - } - } - }); - } - _ => break, - } - } - drop(res); - - println!("[{}] {} events", id, events.len()); diesel::delete(schema::events::dsl::events) - .filter(schema::events::dsl::calendar.eq(id)) + .filter(schema::events::dsl::calendar.eq(cal_id)) .execute(db)?; for event in events.values() { diesel::insert_into(schema::events::dsl::events) @@ -188,7 +200,7 @@ impl Resources { .execute(db)?; } - Ok(false) + Ok(()) })?; Ok(false) @@ -212,6 +224,7 @@ pub enum Error { Db(diesel::result::Error), Http(reqwest::Error), Io(std::io::Error), + Message(String), } impl From for Error {