Fix config and IPC lifecycle issues and remove obsolete code

shell/ferese-shell/src/activities/subscription.rs

@@ -1,62 +1,14 @@
 use std::sync::Arc;
-use std::time::Duration;
 
 use cosmic::iced::Subscription;
-use cosmic::iced::futures::SinkExt;
 use ferese_ipc::activities::Snapshot;
-use serde_json::json;
 
 pub(crate) fn subscription() -> Subscription<Arc<Snapshot>> {
     Subscription::run(stream)
 }
 
 fn stream() -> impl cosmic::iced::futures::Stream<Item = Arc<Snapshot>> {
-    cosmic::iced::stream::channel(1, async |mut output| {
-        loop {
-            let connection = tokio::task::spawn_blocking(ferese_ipc::Connection::connect).await;
-            let Ok(Ok(mut connection)) = connection else {
-                tokio::time::sleep(Duration::from_secs(5)).await;
-                continue;
-            };
-            let Ok(cancellation) = connection.cancellation() else {
-                return;
-            };
-            let (send, mut receive) = tokio::sync::watch::channel(None);
-            tokio::task::spawn_blocking(move || {
-                let Ok(value) = connection.call("activity-get", json!({})) else {
-                    return;
-                };
-                let Ok(mut snapshot) = serde_json::from_value::<Snapshot>(value) else {
-                    return;
-                };
-                loop {
-                    let revision = snapshot.revision;
-                    if send.send(Some(Arc::new(snapshot))).is_err() {
-                        return;
-                    }
-                    match connection
-                        .wait("activity-watch", json!({"since":revision}))
-                        .and_then(|value| serde_json::from_value(value).map_err(|e| e.to_string()))
-                    {
-                        Ok(next) => snapshot = next,
-                        Err(_) => return,
-                    }
-                }
-            });
-            while receive.changed().await.is_ok() {
-                let snapshot = receive.borrow_and_update().clone();
-                if let Some(snapshot) = snapshot
-                    && output.send(snapshot).await.is_err()
-                {
-                    return;
-                }
-            }
-
-            drop(cancellation);
-            if output.send(Arc::new(Snapshot::default())).await.is_err() {
-                return;
-            }
-            tokio::time::sleep(Duration::from_secs(5)).await;
-        }
+    crate::ipc_subscription::stream("activity-get", "activity-watch", |snapshot: &Snapshot| {
+        snapshot.revision
     })
 }

shell/ferese-shell/src/config.rs

@@ -341,14 +341,14 @@ impl Default for SoftShadowConfig {
     }
 }
 
-pub(crate) fn load() -> Result<ShellConfig, String> {
+pub(crate) fn load(snapshot: &ferese_ipc::theme::Snapshot) -> Result<ShellConfig, String> {
     let Some(path) = config_path() else {
         return Ok(ShellConfig::default());
     };
-    load_path(&path)
+    load_path(&path, snapshot)
 }
 
-fn load_path(path: &std::path::Path) -> Result<ShellConfig, String> {
+fn load_path(path: &std::path::Path, snapshot: &ferese_ipc::theme::Snapshot) -> Result<ShellConfig, String> {
     match fs::symlink_metadata(path) {
         Ok(_) => {}
         Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(ShellConfig::default()),
@@ -364,13 +364,15 @@ fn load_path(path: &std::path::Path) -> Result<ShellConfig, String> {
     if source.len() > ferese_config::file::LIMIT {
         return Err(format!("{} exceeds 60 KiB", path.display()));
     }
-    parse_source(&source).map_err(|error| format!("invalid configuration in {}: {error}", path.display()))
+    parse_source(&source, snapshot).map_err(|error| format!("invalid configuration in {}: {error}", path.display()))
 }
 
-pub(crate) fn parse_source(source: &str) -> Result<ShellConfig, ferese_config::Error> {
+pub(crate) fn parse_source(
+    source: &str,
+    snapshot: &ferese_ipc::theme::Snapshot,
+) -> Result<ShellConfig, ferese_config::Error> {
     let document = ferese_config::Document::parse(source)?;
     ferese_config::theme::validate_overrides(&document).map_err(ferese_config::Error::from)?;
-    let snapshot = ferese_theme_client::service::current();
     let mut config = parse_document(
         &document.with_theme(&snapshot.presented),
         snapshot.presented.appearance,
@@ -616,10 +618,13 @@ mod tests {
                     format!("{section} {{ geometry {{ {key} 8; }}; }}")
                 };
                 assert!(
-                    parse_source(&format!("theme {{ {layer}; }}"))
-                        .unwrap_err()
-                        .to_string()
-                        .contains(&key.replace('-', "_"))
+                    parse_source(
+                        &format!("theme {{ {layer}; }}"),
+                        &ferese_theme_client::service::fallback()
+                    )
+                    .unwrap_err()
+                    .to_string()
+                    .contains(&key.replace('-', "_"))
                 );
             }
         }
@@ -669,7 +674,7 @@ mod tests {
     #[test]
     fn removed_desktop_widgets_allow_startup_and_reload() {
         let source = "desktop-widgets { clock { enabled #true; }; }";
-        parse_source(source).unwrap();
+        parse_source(source, &ferese_theme_client::service::fallback()).unwrap();
         parse_test_source(source).unwrap();
     }
 
@@ -802,28 +807,38 @@ theme {
         let source =
             "notifications { do-not-disturb #true; }\nstatus { bar-layout islands; battery-percentage #false; }";
         std::fs::write(&path, source).unwrap();
-        let error = super::load_path(&path).unwrap_err();
+        let error = super::load_path(&path, &ferese_theme_client::service::fallback()).unwrap_err();
         assert!(error.contains("bar_layout"), "{error}");
         assert!(error.contains(path.to_str().unwrap()), "{error}");
         assert_eq!(std::fs::read_to_string(&path).unwrap(), source);
         std::fs::write(&path, "panel main {").unwrap();
-        assert!(super::load_path(&path).is_err());
+        assert!(super::load_path(&path, &ferese_theme_client::service::fallback()).is_err());
         let oversized = " ".repeat(ferese_config::file::LIMIT + 1);
         std::fs::write(&path, &oversized).unwrap();
-        assert!(super::load_path(&path).unwrap_err().contains("60 KiB"));
+        assert!(
+            super::load_path(&path, &ferese_theme_client::service::fallback())
+                .unwrap_err()
+                .contains("60 KiB")
+        );
         assert_eq!(std::fs::read_to_string(&path).unwrap(), oversized);
     }
 
     #[test]
     fn startup_distinguishes_absent_configuration_from_read_errors() {
         let directory = tempfile::tempdir().unwrap();
-        assert!(super::load_path(&directory.path().join("missing.kdl")).is_ok());
-        let error = super::load_path(directory.path()).unwrap_err();
+        assert!(
+            super::load_path(
+                &directory.path().join("missing.kdl"),
+                &ferese_theme_client::service::fallback()
+            )
+            .is_ok()
+        );
+        let error = super::load_path(directory.path(), &ferese_theme_client::service::fallback()).unwrap_err();
         assert!(error.contains("cannot read"), "{error}");
         let path = directory.path().join("config.kdl");
         std::os::unix::fs::symlink(directory.path().join("missing-target"), &path).unwrap();
         assert!(
-            super::load_path(&path).is_err(),
+            super::load_path(&path, &ferese_theme_client::service::fallback()).is_err(),
             "a broken config symlink is a read error"
         );
     }

shell/ferese-shell/src/ipc_subscription.rs

@@ -0,0 +1,186 @@
+use std::sync::Arc;
+use std::time::Duration;
+
+use cosmic::iced::futures::{SinkExt, Stream};
+use serde::de::DeserializeOwned;
+use serde_json::json;
+
+/// Follow a versioned snapshot, clearing stale content when its connection ends.
+pub(crate) fn stream<S>(get: &'static str, watch: &'static str, revision: fn(&S) -> u64) -> impl Stream<Item = Arc<S>>
+where
+    S: DeserializeOwned + Default + Send + Sync + 'static,
+{
+    cosmic::iced::stream::channel(1, async move |mut output| {
+        loop {
+            let connection = tokio::task::spawn_blocking(ferese_ipc::Connection::connect).await;
+            let Ok(Ok(mut connection)) = connection else {
+                tokio::time::sleep(Duration::from_secs(5)).await;
+                continue;
+            };
+            let Ok(cancellation) = connection.cancellation() else {
+                return;
+            };
+            let (send, mut receive) = tokio::sync::watch::channel(None);
+
+            tokio::task::spawn_blocking(move || {
+                let Ok(value) = connection.call(get, json!({})) else {
+                    return;
+                };
+                let Ok(mut snapshot) = serde_json::from_value::<S>(value) else {
+                    return;
+                };
+
+                loop {
+                    let since = revision(&snapshot);
+
+                    if send.send(Some(Arc::new(snapshot))).is_err() {
+                        return;
+                    }
+
+                    match connection
+                        .wait(watch, json!({"since": since}))
+                        .and_then(|value| serde_json::from_value(value).map_err(|e| e.to_string()))
+                    {
+                        Ok(next) => snapshot = next,
+                        Err(_) => return,
+                    }
+                }
+            });
+
+            while receive.changed().await.is_ok() {
+                let snapshot = receive.borrow_and_update().clone();
+                if let Some(snapshot) = snapshot
+                    && output.send(snapshot).await.is_err()
+                {
+                    return;
+                }
+            }
+
+            drop(cancellation);
+
+            if output.send(Arc::new(S::default())).await.is_err() {
+                return;
+            }
+
+            tokio::time::sleep(Duration::from_secs(5)).await;
+        }
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use cosmic::iced::futures::StreamExt;
+    use ferese_ipc::{Request, Response, read_frame, write_frame};
+    use std::io::Read;
+    use std::os::unix::net::UnixListener;
+
+    #[test]
+    fn media_and_activities_reconnect_clear_stale_state_and_cancel_watches() {
+        const CHILD: &str = "FERESE_SNAPSHOT_SUBSCRIPTION_TEST";
+        if std::env::var_os(CHILD).is_none() {
+            let root = tempfile::tempdir().unwrap();
+            let output = std::process::Command::new(std::env::current_exe().unwrap())
+                .args([
+                    "--exact",
+                    "ipc_subscription::tests::media_and_activities_reconnect_clear_stale_state_and_cancel_watches",
+                    "--nocapture",
+                ])
+                .env(CHILD, "1")
+                .env("XDG_RUNTIME_DIR", root.path())
+                .env_remove("FERESE_SOCKET")
+                .output()
+                .unwrap();
+            assert!(
+                output.status.success(),
+                "{}\n{}",
+                String::from_utf8_lossy(&output.stdout),
+                String::from_utf8_lossy(&output.stderr)
+            );
+            return;
+        }
+        check_stream::<ferese_ipc::media::Snapshot>("media-get", "media-watch", |snapshot| snapshot.revision);
+        check_stream::<ferese_ipc::activities::Snapshot>("activity-get", "activity-watch", |snapshot| {
+            snapshot.revision
+        });
+    }
+
+    fn check_stream<S>(get: &'static str, watch: &'static str, revision: fn(&S) -> u64)
+    where
+        S: DeserializeOwned + serde::Serialize + Default + Send + Sync + 'static,
+    {
+        let path = ferese_ipc::socket_path().unwrap();
+        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
+        let listener = UnixListener::bind(&path).unwrap();
+        listener.set_nonblocking(true).unwrap();
+        let (consumed, consumed_receive) = std::sync::mpsc::channel();
+        let (watch_send, watch_receive) = std::sync::mpsc::channel();
+        let server = std::thread::spawn(move || {
+            for initial in [1, 3] {
+                let deadline = std::time::Instant::now() + Duration::from_secs(10);
+                let mut connection = loop {
+                    match listener.accept() {
+                        Ok((connection, _)) => break connection,
+                        Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
+                            assert!(std::time::Instant::now() < deadline, "subscription did not reconnect");
+                            std::thread::sleep(Duration::from_millis(5));
+                        }
+                        Err(error) => panic!("accept: {error}"),
+                    }
+                };
+                connection.set_read_timeout(Some(Duration::from_secs(8))).unwrap();
+                connection.set_write_timeout(Some(Duration::from_secs(8))).unwrap();
+                let request: Request = read_frame(&mut connection).unwrap();
+                assert_eq!(request.command, get);
+                let mut snapshot = serde_json::to_value(S::default()).unwrap();
+                snapshot["revision"] = json!(initial);
+                write_frame(&mut connection, &Response::success(request.id, snapshot.clone())).unwrap();
+                let request: Request = read_frame(&mut connection).unwrap();
+                assert_eq!(request.command, watch);
+                assert_eq!(request.args["since"], initial);
+                if initial == 1 {
+                    // Wait for delivery so watch-channel coalescing cannot skip this snapshot.
+                    assert_eq!(consumed_receive.recv_timeout(Duration::from_secs(8)).unwrap(), 1);
+                    snapshot["revision"] = json!(2);
+                    write_frame(&mut connection, &Response::success(request.id, snapshot)).unwrap();
+                    let next: Request = read_frame(&mut connection).unwrap();
+                    assert_eq!(next.command, watch);
+                    assert_eq!(next.args["since"], 2);
+                    assert_ne!(request.id, next.id);
+                    assert_eq!(consumed_receive.recv_timeout(Duration::from_secs(8)).unwrap(), 2);
+                    // Dropping this connection must clear the old snapshot before retrying.
+                } else {
+                    watch_send.send(()).unwrap();
+                    assert_eq!(
+                        connection.read(&mut [0]).unwrap(),
+                        0,
+                        "dropped subscription kept its socket alive"
+                    );
+                }
+            }
+        });
+        tokio::runtime::Builder::new_current_thread()
+            .enable_time()
+            .build()
+            .unwrap()
+            .block_on(async {
+                let mut subscription = Box::pin(stream::<S>(get, watch, revision));
+                for expected in [1, 2, 0, 3] {
+                    let snapshot = tokio::time::timeout(Duration::from_secs(8), subscription.next())
+                        .await
+                        .unwrap()
+                        .unwrap();
+                    assert_eq!(revision(&snapshot), expected);
+                    if expected == 1 || expected == 2 {
+                        consumed.send(expected).unwrap();
+                    }
+                }
+                tokio::task::spawn_blocking(move || watch_receive.recv_timeout(Duration::from_secs(2)).unwrap())
+                    .await
+                    .unwrap();
+                drop(subscription);
+            });
+        server.join().unwrap();
+        std::fs::remove_file(path).unwrap();
+    }
+}

shell/ferese-shell/src/main.rs

@@ -10,6 +10,7 @@ mod compositor_ipc;
 mod config;
 mod control;
 mod display_mode;
+mod ipc_subscription;
 mod keybinding_guide;
 mod media;
 mod motion;
@@ -101,7 +102,8 @@ impl From<panel::PanelGeometry> for BarMetrics {
 }
 
 fn main() -> cosmic::iced::Result {
-    let (mut config, startup_error) = match config::load() {
+    let theme_snapshot = ferese_theme_client::service::current();
+    let (mut config, startup_error) = match config::load(&theme_snapshot) {
         Ok(config) => (config, None),
         Err(error) => {
             eprintln!("ferese-shell: {error}; starting with defaults");
@@ -140,7 +142,7 @@ fn main() -> cosmic::iced::Result {
         .default_font(shell_font)
         .theme(config.theme.palette().native_theme());
 
-    cosmic::app::run::<FereseShell>(settings, (config, wallpaper, startup_error))
+    cosmic::app::run::<FereseShell>(settings, (config, wallpaper, startup_error, theme_snapshot))
 }
 
 type WallpaperLoad = cosmic::iced::futures::channel::oneshot::Receiver<Result<image::Handle, String>>;
@@ -148,6 +150,7 @@ type WallpaperLoad = cosmic::iced::futures::channel::oneshot::Receiver<Result<im
 struct FereseShell {
     core: Core,
     config: ShellConfig,
+    theme_snapshot: ferese_ipc::theme::Snapshot,
     render: rendering::Rendering,
     wallpaper: Option<image::Handle>,
     control: Option<ShellControl>,
@@ -285,7 +288,12 @@ enum Message {
 
 impl cosmic::Application for FereseShell {
     type Executor = cosmic::executor::Default;
-    type Flags = (ShellConfig, Option<WallpaperLoad>, Option<String>);
+    type Flags = (
+        ShellConfig,
+        Option<WallpaperLoad>,
+        Option<String>,
+        ferese_ipc::theme::Snapshot,
+    );
     type Message = Message;
 
     const APP_ID: &'static str = APP_ID;
@@ -298,7 +306,10 @@ impl cosmic::Application for FereseShell {
         &mut self.core
     }
 
-    fn init(core: Core, (config, wallpaper, startup_error): Self::Flags) -> (Self, Task<Self::Message>) {
+    fn init(
+        core: Core,
+        (config, wallpaper, startup_error, theme_snapshot): Self::Flags,
+    ) -> (Self, Task<Self::Message>) {
         let control = ShellControl::connect()
             .map_err(|error| eprintln!("ferese-shell: shell control unavailable: {error}"))
             .ok();
@@ -309,6 +320,7 @@ impl cosmic::Application for FereseShell {
         notification_center.activity_bridge = config.activities.sources.notification_bridge;
         let app = Self {
             core,
+            theme_snapshot,
             notifications: notification_center,
             notification_surface: None,
             status_service: status::Service::start(config.status.settings_command.clone()),
@@ -511,8 +523,9 @@ impl cosmic::Application for FereseShell {
                 config.theme_mode = snapshot.mode;
                 if snapshot.error != self.theme_error {
                     self.pending_config_error = snapshot.error.clone();
-                    self.theme_error = snapshot.error;
+                    self.theme_error = snapshot.error.clone();
                 }
+                self.theme_snapshot = *snapshot;
                 self.apply_config(config)
             }
             Message::ThemeMode(mode) => cosmic::task::future(async move {
@@ -959,7 +972,7 @@ impl FereseShell {
     }
 
     fn reload_config(&mut self, source: String) -> Task<Message> {
-        let config = match config::parse_source(&source) {
+        let config = match config::parse_source(&source, &self.theme_snapshot) {
             Ok(config) => config,
             Err(error) => {
                 eprintln!("ferese-shell: reload rejected; keeping current config: {error}");
@@ -1137,6 +1150,7 @@ mod tests {
         let output = wl_output::WlOutput::inert(connection.backend().downgrade());
         FereseShell {
             core: Core::default(),
+            theme_snapshot: ferese_theme_client::service::fallback(),
             notifications: notifications::Center::new(config.notifications.clone(), config.animations),
             notification_surface: None,
             status_service: status::Service::start(None),
@@ -1196,6 +1210,29 @@ mod tests {
         }
     }
 
+    #[test]
+    fn reload_preserves_the_last_presented_theme_without_a_control_connection() {
+        use cosmic::Application;
+
+        let mut shell = shell_with_measured_panel();
+        let mut snapshot = ferese_theme_client::service::fallback();
+        snapshot.mode = ferese_config::theme::Mode::Light;
+        snapshot.presented.appearance = ferese_config::theme::Appearance::Light;
+        snapshot.presented.tokens.geometry.shell_radius = 23.;
+        snapshot.presented.tokens.typography.font_family = "Reload test font".into();
+        drop(shell.update(Message::ThemeChanged(Box::new(snapshot.clone()))));
+
+        drop(shell.reload_config("status { low-battery-threshold 35; }".into()));
+
+        assert!(shell.control.is_none());
+        assert_eq!(shell.config.status.low_battery_threshold, 35);
+        assert_eq!(shell.config.theme_mode, snapshot.mode);
+        assert_eq!(shell.config.theme.appearance, snapshot.presented.appearance);
+        assert_eq!(shell.config.theme.material_radius, 23.);
+        assert_eq!(shell.config.font_family.as_deref(), Some("Reload test font"));
+        assert_eq!(shell.theme_snapshot, snapshot);
+    }
+
     #[test]
     fn cosmetic_reloads_preserve_measured_margins_and_the_open_popover() {
         let mut shell = shell_with_measured_panel();
@@ -1230,7 +1267,7 @@ mod tests {
                 .unwrap();
             document.set(&format!("panels.0.{field}"), value).unwrap();
             let source = document.to_string();
-            let mut config = config::parse_source(&source).unwrap();
+            let mut config = config::parse_source(&source, &shell.theme_snapshot).unwrap();
             config.theme = shell.config.theme;
             drop(shell.apply_config(config));
             assert_eq!(shell.outputs[0].panel_resolution, resolution, "{edit}");

shell/ferese-shell/src/media/subscription.rs

@@ -1,147 +1,12 @@
 use std::sync::Arc;
-use std::time::Duration;
 
 use cosmic::iced::Subscription;
-use cosmic::iced::futures::SinkExt;
 use ferese_ipc::media::Snapshot;
-use serde_json::json;
 
 pub(super) fn subscription() -> Subscription<Arc<Snapshot>> {
     Subscription::run(stream)
 }
 
 fn stream() -> impl cosmic::iced::futures::Stream<Item = Arc<Snapshot>> {
-    cosmic::iced::stream::channel(1, async |mut output| {
-        loop {
-            let connection = tokio::task::spawn_blocking(ferese_ipc::Connection::connect).await;
-            let Ok(Ok(mut connection)) = connection else {
-                tokio::time::sleep(Duration::from_secs(5)).await;
-                continue;
-            };
-            let Ok(cancellation) = connection.cancellation() else {
-                return;
-            };
-            let (send, mut receive) = tokio::sync::watch::channel(None);
-            tokio::task::spawn_blocking(move || {
-                let Ok(value) = connection.call("media-get", json!({})) else {
-                    return;
-                };
-                let Ok(mut snapshot) = serde_json::from_value::<Snapshot>(value) else {
-                    return;
-                };
-                loop {
-                    let revision = snapshot.revision;
-                    if send.send(Some(Arc::new(snapshot))).is_err() {
-                        return;
-                    }
-                    match connection
-                        .wait("media-watch", json!({"since":revision}))
-                        .and_then(|value| serde_json::from_value(value).map_err(|e| e.to_string()))
-                    {
-                        Ok(next) => snapshot = next,
-                        Err(_) => return,
-                    }
-                }
-            });
-            while receive.changed().await.is_ok() {
-                let snapshot = receive.borrow_and_update().clone();
-                if let Some(snapshot) = snapshot
-                    && output.send(snapshot).await.is_err()
-                {
-                    return;
-                }
-            }
-
-            drop(cancellation);
-            if output.send(Arc::new(Snapshot::default())).await.is_err() {
-                return;
-            }
-            tokio::time::sleep(Duration::from_secs(5)).await;
-        }
-    })
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use cosmic::iced::futures::StreamExt;
-    use ferese_ipc::{Request, Response, read_frame, write_frame};
-    use std::os::unix::net::UnixListener;
-
-    #[test]
-    fn reconnect_clears_ghost_media_and_drop_cancels_a_watch() {
-        const CHILD: &str = "FERESE_MEDIA_SUBSCRIPTION_TEST";
-        if std::env::var_os(CHILD).is_none() {
-            let root = tempfile::tempdir().unwrap();
-            let output = std::process::Command::new(std::env::current_exe().unwrap())
-                .args([
-                    "--exact",
-                    "media::subscription::tests::reconnect_clears_ghost_media_and_drop_cancels_a_watch",
-                    "--nocapture",
-                ])
-                .env(CHILD, "1")
-                .env("XDG_RUNTIME_DIR", root.path())
-                .env_remove("FERESE_SOCKET")
-                .output()
-                .unwrap();
-            assert!(
-                output.status.success(),
-                "{}\n{}",
-                String::from_utf8_lossy(&output.stdout),
-                String::from_utf8_lossy(&output.stderr)
-            );
-            return;
-        }
-
-        let path = ferese_ipc::socket_path().unwrap();
-        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
-        let listener = UnixListener::bind(path).unwrap();
-        let (watch_send, watch_receive) = std::sync::mpsc::channel();
-        let server = std::thread::spawn(move || {
-            for revision in [1, 2] {
-                let (mut connection, _) = listener.accept().unwrap();
-                connection.set_read_timeout(Some(Duration::from_secs(8))).unwrap();
-                let get: Request = read_frame(&mut connection).unwrap();
-                assert_eq!(get.command, "media-get");
-                let snapshot = Snapshot {
-                    revision,
-                    ..Default::default()
-                };
-                write_frame(
-                    &mut connection,
-                    &Response::success(get.id, serde_json::to_value(snapshot).unwrap()),
-                )
-                .unwrap();
-                let watch: Request = read_frame(&mut connection).unwrap();
-                assert_eq!(watch.command, "media-watch");
-                assert_eq!(watch.args["since"], revision);
-                assert_ne!(get.id, watch.id);
-                if revision == 2 {
-                    watch_send.send(()).unwrap();
-                    use std::io::Read;
-                    assert_eq!(connection.read(&mut [0]).unwrap(), 0);
-                }
-            }
-        });
-        tokio::runtime::Builder::new_current_thread()
-            .enable_time()
-            .build()
-            .unwrap()
-            .block_on(async {
-                let mut subscription = Box::pin(stream());
-                for revision in [1, 0, 2] {
-                    let snapshot = tokio::time::timeout(Duration::from_secs(8), subscription.next())
-                        .await
-                        .unwrap()
-                        .unwrap();
-                    assert_eq!(snapshot.revision, revision);
-                }
-
-                tokio::task::spawn_blocking(move || watch_receive.recv_timeout(Duration::from_secs(2)).unwrap())
-                    .await
-                    .unwrap();
-                drop(subscription);
-            });
-        server.join().unwrap();
-    }
+    crate::ipc_subscription::stream("media-get", "media-watch", |snapshot: &Snapshot| snapshot.revision)
 }