move change publisher to own module

4 files changed, 90 insertions, 82 deletions

diff --git a/src/swarm/changes.rs b/src/swarm/changes.rs
new file mode 100644
index 0000000..10a3bfe
--- /dev/null
+++ b/src/swarm/changes.rs
@@ -0,0 +1,88 @@
+//! The change publisher: after document changes, which store paths they touched
+//! (`AppState::change_events`, for the API's streams, the dashboard and remakes), and
+//! a nudge to topology when the member list changed.
+
+use std::sync::Arc;
+
+use tracing::{info, warn};
+
+use crate::document::store::ChangeSet;
+use crate::state::AppState;
+
+fn patch_store_path(patch: &automerge::Patch) -> Vec<String> {
+    let mut path: Vec<String> = patch
+        .path
+        .iter()
+        .map(|(_, prop)| match prop {
+            automerge::Prop::Map(key) => key.clone(),
+            automerge::Prop::Seq(index) => index.to_string(),
+        })
+        .collect();
+
+    match &patch.action {
+        automerge::PatchAction::PutMap { key, .. }
+        | automerge::PatchAction::DeleteMap { key, .. } => path.push(key.clone()),
+        // List, text and counter changes: the containing object changed.
+        _ => {}
+    }
+    path
+}
+
+pub(crate) async fn run_change_publisher(state: AppState) {
+    let mut changed = state.change_tx.subscribe();
+    let mut last_heads = state.write_doc(|doc| doc.get_heads()).await;
+
+    info!("Starting change publisher");
+
+    while changed.changed().await.is_ok() {
+        let before = last_heads.clone();
+
+        let (heads, paths, conflicts) = state
+            .write_doc(move |doc| {
+                let heads = doc.get_heads();
+                if heads == before {
+                    return (heads, Vec::new(), Vec::new());
+                }
+
+                // After an epoch switch the previous heads belong to the old document:
+                // report the root as changed, which every stream is related to.
+                if !before.iter().all(|h| doc.get_change_by_hash(h).is_some()) {
+                    return (heads, vec![Vec::new()], Vec::new());
+                }
+
+                let mut paths = Vec::new();
+                let mut conflicts = Vec::new();
+                for patch in doc.diff(&before, &heads) {
+                    let path = patch_store_path(&patch);
+                    if matches!(
+                        patch.action,
+                        automerge::PatchAction::PutMap { conflict: true, .. }
+                    ) {
+                        conflicts.push(path.clone());
+                    }
+                    paths.push(path);
+                }
+                (heads, paths, conflicts)
+            })
+            .await;
+
+        if heads == last_heads {
+            continue;
+        }
+        last_heads = heads;
+
+        for path in &conflicts {
+            warn!(path = ?path, "Concurrent writes produced a conflict");
+        }
+
+        if paths
+            .iter()
+            .any(|path| crate::swarm::topology::affects_neighbours(path))
+        {
+            state.topology_changed.notify_one();
+        }
+
+        // Fails only when nobody is subscribed.
+        let _ = state.change_events.send(Arc::new(ChangeSet { paths }));
+    }
+}
diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs
index 7fb8e16..9cdf154 100644
--- a/src/swarm/mod.rs
+++ b/src/swarm/mod.rs
@@ -4,6 +4,7 @@
 pub(crate) mod auth;
 pub(crate) mod blob;
 pub(crate) mod bootstrap;
+pub(crate) mod changes;
 pub(crate) mod links;
 pub(crate) mod members;
 pub(crate) mod pair;
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 05c0c34..e89c2e2 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -10,7 +10,6 @@ use tokio::sync::mpsc;
 use tokio::time::Instant;
 use tracing::{debug, info, trace, warn};
 
-use crate::document::store::ChangeSet;
 use crate::locks::MutexExt;
 use crate::state::{AppState, Liveness, PeerState, Standing};
 use crate::swarm::peer::{peer_swarm, send_hello_once};
@@ -618,83 +617,3 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece
         }
     }
 }
-
-// --- change publisher --------------------------------------------------------
-
-fn patch_store_path(patch: &automerge::Patch) -> Vec<String> {
-    let mut path: Vec<String> = patch
-        .path
-        .iter()
-        .map(|(_, prop)| match prop {
-            automerge::Prop::Map(key) => key.clone(),
-            automerge::Prop::Seq(index) => index.to_string(),
-        })
-        .collect();
-
-    match &patch.action {
-        automerge::PatchAction::PutMap { key, .. }
-        | automerge::PatchAction::DeleteMap { key, .. } => path.push(key.clone()),
-        // List, text and counter changes: the containing object changed.
-        _ => {}
-    }
-    path
-}
-
-pub(crate) async fn run_change_publisher(state: AppState) {
-    let mut changed = state.change_tx.subscribe();
-    let mut last_heads = state.write_doc(|doc| doc.get_heads()).await;
-
-    info!("Starting change publisher");
-
-    while changed.changed().await.is_ok() {
-        let before = last_heads.clone();
-
-        let (heads, paths, conflicts) = state
-            .write_doc(move |doc| {
-                let heads = doc.get_heads();
-                if heads == before {
-                    return (heads, Vec::new(), Vec::new());
-                }
-
-                // After an epoch switch the previous heads belong to the old document:
-                // report the root as changed, which every stream is related to.
-                if !before.iter().all(|h| doc.get_change_by_hash(h).is_some()) {
-                    return (heads, vec![Vec::new()], Vec::new());
-                }
-
-                let mut paths = Vec::new();
-                let mut conflicts = Vec::new();
-                for patch in doc.diff(&before, &heads) {
-                    let path = patch_store_path(&patch);
-                    if matches!(
-                        patch.action,
-                        automerge::PatchAction::PutMap { conflict: true, .. }
-                    ) {
-                        conflicts.push(path.clone());
-                    }
-                    paths.push(path);
-                }
-                (heads, paths, conflicts)
-            })
-            .await;
-
-        if heads == last_heads {
-            continue;
-        }
-        last_heads = heads;
-
-        for path in &conflicts {
-            warn!(path = ?path, "Concurrent writes produced a conflict");
-        }
-
-        if paths
-            .iter()
-            .any(|path| crate::swarm::topology::affects_neighbours(path))
-        {
-            state.topology_changed.notify_one();
-        }
-
-        // Fails only when nobody is subscribed.
-        let _ = state.change_events.send(Arc::new(ChangeSet { paths }));
-    }
-}
diff --git a/src/swarm/topology.rs b/src/swarm/topology.rs
index 063fc41..c861101 100644
--- a/src/swarm/topology.rs
+++ b/src/swarm/topology.rs
@@ -161,7 +161,7 @@ pub(crate) struct NeighbourChange {
 }
 
 /// Recomputes the neighbours whenever something they depend on may have changed
-/// (`AppState::topology_changed`): the member list (`swarm::sync::run_change_publisher`),
+/// (`AppState::topology_changed`): the member list (`swarm::changes::run_change_publisher`),
 /// a peer becoming unreachable or reachable again (`swarm::peer`), and every 5 s as a
 /// fallback. A burst of notifications coalesces into one recomputation.
 pub(crate) async fn run_worker(state: AppState) {