--- a/rust/src/types.rs	Thu Feb 23 23:27:09 2017 +0800
+++ b/rust/src/types.rs	Tue Feb 28 22:58:47 2017 +0800
@@ -2,11 +2,17 @@
 use std::time::{Duration,Instant};
 use std::error::Error;
 use std::fmt;
+use std::io;
 use std::cmp;
 use std::cell::Cell;
 use std::iter::FromIterator;
+use std;
 
+use futures::{Stream,IntoFuture};
 use serde::{Deserialize,Serialize};
+use toml;
+use curl;
+use serde_json;
 
 #[derive(Deserialize, Serialize, Debug)]
 pub struct Params {
@@ -77,34 +83,128 @@
 }
 
 #[derive(Debug)]
+pub enum TemplogErrorKind {
+    None,
+    Io(io::Error),
+    ParseFloat(std::num::ParseFloatError),
+    TomlDe(toml::de::Error),
+    SerdeJson(serde_json::Error),
+    Curl(curl::Error),
+}
+
+#[derive(Debug)]
 pub struct TemplogError {
     desc: String,
+    kind: TemplogErrorKind,
 }
 
 impl Error for TemplogError {
-    fn description(&self) -> &str { &self.desc }
-    fn cause(&self) -> Option<&Error> { None }
+    fn description(&self) -> &str { 
+        &format!("{}", self)
+    }
+
+    fn cause(&self) -> Option<&Error> { 
+        match self.kind {
+            TemplogErrorKind::None => None,
+            TemplogErrorKind::Io(e) => Some(&e),
+            TemplogErrorKind::ParseFloat(e) => Some(&e),
+            TemplogErrorKind::TomlDe(e) => Some(&e),
+            TemplogErrorKind::SerdeJson(e) => Some(&e),
+            TemplogErrorKind::Curl(e) => Some(&e),
+        }
+    }
 }
 
 impl fmt::Display for TemplogError {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        write!(f, "TemplogError: {}", self.desc);
+        match self.kind {
+            TemplogErrorKind::None => write!(f, "Templog Error: {}", self.desc),
+            TemplogErrorKind::Io(e) => write!(f, "Templog IO error {}: {}", self.desc, e),
+            TemplogErrorKind::TomlDe(e) => write!(f, "Templog toml error {}: {}", self.desc, e),
+            TemplogErrorKind::SerdeJson(e) => write!(f, "Json decode error {}: {}", self.desc, e),
+            TemplogErrorKind::ParseFloat(e) => write!(f, "Templog parse error {}: {}", self.desc, e),
+            TemplogErrorKind::Curl(e) => write!(f, "Templog curl http error {}: {}", self.desc, e),
+        };
         Ok(())
     }
-
-
 }
 
 impl TemplogError {
     pub fn new(desc: &str) -> Self {
         TemplogError { 
             desc: desc.to_string(),
+            kind: TemplogErrorKind::None,
+        }
+    }
+
+    pub fn new_io(desc: &str, e: io::Error) -> Self {
+        TemplogError { 
+            desc: desc.to_string(),
+            kind: TemplogErrorKind::Io(e),
+        }
+    }
+
+    pub fn new_toml_de(desc: &str, e: toml::de::Error) -> Self {
+        TemplogError { 
+            desc: desc.to_string(),
+            kind: TemplogErrorKind::TomlDe(e),
+        }
+    }
+
+    pub fn new_parse_float(desc: &str, e: std::num::ParseFloatError) -> Self {
+        TemplogError { 
+            desc: desc.to_string(),
+            kind: TemplogErrorKind::ParseFloat(e),
+        }
+    }
+
+    pub fn new_curl(desc: &str, e: curl::Error) -> Self {
+        TemplogError { 
+            desc: desc.to_string(),
+            kind: TemplogErrorKind::Curl(e),
+        }
+    }
+
+    pub fn new_serde_json(desc: &str, e: serde_json::Error) -> Self {
+        TemplogError { 
+            desc: desc.to_string(),
+            kind: TemplogErrorKind::SerdeJson(e),
         }
     }
 }
 
+impl From<io::Error> for TemplogError {
+    fn from(e: io::Error) -> Self {
+        TemplogError::new_io("", e)
+    }
+}
+
+impl From<toml::de::Error> for TemplogError {
+    fn from(e: toml::de::Error) -> Self {
+        TemplogError::new_toml_de("", e)
+    }
+}
+
+impl From<std::num::ParseFloatError> for TemplogError {
+    fn from(e: std::num::ParseFloatError) -> Self {
+        TemplogError::new_parse_float("", e)
+    }
+}
+
+impl From<curl::Error> for TemplogError {
+    fn from(e: curl::Error) -> Self {
+        TemplogError::new_curl("", e)
+    }
+}
+
+impl From<serde_json::Error> for TemplogError {
+    fn from(e: serde_json::Error) -> Self {
+        TemplogError::new_serde_json("", e)
+    }
+}
 
 /// Call closures with a rate limit. Useful for log message ratelimiting
+#[derive(Clone)]
 pub struct NotTooOften {
     last: Cell<Instant>,
     limit: Duration,
@@ -200,3 +300,22 @@
     }
 }
 
+/// Takes a stream and returns a stream without errors.
+pub fn consume_errors<S>(s: S) -> Box<Stream<Item=S::Item, Error=S::Error>> 
+// XXX not sure why 'static is really necessary here?
+    where 
+    S: Stream+Send+'static, 
+    <S as Stream>::Error: std::fmt::Display+Send+'static,
+    <S as Stream>::Item: Send+'static,
+    {
+    s.then(|r| {
+        match r {
+            Ok(v) => Ok(Some(v)),
+            Err(e) => {
+                debug!("Stream error: {}", e);
+                Ok(None)
+            }
+        }
+    })
+    .filter_map(|p| p).boxed()
+}