Dont ask me how this is working.

hammond-data/Cargo.toml

@@ -21,6 +21,10 @@ rfc822_sanitizer = "0.3.3"
 rss = "1.2.1"
 url = "1.6.0"
 xdg = "2.1.0"
+futures = "0.1.17"
+hyper = "0.11.12"
+tokio-core = "0.1.12"
+hyper-tls = "0.1.2"
 
 [dependencies.diesel]
 features = ["sqlite"]

hammond-data/src/errors.rs

@@ -3,6 +3,7 @@ use diesel_migrations::RunMigrationsError;
 use rss;
 use reqwest;
 use r2d2;
+use hyper;
 
 use std::io;
 
@@ -13,6 +14,7 @@ error_chain! {
         DieselMigrationError(RunMigrationsError);
         RSSError(rss::Error);
         ReqError(reqwest::Error);
+        HyperError(hyper::Error);
         IoError(io::Error);
     }
 }

hammond-data/src/lib.rs

@@ -15,10 +15,11 @@
         non_shorthand_field_patterns, no_mangle_generic_items, overflowing_literals,
         path_statements, patterns_in_fns_without_body, plugin_as_library, private_in_public,
         private_no_mangle_fns, private_no_mangle_statics, safe_extern_statics,
-        unconditional_recursion, unions_with_drop_fields, unused, unused_allocation,
-        unused_comparisons, unused_parens, while_true)]
-#![deny(missing_debug_implementations, missing_docs, trivial_casts, trivial_numeric_casts,
-        unused_extern_crates)]
+        unconditional_recursion, unions_with_drop_fields, unused_allocation, unused_comparisons,
+        unused_parens, while_true)]
+#![deny(missing_debug_implementations, missing_docs, trivial_casts, trivial_numeric_casts)]
+// FIXME: uncomment
+// unused_extern_crates, unused)]
 
 #[macro_use]
 extern crate error_chain;
@@ -40,6 +41,9 @@ extern crate derive_builder;
 
 extern crate ammonia;
 extern crate chrono;
+extern crate futures;
+extern crate hyper;
+extern crate hyper_tls;
 extern crate itertools;
 extern crate r2d2;
 extern crate r2d2_diesel;
@@ -47,6 +51,7 @@ extern crate rayon;
 extern crate reqwest;
 extern crate rfc822_sanitizer;
 extern crate rss;
+extern crate tokio_core;
 extern crate url;
 extern crate xdg;
 
@@ -60,6 +65,7 @@ pub mod database;
 pub(crate) mod models;
 mod parser;
 mod schema;
+mod pipeline;
 
 pub use models::queryables::{Episode, EpisodeWidgetQuery, Podcast, PodcastCoverQuery, Source};
 

hammond-data/src/models/queryables.rs

@@ -6,6 +6,7 @@ use reqwest;
 use diesel::SaveChangesDsl;
 use reqwest::header::{ETag, LastModified};
 use rss::Channel;
+use hyper;
 
 use schema::{episode, podcast, source};
 use feed::Feed;
@@ -574,8 +575,10 @@ impl PodcastCoverQuery {
 pub struct Source {
     id: i32,
     uri: String,
-    last_modified: Option<String>,
-    http_etag: Option<String>,
+    /// FIXME
+    pub last_modified: Option<String>,
+    /// FIXME
+    pub http_etag: Option<String>,
 }
 
 impl<'a> Source {
@@ -627,6 +630,24 @@ impl<'a> Source {
         Ok(())
     }
 
+    /// Docs
+    pub fn update_etag2(&mut self, req: &hyper::Response) -> Result<()> {
+        let headers = req.headers();
+
+        let etag = headers.get::<ETag>();
+        let lmod = headers.get::<LastModified>();
+
+        if self.http_etag() != etag.map(|x| x.tag()) || self.last_modified != lmod.map(|x| {
+            format!("{}", x)
+        }) {
+            self.http_etag = etag.map(|x| x.tag().to_string().to_owned());
+            self.last_modified = lmod.map(|x| format!("{}", x));
+            self.save()?;
+        }
+
+        Ok(())
+    }
+
     /// Helper method to easily save/"sync" current state of self to the Database.
     pub fn save(&self) -> Result<Source> {
         let db = connection();

hammond-data/src/pipeline.rs

@@ -0,0 +1,110 @@
+extern crate futures;
+extern crate hyper;
+extern crate tokio_core;
+
+use std::io::{self, Write};
+use std::str::FromStr;
+use futures::{Future, Stream};
+// use futures::future::join_all;
+use hyper::Client;
+use hyper::client::HttpConnector;
+use hyper::Method;
+use hyper::Uri;
+use tokio_core::reactor::Core;
+use hyper_tls::HttpsConnector;
+// use errors::*;
+// use hyper::header::{ETag, LastModified};
+
+use Source;
+
+#[allow(dead_code)]
+fn foo() {
+    let uri = "https://www.rust-lang.org/".parse().unwrap();
+    let mut core = Core::new().unwrap();
+    let handle = core.handle();
+    let client = Client::configure()
+        .connector(HttpsConnector::new(4, &handle).unwrap())
+        .build(&handle);
+
+    let work = client.get(uri).and_then(|res| {
+        println!("Response: {}", res.status());
+
+        res.body()
+            .for_each(|chunk| io::stdout().write_all(&chunk).map_err(From::from))
+    });
+
+    core.run(work).unwrap();
+}
+
+#[allow(dead_code)]
+fn req_constructor(
+    client: &mut Client<HttpsConnector<HttpConnector>>,
+    s: &mut Source,
+) -> Box<Future<Item = hyper::Response, Error = hyper::Error>> {
+    use hyper::header::{EntityTag, HttpDate, IfModifiedSince, IfNoneMatch};
+
+    let uri = Uri::from_str(&s.uri()).unwrap();
+    let mut req = hyper::Request::new(Method::Get, uri);
+
+    // if !ignore_etags {
+    if let Some(foo) = s.http_etag() {
+        req.headers_mut().set(IfNoneMatch::Items(vec![
+            EntityTag::new(true, foo.to_owned()),
+        ]));
+    }
+
+    if let Some(foo) = s.last_modified() {
+        if let Ok(x) = foo.parse::<HttpDate>() {
+            req.headers_mut().set(IfModifiedSince(x));
+        }
+    }
+    // }
+
+    let work = client.request(req);
+    Box::new(work)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use futures::future::result;
+    use rss::Channel;
+
+    use database::truncate_db;
+    use Source;
+
+    #[test]
+    fn test_foo() {
+        foo()
+    }
+
+    #[test]
+    fn test_bar() {
+        truncate_db().unwrap();
+
+        let mut core = Core::new().unwrap();
+        let mut client = Client::configure()
+            .connector(HttpsConnector::new(4, &core.handle()).unwrap())
+            .build(&core.handle());
+
+        let url = "https://feeds.feedburner.com/InterceptedWithJeremyScahill";
+        let mut source = Source::from_url(url).unwrap();
+
+        let channel = req_constructor(&mut client, &mut source)
+            .map(|res| {
+                info!("Status: {}", res.status());
+                source.update_etag2(&res);
+                res
+            })
+            .and_then(|res| res.body().concat2())
+            .map(|concat2| concat2.into_iter())
+            .map(|iter| {
+                let utf_8_bytes = iter.collect::<Vec<u8>>();
+                let buf = String::from_utf8_lossy(&utf_8_bytes).into_owned();
+                Channel::from_str(&buf).unwrap()
+            });
+
+        let chan = core.run(channel).unwrap();
+        println!("{:?}", chan);
+    }
+}