Emit Download and Append events

Author: Frando

src/event.rs

@@ -1,3 +1,8 @@
-/// Events emitted.
+/// An event emitted by a Feed.
 #[derive(Debug, Clone, PartialEq)]
-pub enum Event {}
+pub enum Event {
+    /// A new block has been appended.
+    Append,
+    /// A new block has been downloaded.
+    Download(u64),
+}

src/feed.rs

@@ -1,5 +1,6 @@
 //! Hypercore's main abstraction. Exposes an append-only, secure log structure.
 
+use crate::event::Event;
 use crate::feed_builder::FeedBuilder;
 use crate::replicate::{Message, Peer};
 pub use crate::storage::{Node, NodeTrait, Storage, Store};
@@ -12,6 +13,10 @@ use crate::crypto::{
 use crate::proof::Proof;
 use anyhow::{bail, ensure, Result};
 use flat_tree as flat;
+use futures::channel::mpsc::{
+    unbounded as channel, UnboundedReceiver as Receiver, UnboundedSender as Sender,
+};
+use futures::sink::SinkExt;
 use pretty_hash::fmt as pretty_fmt;
 use random_access_disk::RandomAccessDisk;
 use random_access_memory::RandomAccessMemory;
@@ -72,6 +77,7 @@ where
     pub(crate) bitfield: Bitfield,
     pub(crate) tree: TreeIndex,
     pub(crate) peers: Vec<Peer>,
+    pub(crate) subscribers: Vec<Sender<Event>>,
 }
 
 impl<T> Feed<T>
@@ -163,9 +169,25 @@ where
         self.tree.set(tree_index(index));
         self.length += 1;
 
+        self.emit(Event::Append).await;
+
         Ok(())
     }
 
+    /// Subscribe to events emitted by this feed.
+    pub fn subscribe(&mut self) -> Receiver<Event> {
+        let (send, recv) = channel();
+        self.subscribers.push(send);
+        recv
+    }
+
+    /// Emit an event on the feed.
+    async fn emit(&self, event: Event) {
+        for mut sender in self.subscribers.iter() {
+            sender.send(event.clone()).await.unwrap();
+        }
+    }
+
     /// Get the block of data at the tip of the feed. This will be the most
     /// recently appended block.
     #[inline]
@@ -399,7 +421,7 @@ where
 
         if let Some(_data) = data {
             if self.bitfield.set(index, true).is_changed() {
-                // TODO: emit "download" event
+                self.emit(Event::Download(index)).await;
             }
             // TODO: check peers.length, call ._announce if peers exist.
         }

src/feed_builder.rs

@@ -56,6 +56,7 @@ where
             secret_key: self.secret_key,
             storage: self.storage,
             peers: vec![],
+            subscribers: vec![],
         })
     }
 }

tests/feed.rs

@@ -3,7 +3,8 @@ extern crate random_access_memory as ram;
 mod common;
 
 use common::create_feed;
-use hypercore::{generate_keypair, Feed, NodeTrait, PublicKey, SecretKey, Storage};
+use futures::stream::StreamExt;
+use hypercore::{generate_keypair, Event, Feed, NodeTrait, PublicKey, SecretKey, Storage};
 use random_access_storage::RandomAccess;
 use std::env::temp_dir;
 use std::fmt::Debug;
@@ -152,14 +153,8 @@ async fn put() {
 async fn put_with_data() {
     // Create a writable feed.
     let mut a = create_feed(50).await.unwrap();
-
     // Create a second feed with the first feed's key.
-    let (public, secret) = copy_keys(&a);
-    let storage = Storage::new_memory().await.unwrap();
-    let mut b = Feed::builder(public, storage)
-        .secret_key(secret)
-        .build()
-        .unwrap();
+    let mut b = create_clone(&a).await.unwrap();
 
     // Append 4 blocks of data to the writable feed.
     a.append(b"hi").await.unwrap();
@@ -217,23 +212,6 @@ async fn create_with_stored_keys() {
     );
 }
 
-fn copy_keys(
-    feed: &Feed<impl RandomAccess<Error = Box<dyn std::error::Error + Send + Sync>> + Debug + Send>,
-) -> (PublicKey, SecretKey) {
-    match &feed.secret_key() {
-        Some(secret) => {
-            let secret = secret.to_bytes();
-            let public = &feed.public_key().to_bytes();
-
-            let public = PublicKey::from_bytes(public).unwrap();
-            let secret = SecretKey::from_bytes(&secret).unwrap();
-
-            (public, secret)
-        }
-        _ => panic!("<tests/common>: Could not access secret key"),
-    }
-}
-
 #[async_std::test]
 async fn audit() {
     let mut feed = create_feed(50).await.unwrap();
@@ -298,3 +276,92 @@ async fn audit_bad_data() {
         }
     }
 }
+
+#[async_std::test]
+async fn events_append() {
+    let mut feed = create_feed(50).await.unwrap();
+    let event_task = collect_events(&mut feed, 3);
+    feed.append(br#"one"#).await.unwrap();
+    feed.append(br#"two"#).await.unwrap();
+    feed.append(br#"three"#).await.unwrap();
+
+    let event_list = event_task.await;
+    let mut expected = vec![];
+    for _i in 0..3 {
+        expected.push(Event::Append);
+    }
+    assert_eq!(event_list, expected, "Correct events emitted")
+}
+
+#[async_std::test]
+async fn events_download() {
+    let mut a = create_feed(50).await.unwrap();
+    // Create a second feed with the first feed's key.
+    let mut b = create_clone(&a).await.unwrap();
+
+    let event_task = collect_events(&mut b, 3);
+
+    a.append(b"one").await.unwrap();
+    a.append(b"two").await.unwrap();
+    a.append(b"three").await.unwrap();
+
+    for i in 0..3 {
+        let a_proof = a.proof(i, false).await.unwrap();
+        let a_data = a.get(i).await.unwrap();
+        b.put(i, a_data.as_deref(), a_proof).await.unwrap();
+    }
+
+    let event_list = event_task.await;
+
+    let mut expected = vec![];
+    for i in 0..3 {
+        expected.push(Event::Download(i));
+    }
+    assert_eq!(event_list, expected, "Correct events emitted")
+}
+
+async fn create_clone(
+    feed: &Feed<impl RandomAccess<Error = Box<dyn std::error::Error + Send + Sync>> + Debug + Send>,
+) -> Result<Feed<ram::RandomAccessMemory>, anyhow::Error> {
+    let (public, secret) = copy_keys(&feed);
+    let storage = Storage::new_memory().await?;
+    let clone = Feed::builder(public, storage).secret_key(secret).build()?;
+    Ok(clone)
+}
+
+fn copy_keys(
+    feed: &Feed<impl RandomAccess<Error = Box<dyn std::error::Error + Send + Sync>> + Debug + Send>,
+) -> (PublicKey, SecretKey) {
+    match &feed.secret_key() {
+        Some(secret) => {
+            let secret = secret.to_bytes();
+            let public = &feed.public_key().to_bytes();
+
+            let public = PublicKey::from_bytes(public).unwrap();
+            let secret = SecretKey::from_bytes(&secret).unwrap();
+
+            (public, secret)
+        }
+        _ => panic!("<tests/common>: Could not access secret key"),
+    }
+}
+
+fn collect_events(
+    feed: &mut Feed<
+        impl RandomAccess<Error = Box<dyn std::error::Error + Send + Sync>> + Debug + Send,
+    >,
+    n: usize,
+) -> async_std::task::JoinHandle<Vec<Event>> {
+    let mut events = feed.subscribe();
+    let event_task = async_std::task::spawn(async move {
+        let mut event_list = vec![];
+        while let Some(event) = events.next().await {
+            event_list.push(event);
+            if event_list.len() == n {
+                return event_list;
+            }
+        }
+        event_list
+    });
+    event_task
+}