update to samod 0.3.0 and improve document subscription handling

iroh-automerge-repo/Cargo.toml

@@ -9,7 +9,7 @@ bytes = "1.10.1"
 derive_more = { version = "2.0.1", features = ["debug"] }
 iroh = "0.91.1"
 n0-future = "0.2"
-samod = { version = "0.2.1", features = ["tokio"] }
+samod = { version = "0.3.0", features = ["tokio"] }
 tokio = { version = "1.47.1", features = ["sync", "net"] }
 tokio-util = { version = "0.7.15", features = ["codec"] }
 tracing = "0.1.41"

iroh-automerge-repo/src/lib.rs

@@ -2,32 +2,32 @@
 //! in rust that speak the automerge repo protocol.
 use anyhow::Result;
 use codec::Codec;
-use samod::{ConnDirection, ConnFinishedReason, Samod};
+use samod::{ConnDirection, ConnFinishedReason, Repo};
 use tokio_util::codec::{FramedRead, FramedWrite};
 
 mod codec;
 
-/// Combines an [`iroh::Endpoint`] with a [`Samod`] (automerge repository) and
+/// Combines an [`iroh::Endpoint`] with a [`Repo`] (automerge repository) and
 /// implements [`iroh::protocol::ProtocolHandler`] to accept incoming connections
 /// in an [`iroh::protocol::Router`].
 #[derive(derive_more::Debug, Clone)]
 pub struct IrohRepo {
     endpoint: iroh::Endpoint,
     #[debug(skip)]
-    repo: Samod,
+    repo: Repo,
 }
 
 impl IrohRepo {
     pub const SYNC_ALPN: &[u8] = b"iroh/automerge-repo/1";
 
     /// Constructs a new [`IrohRepo`].
-    pub fn new(endpoint: iroh::Endpoint, repo: Samod) -> Self {
+    pub fn new(endpoint: iroh::Endpoint, repo: Repo) -> Self {
         IrohRepo { endpoint, repo }
     }
 
     /// Attempts to continuously sync with a peer at given address.
     ///
-    /// To wait for the connection to be established use [`Samod::when_connected`]
+    /// To wait for the connection to be established use [`Repo::when_connected`]
     /// (accessible via [`Self::repo`]: `iroh_repo.repo().when_connected(..)`).
     /// with the other node's string-encoded [`NodeId`] as the [`PeerId`].
     ///
@@ -56,8 +56,8 @@ impl IrohRepo {
         Ok(conn_finished)
     }
 
-    /// Returns a reference to the stored [`Samod`] instance inside.
-    pub fn repo(&self) -> &Samod {
+    /// Returns a reference to the stored [`Repo`] instance inside.
+    pub fn repo(&self) -> &Repo {
         &self.repo
     }
 }

iroh-automerge-repo/src/main.rs

@@ -7,7 +7,7 @@ use hex::encode;
 use iroh::NodeId;
 use iroh_automerge_repo::IrohRepo;
 
-use samod::{DocumentId, PeerId, Samod, storage::TokioFilesystemStorage};
+use samod::{DocumentId, PeerId, Repo, storage::TokioFilesystemStorage};
 
 #[derive(Parser, Debug)]
 #[command(author, version, about, long_about = None)]
@@ -123,7 +123,7 @@ async fn main() -> anyhow::Result<()> {
 
     println!("Node ID: {}", endpoint.node_id());
 
-    let samod = Samod::build_tokio()
+    let samod = Repo::build_tokio()
         .with_peer_id(PeerId::from_string(endpoint.node_id().to_string()))
         .with_storage(TokioFilesystemStorage::new(format!(
             "{}/{}",
@@ -217,23 +217,11 @@ async fn main() -> anyhow::Result<()> {
                 anyhow::Ok(())
             })?;
 
-            // Set up polling for changes (no push available yet)
+            // Listen for change events and print latest document contents
             tokio::spawn(async move {
-                // Track the last known heads to detect changes
-                let mut last_heads = doc.with_document(|doc| doc.get_heads());
-                loop {
-                    tokio::time::sleep(tokio::time::Duration::from_millis(1000)).await;
-
-                    let current_heads = doc.with_document(|doc| doc.get_heads());
-                    if current_heads == last_heads {
-                        continue;
-                    }
-
-                    last_heads = current_heads;
-
-                    println!("Document changed! New state:");
-
-                    // When changes are detected, print the updated document contents...
+                use n0_future::StreamExt;
+                let mut changes = doc.changes();
+                while let Some(_change) = changes.next().await {
                     if let Err(e) = doc.with_document(|current_doc| {
                         for key in current_doc.keys(automerge::ROOT) {
                             let (value, _) = current_doc