Re-architecting the application to use event sourcing (fixes: #35)

src/api/mod.rs

@@ -1,14 +1,143 @@
 //! Data structures for the API, shared between client and server.
-pub mod publishers;
-pub mod publication;
-pub mod rrdp_data;
+pub mod publisher_data;
+pub mod publication_data;
+pub mod repo_data;
+
+use bytes::Bytes;
+use serde::{Deserialize, Deserializer, Serialize, Serializer};
+use crate::util::sha256;
+
+
+//------------ Base64 --------------------------------------------------------
+
+/// This type contains a base64 encoded structure. The publication protocol
+/// deals with objects in their base64 encoded form.
+///
+/// Note that we store this in a Bytes to make it cheap to clone this.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct Base64(Bytes);
+
+impl Base64 {
+    pub fn from_content(content: &[u8]) -> Self {
+        Base64::from(base64::encode(content))
+    }
+
+    /// Decodes into bytes (e.g. for saving to disk for rcync)
+    pub fn to_bytes(&self) -> Bytes {
+        Bytes::from(base64::decode(&self.0).unwrap())
+    }
+
+    pub fn to_hex_hash(&self) -> String {
+        hex::encode(sha256(&self.to_bytes()))
+    }
+
+    pub fn to_encoded_hash(&self) -> EncodedHash {
+        EncodedHash::from(self.to_hex_hash())
+    }
+}
+
+impl AsRef<str> for Base64 {
+    fn as_ref(&self) -> &str {
+        use std::str;
+        str::from_utf8(&self.0).unwrap()
+    }
+}
+
+impl From<String> for Base64 {
+    fn from(s: String) -> Self {
+        Base64(Bytes::from(s))
+    }
+}
+
+
+impl ToString for Base64 {
+    fn to_string(&self) -> String {
+        unsafe {
+            String::from_utf8_unchecked(self.0.to_vec())
+        }
+    }
+}
+
+impl Serialize for Base64 {
+    fn serialize<S>(
+        &self, serializer: S
+    ) -> Result<S::Ok, S::Error> where S: Serializer {
+        self.to_string().serialize(serializer)
+    }
+}
+
+impl<'de> Deserialize<'de> for Base64 {
+    fn deserialize<D>(
+        deserializer: D
+    ) -> Result<Base64, D::Error> where D: Deserializer<'de> {
+        let string = String::deserialize(deserializer)?;
+        Ok(Base64::from(string))
+    }
+}
+
+
+//------------ EncodedHash ---------------------------------------------------
+
+/// This type contains a hex encoded sha256 hash.
+///
+/// Note that we store this in a Bytes for cheap cloning.
+#[derive(Clone, Debug, Eq, Hash, PartialEq)]
+pub struct EncodedHash(Bytes);
+
+impl EncodedHash {
+    pub fn from_content(content: &[u8]) -> Self {
+        let sha256 = sha256(content);
+        let hex = hex::encode(sha256);
+        EncodedHash::from(hex)
+    }
+}
+
+impl AsRef<str> for EncodedHash {
+    fn as_ref(&self) -> &str {
+        use std::str;
+        str::from_utf8(&self.0).unwrap()
+    }
+}
+
+impl From<String> for EncodedHash {
+    fn from(s: String) -> Self {
+        EncodedHash(Bytes::from(s.to_lowercase()))
+    }
+}
+
+impl ToString for EncodedHash {
+    fn to_string(&self) -> String {
+        unsafe {
+            String::from_utf8_unchecked(self.0.to_vec())
+        }
+    }
+}
+
+impl Serialize for EncodedHash {
+    fn serialize<S>(
+        &self, serializer: S
+    ) -> Result<S::Ok, S::Error> where S: Serializer {
+        self.to_string().serialize(serializer)
+    }
+}
+
+impl<'de> Deserialize<'de> for EncodedHash {
+    fn deserialize<D>(
+        deserializer: D
+    ) -> Result<EncodedHash, D::Error> where D: Deserializer<'de> {
+        let string = String::deserialize(deserializer)?;
+        Ok(EncodedHash::from(string))
+    }
+}
+
+
 
 //------------ Link ----------------------------------------------------------
 
 /// Defines a link element to include as part of a links array in a Json
 /// response.
 #[derive(Clone, Debug, Serialize)]
-pub struct Link<'a> {
-    rel: &'a str,
+pub struct Link {
+    rel: String,
     link: String
 }

src/api/publication_data.rs

@@ -1,9 +1,8 @@
 //! Support for requests sent to the Json API
-use bytes::Bytes;
 use rpki::uri;
+use crate::api::{ Base64, EncodedHash };
 use crate::util::ext_serde;
 use crate::util::file::CurrentFile;
-use crate::util::sha256;
 
 
 //------------ PublishRequest ------------------------------------------------
@@ -51,6 +50,10 @@ impl PublishDelta {
     }
 
     pub fn is_empty(&self) -> bool { self.len() == 0 }
+
+    pub fn unwrap(self) -> (Vec<Publish>, Vec<Update>, Vec<Withdraw>) {
+        (self.publishes, self.updates, self.withdraws)
+    }
 }
 
 
@@ -104,18 +107,15 @@ pub struct Publish {
         serialize_with = "ext_serde::ser_rsync_uri")]
     uri: uri::Rsync,
 
-    #[serde(
-        deserialize_with = "ext_serde::de_bytes",
-        serialize_with = "ext_serde::ser_bytes")]
-    content: Bytes
+    content: Base64
 }
 
 impl Publish {
-    pub fn new(tag: Option<String>, uri: uri::Rsync, content: Bytes) -> Self {
+    pub fn new(tag: Option<String>, uri: uri::Rsync, content: Base64) -> Self {
         Publish { tag, uri, content }
     }
-    pub fn with_hash_tag(uri: uri::Rsync, content: Bytes) -> Self {
-        let tag = Some(hex::encode(sha256(&content)));
+    pub fn with_hash_tag(uri: uri::Rsync, content: Base64) -> Self {
+        let tag = Some(content.to_hex_hash());
         Publish { tag, uri, content }
     }
 
@@ -127,7 +127,11 @@ impl Publish {
         }
     }
     pub fn uri(&self) -> &uri::Rsync{ &self.uri}
-    pub fn content(&self) -> &Bytes{ &self.content }
+    pub fn content(&self) -> &Base64{ &self.content }
+
+    pub fn unwrap(self) -> (Option<String>, uri::Rsync, Base64) {
+        (self.tag, self.uri, self.content)
+    }
 }
 
 
@@ -145,32 +149,26 @@ pub struct Update {
         serialize_with = "ext_serde::ser_rsync_uri")]
     uri: uri::Rsync,
 
-    #[serde(
-        deserialize_with = "ext_serde::de_bytes",
-        serialize_with = "ext_serde::ser_bytes")]
-    content: Bytes,
+    content: Base64,
 
-    #[serde(
-        deserialize_with = "ext_serde::de_bytes",
-        serialize_with = "ext_serde::ser_bytes")]
-    hash: Bytes,
+    hash: EncodedHash,
 }
 
 impl Update {
     pub fn new(
         tag: Option<String>,
         uri: uri::Rsync,
-        content: Bytes,
-        old_hash: Bytes
+        content: Base64,
+        old_hash: EncodedHash
     ) -> Self {
         Update { tag, uri, content, hash: old_hash }
     }
     pub fn with_hash_tag(
         uri: uri::Rsync,
-        content: Bytes,
-        old_hash: Bytes
+        content: Base64,
+        old_hash: EncodedHash
     ) -> Self {
-        let tag = Some(hex::encode(sha256(&content)));
+        let tag = Some(content.to_hex_hash());
         Update { tag, uri, content, hash: old_hash }
     }
 
@@ -182,8 +180,12 @@ impl Update {
         }
     }
     pub fn uri(&self) -> &uri::Rsync { &self.uri}
-    pub fn content(&self) -> &Bytes { &self.content }
-    pub fn hash(&self) -> &Bytes { &self.hash }
+    pub fn content(&self) -> &Base64 { &self.content }
+    pub fn hash(&self) -> &EncodedHash { &self.hash }
+
+    pub fn unwrap(self) -> (Option<String>, uri::Rsync, Base64, EncodedHash) {
+        (self.tag, self.uri, self.content, self.hash)
+    }
 }
 
 
@@ -201,19 +203,16 @@ pub struct Withdraw {
         serialize_with = "ext_serde::ser_rsync_uri")]
     uri: uri::Rsync,
 
-    #[serde(
-        deserialize_with = "ext_serde::de_bytes",
-        serialize_with = "ext_serde::ser_bytes")]
-    hash: Bytes,
+    hash: EncodedHash,
 }
 
 impl Withdraw {
-    pub fn new(tag: Option<String>, uri: uri::Rsync, hash: Bytes) -> Self {
+    pub fn new(tag: Option<String>, uri: uri::Rsync, hash: EncodedHash) -> Self {
         Withdraw { tag, uri, hash }
     }
 
-    pub fn with_hash_tag(uri: uri::Rsync, hash: Bytes) -> Self {
-        let tag = Some(hex::encode(&hash));
+    pub fn with_hash_tag(uri: uri::Rsync, hash: EncodedHash) -> Self {
+        let tag = Some(hash.to_string());
         Withdraw { tag, uri, hash }
     }
 
@@ -233,7 +232,11 @@ impl Withdraw {
         }
     }
     pub fn uri(&self) -> &uri::Rsync { &self.uri}
-    pub fn hash(&self) -> &Bytes { &self.hash }
+    pub fn hash(&self) -> &EncodedHash { &self.hash }
+
+    pub fn unwrap(self) -> (Option<String>, uri::Rsync, EncodedHash) {
+        (self.tag, self.uri, self.hash)
+    }
 }
 
 //------------ PublishReply --------------------------------------------------
@@ -281,21 +284,14 @@ pub struct ListElement {
     serialize_with = "ext_serde::ser_rsync_uri")]
     uri:     uri::Rsync,
 
-    #[serde(
-    deserialize_with = "ext_serde::de_bytes",
-    serialize_with = "ext_serde::ser_bytes")]
-    /// The sha-256 hash of the file (as is used on the RPKI manifests and
-    /// in the publication protocol for list, update and withdraw). Saving
-    /// this rather than calculating on demand seems a small price for some
-    /// performance gain.
-    hash:    Bytes
+    hash:    EncodedHash
 }
 
 impl ListElement {
-    pub fn new(uri: uri::Rsync, hash: Bytes) -> Self {
+    pub fn new(uri: uri::Rsync, hash: EncodedHash) -> Self {
         ListElement { uri, hash }
     }
 
     pub fn uri(&self) -> &uri::Rsync { &self.uri }
-    pub fn hash(&self) -> &Bytes { &self.hash}
+    pub fn hash(&self) -> &EncodedHash { &self.hash }
 }

src/api/publisher_data.rs

@@ -2,13 +2,21 @@
 //!
 //! i.e. this is stuff the the server needs to serialize only, so typically
 //! we can work with references here.
-use std::sync::Arc;
 use rpki::uri;
 use crate::api::Link;
+use crate::eventsourcing::AggregateId;
+use crate::krilld::pubd::publishers::Publisher;
 use crate::remote::id::IdCert;
 use crate::util::ext_serde;
 
 
+//------------ PublisherHandle -----------------------------------------------
+
+/// A type for referring to publishers, both in the api as well as to the
+/// aggregates.
+pub type PublisherHandle = AggregateId;
+
+
 //------------ CmsAuthData ---------------------------------------------------
 
 /// This type contains the data needed for handling RFC8183 requests/responses,
@@ -51,11 +59,12 @@ impl PartialEq for CmsAuthData {
 impl Eq for CmsAuthData {}
 
 
-//------------ Publisher -----------------------------------------------------
+//------------ PublisherRequest ----------------------------------------------
 
-/// This type defines Publisher CAs that are allowed to publish.
+/// This type defines request for a new Publisher (CA that is allowed to
+/// publish).
 #[derive(Clone, Debug, Deserialize, Serialize)]
-pub struct Publisher {
+pub struct PublisherRequest {
     handle:        String,
 
     /// The token used by the API
@@ -69,14 +78,14 @@ pub struct Publisher {
     cms_auth_data: Option<CmsAuthData>
 }
 
-impl Publisher {
+impl PublisherRequest {
     pub fn new(
         handle:   String,
         token:    String,
         base_uri: uri::Rsync,
         rfc8181:  Option<CmsAuthData>
     ) -> Self {
-        Publisher {
+        PublisherRequest {
             handle,
             token,
             base_uri,
@@ -85,7 +94,7 @@ impl Publisher {
     }
 }
 
-impl Publisher {
+impl PublisherRequest {
     pub fn handle(&self) -> &String {
         &self.handle
     }
@@ -101,17 +110,22 @@ impl Publisher {
     pub fn cms_auth_data(&self) -> &Option<CmsAuthData> {
         &self.cms_auth_data
     }
+
+    /// Return all the values (handle, token, base_uri, rfc8181opt).
+    pub fn unwrap(self) -> (String, String, uri::Rsync, Option<CmsAuthData>) {
+        (self.handle, self.token, self.base_uri, self.cms_auth_data)
+    }
 }
 
-impl PartialEq for Publisher {
-    fn eq(&self, other: &Publisher) -> bool {
+impl PartialEq for PublisherRequest {
+    fn eq(&self, other: &PublisherRequest) -> bool {
         self.handle == other.handle &&
             self.base_uri == other.base_uri &&
             self.cms_auth_data == other.cms_auth_data
     }
 }
 
-impl Eq for Publisher {}
+impl Eq for PublisherRequest {}
 
 
 //------------ PublisherSummaryInfo ------------------------------------------
@@ -119,33 +133,25 @@ impl Eq for Publisher {}
 /// Defines a summary of publisher information to be used in the publisher
 /// list.
 #[derive(Clone, Debug, Serialize)]
-pub struct PublisherSummaryInfo<'a> {
-    id: &'a str,
-    links: Vec<Link<'a>>
+pub struct PublisherSummaryInfo {
+    id: String,
+    links: Vec<Link>
 }
 
-impl<'a> PublisherSummaryInfo<'a> {
+impl PublisherSummaryInfo {
     pub fn from(
-        publisher: &'a Publisher,
-        path_publishers: &'a str
-    ) -> PublisherSummaryInfo<'a> {
-        let id = publisher.handle().as_str();
+        handle: &PublisherHandle,
+        path_publishers: &str
+    ) -> PublisherSummaryInfo  {
         let mut links = Vec::new();
-
-        let response_link = Link {
-            rel: "response.xml",
-            link: format!("{}/{}/response.xml", path_publishers, id)
-        };
         let self_link = Link {
-            rel: "self",
-            link: format!("{}/{}", path_publishers, id)
+            rel: "self".to_string(),
+            link: format!("{}/{}", path_publishers, handle.as_ref())
         };
-
-        links.push(response_link);
         links.push(self_link);
 
         PublisherSummaryInfo {
-            id,
+            id: handle.to_string(),
             links
         }
     }
@@ -156,15 +162,15 @@ impl<'a> PublisherSummaryInfo<'a> {
 
 /// This type represents a list of (all) current publishers to show in the API
 #[derive(Clone, Debug, Serialize)]
-pub struct PublisherList<'a> {
-    publishers: Vec<PublisherSummaryInfo<'a>>
+pub struct PublisherList {
+    publishers: Vec<PublisherSummaryInfo>
 }
 
-impl<'a> PublisherList<'a> {
-    pub fn from(
-        publishers: &'a[Arc<Publisher>],
-        path_publishers: &'a str
-    ) -> PublisherList<'a> {
+impl PublisherList {
+    pub fn build(
+        publishers: &[PublisherHandle],
+        path_publishers: &str
+    ) -> PublisherList {
         let publishers: Vec<PublisherSummaryInfo> = publishers.iter().map(|p|
             PublisherSummaryInfo::from(&p, path_publishers)
         ).collect();
@@ -194,24 +200,27 @@ pub struct Rfc8181Details<'a> {
 
 #[derive(Clone, Debug, Serialize)]
 pub struct PublisherDetails<'a> {
-    publisher_handle: &'a str,
+    handle: &'a str,
+
+    retired: bool,
 
     #[serde(serialize_with = "ext_serde::ser_rsync_uri")]
     base_uri: &'a uri::Rsync,
 
     rfc8181: Option<Rfc8181Details<'a>>,
 
-    links: Vec<Link<'a>>
+    links: Vec<Link>
 }
 
 impl<'a> PublisherDetails<'a> {
     pub fn from(
-        publisher: &'a Arc<Publisher>,
+        publisher: &'a Publisher,
         path_publishers: &'a str,
         base_service_uri: &uri::Http
     ) -> PublisherDetails<'a> {
-        let handle = publisher.handle().as_str();
+        let handle = publisher.id().as_ref();
         let base_uri = publisher.base_uri();
+        let retired = publisher.retired();
 
         // Derive the RFC8181 service URI.
         let service_uri = format!("{}{}", base_service_uri, handle);
@@ -229,12 +238,13 @@ impl<'a> PublisherDetails<'a> {
 
         let mut links = Vec::new();
         links.push(Link {
-            rel: "response.xml",
+            rel: "response.xml".to_string(),
             link: format!("{}/{}/response.xml", path_publishers, handle)
         });
 
         PublisherDetails {
-            publisher_handle: handle,
+            handle,
+            retired,
             base_uri,
             rfc8181,
             links

src/api/repo_data.rs

@@ -0,0 +1,671 @@
+//! Data objects used in the (RRDP) repository. I.e. the publish, update, and
+//! withdraw elements, as well as the notification, snapshot and delta file
+//! definitions.
+use std::collections::HashMap;
+use std::io;
+use std::path::PathBuf;
+use bytes::Bytes;
+use rpki::uri;
+use crate::api::publication_data;
+use crate::api::Base64;
+use crate::api::EncodedHash;
+use crate::util::ext_serde;
+use crate::util::file;
+use crate::util::Time;
+use crate::util::xml::XmlWriter;
+
+
+const VERSION: &str = "1";
+const NS: &str = "http://www.ripe.net/rpki/rrdp";
+
+//------------ PublishElement ------------------------------------------------
+
+/// The publishes as used in the RRDP protocol.
+///
+/// Note that the difference with the publication protocol is the absence of
+/// the tag.
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct PublishElement {
+    base64: Base64,
+
+    #[serde(
+    deserialize_with = "ext_serde::de_rsync_uri",
+    serialize_with = "ext_serde::ser_rsync_uri")]
+    uri: uri::Rsync
+}
+
+impl PublishElement {
+    pub fn new(base64: Base64, uri: uri::Rsync) -> Self {
+        PublishElement { base64, uri }
+    }
+
+    pub fn base64(&self) -> &Base64 { &self.base64 }
+    pub fn uri(&self) -> &uri::Rsync { &self.uri }
+}
+
+impl From<publication_data::Publish> for PublishElement {
+    fn from(p: publication_data::Publish) -> Self {
+        let (_tag, uri, base64) = p.unwrap();
+        PublishElement { uri, base64 }
+    }
+}
+
+
+//------------ UpdateElement -------------------------------------------------
+
+/// The updates as used in the RRDP protocol.
+///
+/// Note that the difference with the publication protocol is the absence of
+/// the tag.
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct UpdateElement {
+    #[serde(
+    deserialize_with = "ext_serde::de_rsync_uri",
+    serialize_with = "ext_serde::ser_rsync_uri")]
+    uri: uri::Rsync,
+    hash: EncodedHash,
+    base64: Base64
+}
+
+impl UpdateElement {
+    pub fn uri(&self) -> &uri::Rsync { &self.uri }
+    pub fn hash(&self) -> &EncodedHash { &self.hash }
+    pub fn base64(&self) -> &Base64 { &self.base64 }
+}
+
+impl From<publication_data::Update> for UpdateElement {
+    fn from(u: publication_data::Update) -> Self {
+        let (_tag, uri, base64, hash) = u.unwrap();
+        UpdateElement { uri, base64, hash }
+    }
+}
+
+impl Into<PublishElement> for UpdateElement {
+    fn into(self) -> PublishElement {
+        PublishElement { uri: self.uri, base64: self.base64 }
+    }
+}
+
+
+//------------ WithdrawElement -----------------------------------------------
+
+/// The withdraws as used in the RRDP protocol.
+///
+/// Note that the difference with the publication protocol is the absence of
+/// the tag.
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct WithdrawElement {
+    #[serde(
+    deserialize_with = "ext_serde::de_rsync_uri",
+    serialize_with = "ext_serde::ser_rsync_uri")]
+    uri: uri::Rsync,
+    hash: EncodedHash
+}
+
+impl WithdrawElement {
+    pub fn uri(&self) -> &uri::Rsync { &self.uri }
+    pub fn hash(&self) -> &EncodedHash { &self.hash }
+}
+
+impl From<publication_data::Withdraw> for WithdrawElement {
+    fn from(w: publication_data::Withdraw) -> Self {
+        let (_tag, uri, hash) = w.unwrap();
+        WithdrawElement { uri, hash }
+    }
+}
+
+
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct Notification {
+    session:    String,
+    serial:     u64,
+    time:       Time,
+    snapshot:   SnapshotRef,
+    deltas:     Vec<DeltaRef>,
+    old_refs:   Vec<(Time, FileRef)>
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct NotificationUpdate {
+    time:       Time,
+    session:    Option<String>,
+    snapshot:   SnapshotRef,
+    delta:      DeltaRef,
+    last_delta: u64
+}
+
+impl NotificationUpdate {
+    pub fn new(
+        time: Time,
+        session: Option<String>,
+        snapshot: SnapshotRef,
+        delta: DeltaRef,
+        last_delta: u64
+    ) -> Self {
+        NotificationUpdate { time, session, snapshot, delta, last_delta }
+    }
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct NotificationCreate {
+    session:   String,
+    snapshot:  SnapshotRef
+}
+
+impl NotificationUpdate {
+    pub fn unwrap(self) -> (Time, Option<String>, SnapshotRef, DeltaRef, u64) {
+        (self.time, self.session, self.snapshot, self.delta, self.last_delta)
+    }
+}
+
+impl Notification {
+    pub fn old_refs(&self) -> &Vec<(Time, FileRef)> {
+        &self.old_refs
+    }
+
+    pub fn update(&mut self, update: NotificationUpdate) {
+        let (time, session_opt, snapshot, delta, last_delta) = update.unwrap();
+        if let Some(session) = session_opt {
+            self.session = session;
+        }
+
+        self.serial += 1;
+        self.time = time;
+
+        let mut refs_to_retire = vec![];
+
+        refs_to_retire.push((Time::now(), self.snapshot.clone()));
+        self.snapshot = snapshot;
+
+        for d in &self.deltas {
+            if d.serial < last_delta {
+                refs_to_retire.push((Time::now(), d.file_ref.clone()));
+            }
+        }
+
+        self.deltas.insert(0, delta);
+        self.deltas.retain(|delta| delta.serial >= last_delta);
+        self.old_refs.append(&mut refs_to_retire);
+    }
+
+    /// Cleans up all old references from before the given time.
+    pub fn clean_up(&mut self, t: Time) {
+        self.old_refs.retain(|old_ref| {! old_ref.0.on_or_before(&t)})
+    }
+
+    pub fn create(session: String, snapshot: SnapshotRef) -> Self {
+        Notification {
+            session,
+            serial: 0,
+            time: Time::now(),
+            snapshot,
+            deltas: vec![],
+            old_refs: vec![]
+        }
+    }
+
+    pub fn write_xml(&self, path: &PathBuf) -> Result<(), io::Error> {
+        debug!("Writing notification file: {}", path.to_string_lossy());
+        let mut file = file::create_file_with_path(&path)?;
+
+        XmlWriter::encode_to_file(& mut file, |w| {
+
+            let a = [
+                ("xmlns", NS),
+                ("version", VERSION),
+                ("session_id", self.session.as_ref()),
+                ("serial", &format!("{}", self.serial)),
+            ];
+
+            w.put_element(
+                "notification",
+                Some(&a),
+                |w| {
+                    {
+                        // snapshot ref
+                        let uri = self.snapshot.uri.to_string();
+                        let a = [
+                            ("uri", uri.as_str()),
+                            ("hash", self.snapshot.hash.as_ref())
+                        ];
+                        w.put_element(
+                            "snapshot",
+                            Some(&a),
+                            |w| { w.empty() }
+                        )?;
+                    }
+
+                    {
+                        // delta refs
+                        for delta in &self.deltas {
+                            let serial = format!("{}", delta.serial);
+                            let uri = delta.file_ref.uri.to_string();
+                            let a = [
+                                ("serial", serial.as_ref()),
+                                ("uri", uri.as_str()),
+                                ("hash", delta.file_ref.hash.as_ref())
+                            ];
+                            w.put_element(
+                                "delta",
+                                Some(&a),
+                                |w| { w.empty() }
+                            )?;
+                        }
+                    }
+
+                    Ok(())
+                }
+            )
+        })?;
+
+        Ok(())
+
+    }
+
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct FileRef {
+    #[serde(
+    deserialize_with = "ext_serde::de_http_uri",
+    serialize_with = "ext_serde::ser_http_uri")]
+    uri:   uri::Http,
+    path:  PathBuf,
+    hash:  EncodedHash,
+}
+
+impl FileRef {
+    pub fn new(uri: uri::Http, path: PathBuf, hash: EncodedHash) -> Self {
+        FileRef { uri, path, hash }
+    }
+    pub fn uri(&self) -> &uri::Http { &self.uri }
+    pub fn path(&self) -> &PathBuf { &self.path }
+    pub fn hash(&self) -> &EncodedHash { &self.hash }
+}
+
+pub type SnapshotRef = FileRef;
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct DeltaRef {
+    serial:     u64,
+    file_ref:   FileRef
+}
+
+impl DeltaRef {
+    pub fn new(serial: u64, file_ref: FileRef) -> Self {
+        DeltaRef { serial, file_ref }
+    }
+
+    pub fn serial(&self) -> u64 { self.serial }
+}
+
+impl AsRef<FileRef> for DeltaRef {
+    fn as_ref(&self) -> &FileRef {
+        &self.file_ref
+    }
+}
+
+
+//------------ CurrentObjects ------------------------------------------------
+
+/// Defines a current set of published elements.
+///
+// Note this is mapped internally for speedy access, by hash, rather than uri
+// for two reasons:
+// a) URIs in RPKI may change in future
+// b) The publish element as it appears in an RFC8182 snapshot.xml includes
+// the uri and the base64, but not the hash. So keeping the actual elements
+// around means we can be more efficient in producing that output.
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct CurrentObjects(HashMap<EncodedHash, PublishElement>);
+
+impl Default for CurrentObjects {
+    fn default() -> Self {
+        CurrentObjects(HashMap::new())
+    }
+}
+
+impl CurrentObjects {
+    fn elements(&self) -> Vec<&PublishElement> {
+        let mut res = vec![];
+        for el in self.0.values() {
+            res.push(el)
+        }
+        res
+    }
+}
+
+
+//------------ VerificationError ---------------------------------------------
+
+/// Issues with relation to verifying deltas.
+#[derive(Clone, Debug, Display)]
+pub enum VerificationError {
+    #[display(fmt="Publishing ({}) outside of jail URI ({}) is not allowed.", _0, _1)]
+    UriOutsideJail(uri::Rsync, uri::Rsync),
+
+    #[display(fmt="File already exists for uri (use update!): {}", _0)]
+    ObjectAlreadyPresent(uri::Rsync),
+
+    #[display(fmt="File does not match hash at uri: {}", _0)]
+    NoObjectForHashAndOrUri(uri::Rsync),
+}
+
+impl VerificationError {
+    fn outside(jail: &uri::Rsync, uri: &uri::Rsync) -> Self {
+        VerificationError::UriOutsideJail(uri.clone(), jail.clone())
+    }
+
+    fn present(uri: &uri::Rsync) -> Self {
+        VerificationError::ObjectAlreadyPresent(uri.clone())
+    }
+
+    fn no_match(uri: &uri::Rsync) -> Self {
+        VerificationError::NoObjectForHashAndOrUri(uri.clone())
+    }
+}
+
+impl CurrentObjects {
+
+    fn has_match(
+        &self,
+        hash: &EncodedHash,
+        uri: &uri::Rsync
+    ) -> bool {
+        match self.0.get(hash) {
+            Some(el) => el.uri() == uri,
+            None => false
+        }
+    }
+
+    pub fn verify_delta(
+        &self,
+        delta: &DeltaElements,
+        jail: &uri::Rsync
+    ) -> Result<(), VerificationError> {
+
+        for p in delta.publishes() {
+            if ! jail.is_parent_of(p.uri()) {
+                return Err(VerificationError::outside(jail, p.uri()))
+            }
+            let hash = p.base64().to_encoded_hash();
+            if self.0.contains_key(&hash) {
+                return Err(VerificationError::present(p.uri()))
+            }
+        }
+
+        for u in delta.updates() {
+            if ! self.has_match(u.hash(), u.uri()) {
+                return Err(VerificationError::no_match(u.uri()));
+            }
+        }
+
+        for w in delta.withdraws() {
+            if ! self.has_match(w.hash(), w.uri()) {
+                return Err(VerificationError::no_match(w.uri()));
+            }
+        }
+
+        Ok(())
+    }
+
+    /// Applies a delta to CurrentObjects. This will asume that the delta
+    /// contains only valid updates for this delta.
+    pub fn apply_delta(&mut self, delta: DeltaElements) {
+        let (publishes, updates, withdraws) = delta.unwrap();
+
+        for p in publishes {
+            let hash = p.base64().to_encoded_hash();
+            self.0.insert(hash, p);
+        }
+
+        for u in updates {
+            self.0.remove(u.hash());
+            let p: PublishElement = u.into();
+            let hash = p.base64().to_encoded_hash();
+            self.0.insert(hash, p);
+        }
+
+        for w in withdraws {
+            self.0.remove(w.hash());
+        }
+    }
+
+    pub fn len(&self) -> usize { self.0.len() }
+
+    pub fn is_empty(&self) -> bool { self.0.is_empty() }
+
+    pub fn to_list_reply(&self) -> publication_data::ListReply {
+        let elements = self.0.iter().map(|el| {
+            let hash = el.0.clone();
+            let uri = el.1.uri().clone();
+            publication_data::ListElement::new(uri, hash)
+        }).collect();
+
+        publication_data::ListReply::new(elements)
+    }
+}
+
+
+//------------ Snapshot ------------------------------------------------------
+
+/// A structure to contain the RRDP snapshot data.
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct Snapshot {
+    session: String,
+    serial: u64,
+    current_objects: CurrentObjects
+}
+
+impl Snapshot {
+    pub fn new(session: String) -> Self {
+        let current_objects = CurrentObjects::default();
+        Snapshot { session, serial: 0, current_objects }
+    }
+
+    pub fn apply_delta(&mut self, delta: Delta) {
+        let (session, serial, elements) = delta.unwrap();
+        self.session = session;
+        self.serial = serial;
+        self.current_objects.apply_delta(elements)
+    }
+
+    pub fn len(&self) -> usize { self.current_objects.len() }
+
+    pub fn is_empty(&self) -> bool { self.current_objects.is_empty() }
+
+    pub fn write_xml(&self, path: &PathBuf) -> Result<EncodedHash, io::Error> {
+        let vec = XmlWriter::encode_vec(|w| {
+            let a = [
+                ("xmlns", NS),
+                ("version", VERSION),
+                ("session_id", self.session.as_ref()),
+                ("serial", &format!("{}", self.serial)),
+            ];
+
+            w.put_element(
+                "snapshot",
+                Some(&a),
+                |w| {
+                    for el in self.current_objects.elements() {
+                        let uri = el.uri.to_string();
+                        let atr = [ ("uri", uri.as_ref())];
+                        w.put_element(
+                            "publish",
+                            Some(&atr),
+                            |w| {
+                                w.put_text(el.base64.as_ref())
+                            }
+                        )?;
+                    }
+                    Ok(())
+                }
+            )
+        });
+        let bytes = Bytes::from(vec);
+
+        file::save(&bytes, path)?;
+        let hash = EncodedHash::from_content(&bytes);
+
+        Ok(hash)
+    }
+}
+
+
+//------------ DeltaElements -------------------------------------------------
+
+/// Defines the elements for an RRDP delta.
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct DeltaElements {
+    publishes: Vec<PublishElement>,
+    updates: Vec<UpdateElement>,
+    withdraws: Vec<WithdrawElement>
+}
+
+impl From<publication_data::PublishDelta> for DeltaElements {
+    fn from(d: publication_data::PublishDelta) -> Self {
+        let (pbls, upds, wdrs) = d.unwrap();
+
+        let publishes = pbls.into_iter().map(PublishElement::from).collect();
+        let updates = upds.into_iter().map(UpdateElement::from).collect();
+        let withdraws = wdrs.into_iter().map(WithdrawElement::from).collect();
+
+        DeltaElements { publishes, updates, withdraws }
+    }
+}
+
+impl DeltaElements {
+    pub fn unwrap(
+        self
+    ) -> (Vec<PublishElement>, Vec<UpdateElement>, Vec<WithdrawElement>) {
+        (self.publishes, self.updates, self.withdraws)
+    }
+
+    pub fn len(&self) -> usize {
+        self.publishes.len() + self.updates.len() + self.withdraws.len()
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.len() == 0
+    }
+
+    pub fn publishes(&self) -> &Vec<PublishElement> {
+        &self.publishes
+    }
+
+    pub fn updates(&self) -> &Vec<UpdateElement> {
+        &self.updates
+    }
+
+    pub fn withdraws(&self) -> &Vec<WithdrawElement> {
+        &self.withdraws
+    }
+}
+
+
+//------------ Delta ---------------------------------------------------------
+
+/// Defines an RRDP delta.
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct Delta {
+    session:  String,
+    serial:   u64,
+    time:     Time,
+    elements: DeltaElements
+}
+
+impl Delta {
+    pub fn new(
+        session: String,
+        serial: u64,
+        elements: DeltaElements
+    ) -> Self {
+        Delta { session, time: Time::now(), serial, elements }
+    }
+
+    pub fn session(&self) -> &str { &self.session }
+    pub fn serial(&self) -> u64 { self.serial }
+    pub fn time(&self) -> &Time { &self.time }
+    pub fn elements(&self) -> &DeltaElements { &self.elements }
+
+    /// Total number of elements
+    ///
+    /// This is a cheap approximation of the size of the delta that can help
+    /// in determining the choice of how many deltas to include in a
+    /// notification file.
+    pub fn len(&self) -> usize { self.elements.len() }
+
+    pub fn is_empty(&self) -> bool { self.elements.is_empty() }
+
+    pub fn unwrap(self) -> (String, u64, DeltaElements) {
+        (self.session, self.serial, self.elements)
+    }
+
+    pub fn write_xml(&self, path: &PathBuf) -> Result<EncodedHash, io::Error> {
+
+        let vec = XmlWriter::encode_vec(|w| {
+            let a = [
+                ("xmlns", NS),
+                ("version", VERSION),
+                ("session_id", self.session.as_ref()),
+                ("serial", &format!("{}", self.serial)),
+            ];
+
+            w.put_element(
+                "delta",
+                Some(&a),
+                |w| {
+                    for el in &self.elements.publishes {
+                        let uri = el.uri.to_string();
+                        let atr = [ ("uri", uri.as_ref())];
+                        w.put_element(
+                            "publish",
+                            Some(&atr),
+                            |w| {
+                                w.put_text(el.base64.as_ref())
+                            }
+                        )?;
+                    }
+
+                    for el in &self.elements.updates {
+                        let uri = el.uri.to_string();
+                        let atr = [
+                            ("uri", uri.as_ref()),
+                            ("hash", el.hash.as_ref())
+                        ];
+                        w.put_element(
+                            "publish",
+                            Some(&atr),
+                            |w| {
+                                w.put_text(el.base64.as_ref())
+                            }
+                        )?;
+                    }
+
+                    for el in &self.elements.withdraws {
+                        let uri = el.uri.to_string();
+                        let atr = [
+                            ("uri", uri.as_ref()),
+                            ("hash", el.hash.as_ref())
+                        ];
+                        w.put_element(
+                            "withdraw",
+                            Some(&atr),
+                            |w| { w.empty() }
+                        )?;
+                    }
+
+                    Ok(())
+                }
+            )
+        });
+
+        let bytes = Bytes::from(vec);
+        file::save(&bytes, &path)?;
+        let hash = EncodedHash::from_content(&bytes);
+
+        Ok(hash)
+    }
+
+}

src/api/rrdp_data.rs

@@ -1,537 +0,0 @@
-use std::collections::HashMap;
-use std::io;
-use std::fs::File;
-use std::num::ParseIntError;
-use std::ops::Deref;
-use std::path::PathBuf;
-use std::str::FromStr;
-use bytes::Bytes;
-use rpki::uri;
-use crate::util::xml::{AttributesError, XmlReader, XmlReaderErr, XmlWriter};
-use crate::util::file::{self, RecursorError};
-use crate::util::ext_serde;
-use util::sha256;
-
-const VERSION: &str = "1";
-const NS: &str = "http://www.ripe.net/rpki/rrdp";
-
-#[derive(Clone, Debug, Deserialize, Eq, Serialize, PartialEq)]
-pub struct PublishedObject {
-    #[serde(
-    deserialize_with = "ext_serde::de_rsync_uri",
-    serialize_with = "ext_serde::ser_rsync_uri")]
-    uri: uri::Rsync,
-
-    #[serde(
-    deserialize_with = "ext_serde::de_bytes",
-    serialize_with = "ext_serde::ser_bytes")]
-    content: Bytes,
-
-    #[serde(
-    deserialize_with = "ext_serde::de_bytes",
-    serialize_with = "ext_serde::ser_bytes")]
-    hash: Bytes
-}
-
-impl PublishedObject {
-    pub fn new(uri: uri::Rsync, content: Bytes) -> Self {
-        let hash = sha256(&content);
-        PublishedObject { uri , content, hash }
-    }
-
-    pub fn uri(&self) -> &uri::Rsync{ &self.uri }
-    pub fn content(&self) -> &Bytes { &self.content }
-    pub fn hash(&self) -> &Bytes { &self.hash }
-}
-
-#[derive(Clone, Debug, Deserialize, Serialize)]
-pub struct Snapshot {
-    session: String,
-    serial: usize,
-    objects: HashMap<String, Vec<PublishedObject>>
-}
-
-impl Snapshot {
-    pub fn new(
-        session: String,
-        serial: usize,
-        objects: HashMap<String, Vec<PublishedObject>>
-    ) -> Self {
-        Snapshot { session, serial, objects }
-    }
-
-    pub fn objects(&self) -> &HashMap<String, Vec<PublishedObject>> {
-        &self.objects
-    }
-
-    pub fn to_xml(&self) -> Vec<u8> {
-        XmlWriter::encode_vec(|w| {
-            let a = [
-                ("xmlns", NS),
-                ("version", VERSION),
-                ("session_id", self.session.as_ref()),
-                ("serial", &format!("{}", self.serial)),
-            ];
-
-            w.put_element(
-                "snapshot",
-                Some(&a),
-                |w| {
-                    for uri in self.objects.keys() {
-                        let objects = &self.objects[uri];
-                        for cf in objects {
-                            let uri = cf.uri.to_string();
-                            let a = [ ("uri", uri.as_ref()) ];
-                            w.put_element(
-                                "publish",
-                                Some(&a),
-                                |w| {
-                                    w.put_blob(&cf.content)
-                                }
-                            )?;
-                        }
-                    }
-                    Ok(())
-                }
-            )
-        })
-    }
-}
-
-
-
-//------------ Notification --------------------------------------------------
-
-#[derive(Clone, Debug, Deserialize, Serialize)]
-pub struct Notification {
-    session_id: String,
-    serial:     usize,
-    snapshot:   SnapshotRef,
-    deltas:     Vec<DeltaRef>
-}
-
-/// # Accessors
-///
-impl Notification {
-    pub fn serial(&self) -> usize {
-        self.serial
-    }
-    pub fn session_id(&self) -> &String { &self.session_id }
-
-    pub fn deltas(&self) -> &Vec<DeltaRef> {
-        &self.deltas
-    }
-}
-
-/// # Load and save
-///
-impl Notification {
-
-    /// Build up the current notification based on what's on disk.
-    /// Note that this will return None, if there is nothing on disk.
-    ///
-    /// Also note that in future we may want to cache things for
-    /// efficiency (but make it work with multi master).
-    pub fn build(
-        path: &PathBuf,
-        base_uri: &uri::Http,
-        rrdp_base: &PathBuf
-    ) -> Option<Notification> {
-        let mut builder = NotificationBuilder::new();
-
-        match XmlReader::open(path, |r| -> Result<(), Error> {
-            r.take_named_element("notification", |mut a, r| {
-                {
-                    // process attributes
-                    builder.with_session_id(a.take_req("session_id")?);
-                    let serial = usize::from_str(a.take_req("serial")?.as_ref())?;
-                    builder.with_serial(serial);
-                    // about NS
-                }
-
-                {
-                    // expect snapshot ref
-                    r.take_named_element(
-                        "snapshot",
-                        |mut a, _r| -> Result<(), Error> {
-                            let uri = uri::Http::from_string(a.take_req("uri")?)?;
-                            let file_info = FileInfo::for_uri(
-                                &uri,
-                                base_uri,
-                                rrdp_base
-                            )?;
-
-                            builder.with_snapshot(
-                                SnapshotRef { file_info }
-                            );
-                            Ok(())
-                        })?;
-                }
-
-                {
-                    // deltas
-                    loop {
-                        let d = r.take_opt_element(|t, mut a, _r| {
-                            match t.name.as_ref() {
-                                "delta" => {
-                                    let uri = uri::Http::from_string(
-                                        a.take_req("uri")?
-                                    )?;
-                                    let serial = usize::from_str(
-                                        a.take_req("serial")?.as_ref()
-                                    )?;
-                                    let file_info = FileInfo::for_uri(
-                                        &uri,
-                                        base_uri,
-                                        rrdp_base
-                                    )?;
-
-                                    Ok(Some(DeltaRef {
-                                        serial,
-                                        file_info
-                                    }))
-                                },
-                                _ => Err(Error::NotificationFileError)
-                            }
-                        })?;
-                        match d {
-                            None => break,
-                            Some(d) => builder.add_delta(d)
-                        }
-                    }
-                    Ok(())
-                }
-            })
-        }).map_err(|_| Error::NotificationFileError) {
-            Ok(_) => Some(builder.build()),
-            Err(_) => None
-        }
-    }
-
-    /// Saves a notification file as RFC8182 XML.
-    pub fn save(&self, path: &PathBuf) -> Result<(), Error> {
-        debug!("Writing notification file: {}", path.to_string_lossy());
-        let mut file = file::create_file_with_path(&path)?;
-
-        XmlWriter::encode_to_file(& mut file, |w| {
-
-            let a = [
-                ("xmlns", NS),
-                ("version", VERSION),
-                ("session_id", self.session_id.as_ref()),
-                ("serial", &format!("{}", self.serial)),
-            ];
-
-            w.put_element(
-                "notification",
-                Some(&a),
-                |w| {
-                    {
-                        // snapshot ref
-                        let uri = self.snapshot.uri.to_string();
-                        let hash = &self.snapshot.hash;
-                        let a = [
-                            ("uri", uri.as_str()),
-                            ("hash", hash)
-                        ];
-                        w.put_element(
-                            "snapshot",
-                            Some(&a),
-                            |w| { w.empty() }
-                        )?;
-                    }
-
-                    {
-                        // delta refs
-                        for delta in &self.deltas {
-                            let serial = format!("{}", delta.serial);
-                            let uri = delta.uri.to_string();
-                            let hash = &delta.hash;
-                            let a = [
-                                ("serial", serial.as_ref()),
-                                ("uri", uri.as_str()),
-                                ("hash", hash)
-                            ];
-                            w.put_element(
-                                "delta",
-                                Some(&a),
-                                |w| { w.empty() }
-                            )?;
-                        }
-                    }
-
-                    Ok(())
-                }
-            )
-        })?;
-
-        Ok(())
-    }
-}
-
-
-//------------ SnapshotRef ---------------------------------------------------
-
-#[derive(Clone, Debug, Deserialize, Serialize)]
-pub struct SnapshotRef {
-    file_info:  FileInfo
-}
-
-impl SnapshotRef {
-    pub fn new(file_info: FileInfo) -> Self {
-        SnapshotRef { file_info }
-    }
-}
-
-impl Deref for SnapshotRef {
-    type Target = FileInfo;
-
-    fn deref(&self) -> &FileInfo {
-        &self.file_info
-    }
-}
-
-
-//------------ DeltaRef ------------------------------------------------------
-
-#[derive(Clone, Debug, Deserialize, Serialize)]
-pub struct DeltaRef {
-    serial:     usize,
-    file_info:  FileInfo,
-}
-
-impl DeltaRef {
-    pub fn new(serial: usize, file_info: FileInfo) -> Self {
-        DeltaRef { serial, file_info }
-    }
-    pub fn serial(&self) -> &usize {
-        &self.serial
-    }
-}
-
-
-impl Deref for DeltaRef {
-    type Target = FileInfo;
-
-    fn deref(&self) -> &FileInfo {
-        &self.file_info
-    }
-}
-
-
-//------------ FileInfo ------------------------------------------------------
-
-#[derive(Clone, Debug, Deserialize, Serialize)]
-pub struct FileInfo {
-    #[serde(
-    deserialize_with = "ext_serde::de_http_uri",
-    serialize_with = "ext_serde::ser_http_uri")]
-    uri:   uri::Http,
-    hash:  String,
-    size:  usize
-}
-
-impl FileInfo {
-    fn new(uri: uri::Http, hash: String, size: usize) -> FileInfo {
-        FileInfo { uri, hash, size}
-    }
-
-    pub fn for_path_and_uri(
-        path: &PathBuf,
-        uri: uri::Http
-    ) -> Result<FileInfo, Error> {
-        let bytes = {
-            use std::io::Read;
-
-            let mut f = File::open(path)?;
-            let mut bytes = Vec::new();
-            f.read_to_end(&mut bytes)?;
-            bytes
-        };
-
-        let size = bytes.len();
-
-        let hash = {
-            use crate::util::sha256;
-            use bytes::Bytes;
-
-            hex::encode(&sha256(&Bytes::from(bytes)))
-        };
-
-        Ok(FileInfo::new(uri, hash, size))
-    }
-
-    pub fn for_uri(
-        uri: &uri::Http,
-        base_uri: &uri::Http,
-        rrdp_base: &PathBuf
-    ) -> Result<FileInfo, Error> {
-        let base_string = base_uri.to_string();
-        let uri_string  = uri.to_string();
-
-        if ! uri_string.as_str().starts_with(base_string.as_str()) {
-            Err(Error::NotificationFileError)
-        } else {
-            let (_, rel) = uri_string.split_at(base_string.len());
-            let mut path = rrdp_base.clone();
-            path.push(rel);
-
-            FileInfo::for_path_and_uri(&path, uri.clone())
-        }
-    }
-
-    pub fn for_path(
-        path: &PathBuf,
-        base_uri: &uri::Http,
-        rrdp_base: &PathBuf
-    ) -> Result<FileInfo, Error> {
-        let relative = path.strip_prefix(rrdp_base)
-            .map_err(|_| Error::UriConfigError)?.to_string_lossy();
-        let base_uri = base_uri.to_string();
-        let uri = uri::Http::from_string(
-            format!("{}{}", base_uri, relative)
-        ).map_err(|_| Error::UriConfigError)?;
-
-        FileInfo::for_path_and_uri(path, uri)
-    }
-
-
-}
-
-
-//------------ NotificationBuilder -------------------------------------------
-
-#[derive(Default)]
-pub struct NotificationBuilder {
-    serial: Option<usize>,
-    session_id: Option<String>,
-    snapshot: Option<SnapshotRef>,
-    deltas: Vec<DeltaRef>
-}
-
-impl NotificationBuilder {
-    pub fn new() -> Self {
-        Self::default()
-    }
-
-    pub fn with_serial(&mut self, serial: usize) {
-        self.serial = Some(serial);
-    }
-
-    pub fn with_session_id(&mut self, session_id: String) {
-        self.session_id = Some(session_id);
-    }
-
-    pub fn with_snapshot(&mut self, snapshot: SnapshotRef) {
-        self.snapshot = Some(snapshot);
-    }
-
-    pub fn with_deltas(&mut self, deltas: Vec<DeltaRef>) {
-        self.deltas = deltas;
-    }
-
-    pub fn add_delta(&mut self, delta: DeltaRef) {
-        self.deltas.push(delta);
-    }
-
-    pub fn add_delta_to_start(&mut self, delta: DeltaRef) {
-        self.deltas.insert(0, delta);
-    }
-
-    /// Keeps at least two deltas, and beyond that only if the size is
-    /// smaller than the snapshot.
-    ///
-    /// Note we may add something to exclude old deltas later, if we find
-    /// that e.g. access to old deltas is very infrequent and excluding
-    /// them would shrink the notification file size.
-    fn curate_deltas(&mut self) {
-        let size_snapshot = match &self.snapshot {
-            Some(snapshot) => snapshot.size,
-            None => 0
-        };
-        let mut total_deltas = 0;
-        let mut count = 0;
-
-        self.deltas.retain(|d| {
-            count += 1;
-            total_deltas += d.size;
-            count <= 2 || total_deltas < size_snapshot
-        })
-
-    }
-
-
-    /// Builds the notification, panics if any of the options are not set.
-    /// This can only happen if there is a bug.
-    pub fn build(mut self) -> Notification {
-        self.curate_deltas();
-        Notification {
-            serial: self.serial.unwrap(),
-            session_id: self.session_id.unwrap(),
-            snapshot: self.snapshot.unwrap(),
-            deltas: self.deltas
-        }
-    }
-}
-
-
-//------------ Error ---------------------------------------------------------
-
-#[derive(Debug, Display)]
-pub enum Error {
-    #[display(fmt="{}", _0)]
-    IoError(io::Error),
-
-    #[display(fmt="{}", _0)]
-    RecursorError(RecursorError),
-
-    #[display(fmt="{}", _0)]
-    UriError(uri::Error),
-
-    #[display(fmt="{}", _0)]
-    XmlReaderErr(XmlReaderErr),
-
-    #[display(fmt="{}", _0)]
-    AttributesError(AttributesError),
-
-    #[display(fmt="{}", _0)]
-    ParseIntError(ParseIntError),
-
-    #[display(fmt="Error with notification file.")]
-    NotificationFileError,
-
-    #[display(fmt="Error with uri config.")]
-    UriConfigError
-}
-
-impl From<io::Error> for Error {
-    fn from(e: io::Error) -> Self {
-        Error::IoError(e)
-    }
-}
-
-impl From<RecursorError> for Error {
-    fn from(e: RecursorError) -> Self {
-        Error::RecursorError(e)
-    }
-}
-
-impl From<uri::Error> for Error {
-    fn from(e: uri::Error) -> Self { Error::UriError(e) }
-}
-
-impl From<XmlReaderErr> for Error {
-    fn from(e: XmlReaderErr) -> Self {
-        Error::XmlReaderErr(e)
-    }
-}
-
-impl From<AttributesError> for Error {
-    fn from(e: AttributesError) -> Self {
-        Error::AttributesError(e)
-    }
-}
-
-impl From<ParseIntError> for Error {
-    fn from(e: ParseIntError) -> Self { Error::ParseIntError(e) }
-}