swap async-stream in dev-deps with asynk-strim

GlenDC · GlenDC · commit 1ab752347a17 · 2025-09-06T09:13:39.000+02:00
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -43,7 +43,7 @@ serde_json = { version = "1", default-features = false, optional = true, feature
 tracing = { version = "0.1.41", optional = true }
 
 [dev-dependencies]
-async-stream = { version = "0.3.6", default-features = false }
+asynk-strim = { version = "0.1" }
 axum = { version = "0.8.4" }
 indexmap = { version = "2.11", features = ["serde"] }
 reqwest = { version = "0.12.23", features = ["json", "stream"] }
diff --git a/examples/axum-activity-feed.rs b/examples/axum-activity-feed.rs
@@ -1,5 +1,5 @@
 use {
-    async_stream::stream,
+    asynk_strim::stream_fn,
     axum::{
         Router,
         extract::Path,
@@ -80,34 +80,36 @@ async fn generate(ReadSignals(signals): ReadSignals<Signals>) -> impl IntoRespon
     let mut done = signals.done;
 
     // Start the SSE stream
-    Sse::new(stream! {
+    Sse::new(stream_fn(move |mut yielder| async move {
         // Signal event generation start
         let patch = PatchSignals::new(r#"{{"generating": true}}"#);
         let sse_event = patch.write_as_axum_sse_event();
-        yield Ok::<_, Infallible>(sse_event);
+        yielder.yield_item(Ok::<_, Infallible>(sse_event)).await;
 
         // Yield the events elements and signals to the stream
         for _ in 1..=signals.events {
             total += 1;
             done += 1;
             // Append a new entry to the activity feed
             let elements = event_entry(&Status::Done, total, "Auto");
-            let patch = PatchElements::new(elements).selector("#feed").mode(ElementPatchMode::After);
+            let patch = PatchElements::new(elements)
+                .selector("#feed")
+                .mode(ElementPatchMode::After);
             let sse_event = patch.write_as_axum_sse_event();
-            yield Ok::<_, Infallible>(sse_event);
+            yielder.yield_item(Ok::<_, Infallible>(sse_event)).await;
 
             // Update the event counts
             let patch = PatchSignals::new(format!(r#"{{"total": {total}, "done": {done}}}"#));
             let sse_event = patch.write_as_axum_sse_event();
-            yield Ok::<_, Infallible>(sse_event);
+            yielder.yield_item(Ok::<_, Infallible>(sse_event)).await;
             tokio::time::sleep(Duration::from_millis(signals.interval)).await;
         }
 
         // Signal event generation end
         let patch = PatchSignals::new(r#"{{"generating": false}}"#);
         let sse_event = patch.write_as_axum_sse_event();
-        yield Ok::<_, Infallible>(sse_event);
-    })
+        yielder.yield_item(Ok::<_, Infallible>(sse_event)).await;
+    }))
 }
 
 /// Creates one event with a given status
@@ -116,7 +118,7 @@ async fn event(
     ReadSignals(signals): ReadSignals<Signals>,
 ) -> impl IntoResponse {
     // Create the event stream, since we're patching both an element and a signal.
-    Sse::new(stream! {
+    Sse::new(stream_fn(move |mut yielder| async move {
         // Signal the updated event counts
         let total = signals.total + 1;
         let signals = match status {
@@ -127,14 +129,16 @@ async fn event(
         };
         let patch = PatchSignals::new(signals);
         let sse_signal = patch.write_as_axum_sse_event();
-        yield Ok::<_, Infallible>(sse_signal);
+        yielder.yield_item(Ok::<_, Infallible>(sse_signal)).await;
 
         // Patch an element and append it to the feed
         let elements = event_entry(&status, total, "Manual");
-        let patch = PatchElements::new(elements).selector("#feed").mode(ElementPatchMode::After);
+        let patch = PatchElements::new(elements)
+            .selector("#feed")
+            .mode(ElementPatchMode::After);
         let sse_event = patch.write_as_axum_sse_event();
-        yield Ok::<_, Infallible>(sse_event);
-    })
+        yielder.yield_item(Ok::<_, Infallible>(sse_event)).await;
+    }))
 }
 
 /// Returns an HTML string for the entry
diff --git a/examples/axum-hello.rs b/examples/axum-hello.rs
@@ -1,5 +1,5 @@
 use {
-    async_stream::stream,
+    asynk_strim::stream_fn,
     axum::{
         Router,
         response::{Html, IntoResponse, Sse},
@@ -49,15 +49,15 @@ pub struct Signals {
 }
 
 async fn hello_world(ReadSignals(signals): ReadSignals<Signals>) -> impl IntoResponse {
-    Sse::new(stream! {
+    Sse::new(stream_fn(move |mut yielder| async move {
         for i in 0..MESSAGE.len() {
             let elements = format!("<div id='message'>{}</div>", &MESSAGE[0..i + 1]);
             let patch = PatchElements::new(elements);
             let sse_event = patch.write_as_axum_sse_event();
 
-            yield Ok::<_, Infallible>(sse_event);
+            yielder.yield_item(Ok::<_, Infallible>(sse_event)).await;
 
             tokio::time::sleep(Duration::from_millis(signals.delay)).await;
         }
-    })
+    }))
 }
diff --git a/examples/axum-live-reload.rs b/examples/axum-live-reload.rs
@@ -1,5 +1,5 @@
 use {
-    async_stream::stream,
+    asynk_strim::stream_fn,
     axum::{
         Router,
         response::{Html, IntoResponse, Sse},
@@ -75,14 +75,14 @@ async fn hotreload() -> impl IntoResponse {
     use datastar::prelude::ExecuteScript;
     static ONCE: atomic::AtomicBool = atomic::AtomicBool::new(false);
 
-    Sse::new(stream! {
+    Sse::new(stream_fn(move |mut yielder| async move {
         if !ONCE.swap(true, atomic::Ordering::SeqCst) {
             let script = ExecuteScript::new("window.location.reload()");
             let sse_event = script.write_as_axum_sse_event();
-            yield Ok::<_, Infallible>(sse_event);
+            yielder.yield_item(Ok::<_, Infallible>(sse_event)).await;
         }
         std::future::pending().await
-    })
+    }))
 }
 
 const MESSAGE: &str = "Hello, world!";
@@ -93,15 +93,15 @@ pub struct Signals {
 }
 
 async fn hello_world(ReadSignals(signals): ReadSignals<Signals>) -> impl IntoResponse {
-    Sse::new(stream! {
+    Sse::new(stream_fn(move |mut yielder| async move {
         for i in 0..MESSAGE.len() {
             let elements = format!("<div id='message'>{}</div>", &MESSAGE[0..i + 1]);
             let patch = PatchElements::new(elements);
             let sse_event = patch.write_as_axum_sse_event();
 
-            yield Ok::<_, Infallible>(sse_event);
+            yielder.yield_item(Ok::<_, Infallible>(sse_event)).await;
 
             tokio::time::sleep(Duration::from_millis(signals.delay)).await;
         }
-    })
+    }))
 }
diff --git a/examples/axum-test-suite.rs b/examples/axum-test-suite.rs
@@ -1,5 +1,5 @@
 use {
-    async_stream::stream,
+    asynk_strim::stream_fn,
     axum::{
         Router,
         response::{IntoResponse, Sse},
@@ -87,22 +87,50 @@ pub enum TestCaseEvent {
 }
 
 async fn test(ReadSignals(test_case): ReadSignals<TestCase>) -> impl IntoResponse {
-    Sse::new(stream! {
+    Sse::new(stream_fn(move |mut yielder| async move {
         for event in test_case.events {
             let sse_event = match event {
-                TestCaseEvent::ExecuteScript { script, event_id, retry_duration, attributes, auto_remove } => {
-                    ExecuteScript {
-                        script,
-                        id: event_id,
-                        retry: Duration::from_millis(retry_duration.unwrap_or(consts::DEFAULT_SSE_RETRY_DURATION)),
-                        auto_remove,
-                        attributes: attributes.map(|attributes| {
-                            attributes.into_iter().map(|(key, value)| format!("{key}=\"{}\"", value.to_string().trim_matches('"'))).collect()
-                        }).unwrap_or_default(),
-                    }.into_datastar_event().write_as_axum_sse_event()
+                TestCaseEvent::ExecuteScript {
+                    script,
+                    event_id,
+                    retry_duration,
+                    attributes,
+                    auto_remove,
+                } => ExecuteScript {
+                    script,
+                    id: event_id,
+                    retry: Duration::from_millis(
+                        retry_duration.unwrap_or(consts::DEFAULT_SSE_RETRY_DURATION),
+                    ),
+                    auto_remove,
+                    attributes: attributes
+                        .map(|attributes| {
+                            attributes
+                                .into_iter()
+                                .map(|(key, value)| {
+                                    format!("{key}=\"{}\"", value.to_string().trim_matches('"'))
+                                })
+                                .collect()
+                        })
+                        .unwrap_or_default(),
                 }
-                TestCaseEvent::PatchElements { elements, event_id, retry_duration, mode, selector, use_view_transition } => {
-                    PatchElements { id: event_id, retry: Duration::from_millis(retry_duration.unwrap_or(consts::DEFAULT_SSE_RETRY_DURATION)), elements, selector, mode: match mode.as_deref().unwrap_or_default() {
+                .into_datastar_event()
+                .write_as_axum_sse_event(),
+                TestCaseEvent::PatchElements {
+                    elements,
+                    event_id,
+                    retry_duration,
+                    mode,
+                    selector,
+                    use_view_transition,
+                } => PatchElements {
+                    id: event_id,
+                    retry: Duration::from_millis(
+                        retry_duration.unwrap_or(consts::DEFAULT_SSE_RETRY_DURATION),
+                    ),
+                    elements,
+                    selector,
+                    mode: match mode.as_deref().unwrap_or_default() {
                         "outer" => consts::ElementPatchMode::Outer,
                         "inner" => consts::ElementPatchMode::Inner,
                         "remove" => consts::ElementPatchMode::Remove,
@@ -112,19 +140,34 @@ async fn test(ReadSignals(test_case): ReadSignals<TestCase>) -> impl IntoRespons
                         "before" => consts::ElementPatchMode::Before,
                         "after" => consts::ElementPatchMode::After,
                         _ => consts::ElementPatchMode::Outer,
-                    }, use_view_transition: use_view_transition.unwrap_or_default() }.into_datastar_event().write_as_axum_sse_event()
-                },
-                TestCaseEvent::PatchSignals { signals, signals_raw, event_id, retry_duration, only_if_missing } => {
-                    PatchSignals {
-                        id: event_id,
-                        retry: Duration::from_millis(retry_duration.unwrap_or(consts::DEFAULT_SSE_RETRY_DURATION)),
-                        signals: signals_raw.unwrap_or_else(|| signals.map(|s| serde_json::to_string(&s).unwrap_or_default()).unwrap_or_default()),
-                        only_if_missing: only_if_missing.unwrap_or_default(),
-                    }.into_datastar_event().write_as_axum_sse_event()
-                },
+                    },
+                    use_view_transition: use_view_transition.unwrap_or_default(),
+                }
+                .into_datastar_event()
+                .write_as_axum_sse_event(),
+                TestCaseEvent::PatchSignals {
+                    signals,
+                    signals_raw,
+                    event_id,
+                    retry_duration,
+                    only_if_missing,
+                } => PatchSignals {
+                    id: event_id,
+                    retry: Duration::from_millis(
+                        retry_duration.unwrap_or(consts::DEFAULT_SSE_RETRY_DURATION),
+                    ),
+                    signals: signals_raw.unwrap_or_else(|| {
+                        signals
+                            .map(|s| serde_json::to_string(&s).unwrap_or_default())
+                            .unwrap_or_default()
+                    }),
+                    only_if_missing: only_if_missing.unwrap_or_default(),
+                }
+                .into_datastar_event()
+                .write_as_axum_sse_event(),
             };
 
-            yield Ok::<_, Infallible>(sse_event);
+            yielder.yield_item(Ok::<_, Infallible>(sse_event)).await;
         }
-    })
+    }))
 }
