Simplify the event stream

Cargo.toml

@@ -11,13 +11,14 @@ anyhow = "1.0.56"
 tracing = "0.1.32"
 tracing-subscriber = { version = "0.3.9", features = ["env-filter"] }
 netlink-sys = { version = "0.8.5", features = ["tokio_socket"] }
-futures-util = "0.3.21"
 tokio = { version = "1.17.0", features = ["macros", "rt-multi-thread", "sync", "fs"] }
 fork = "0.1.19"
 walkdir = "2.3.2"
 clap = { version = "4.5.1", features = ["derive", "wrap_help"] }
 nix = { version = "0.27.1", features = ["user", "fs"] }
 bytes = "1.5.0"
+async-stream = "0.3.6"
+futures-util = "0.3.31"
 
 [dev-dependencies]
 regex = "1.5.5"

src/bin/mdev.rs

@@ -180,23 +180,26 @@ impl Opt {
             mdev::stream::uevents()?
                 .for_each(|ev| async {
                     info!("event {:?}", ev);
-                    let result = ev.map(|ev| async {
-                        let result =
-                            react_to_event(&ev.devpath, &ev.env, ev.action, conf, &self.devpath)
-                                .await;
-                        if let Some(rebroadcast_sender) = &rebroadcast_sender {
-                            if rebroadcast_sender
-                                .send(RebroadcastMessage::Event(ev))
-                                .await
-                                .is_err()
+
+                    match ev {
+                        Ok(ev) => {
+                            if let Err(e) =
+                                react_to_event(&ev.devpath, &ev.env, ev.action, conf, &self.devpath)
+                                    .await
                             {
-                                panic!("rebroadcaster channel is closed");
+                                warn!("{e}");
+                            }
+                            if let Some(rebroadcast_sender) = &rebroadcast_sender {
+                                if rebroadcast_sender
+                                    .send(RebroadcastMessage::Event(ev))
+                                    .await
+                                    .is_err()
+                                {
+                                    warn!("rebroadcaster channel is closed");
+                                }
                             }
                         }
-                        result
-                    });
-                    if let Err(e) = result {
-                        warn!("{}", e);
+                        Err(e) => warn!("{}", e),
                     }
                 })
                 .await;

src/stream.rs

@@ -1,11 +1,9 @@
 use std::process;
 
 use anyhow::anyhow;
-
-use futures_util::stream::{unfold, Stream};
-
+use async_stream::try_stream;
+use futures_util::Stream;
 use kobject_uevent::UEvent;
-
 use netlink_sys::{
     protocols::NETLINK_KOBJECT_UEVENT, AsyncSocket, AsyncSocketExt, SocketAddr, TokioSocket,
 };
@@ -20,22 +18,10 @@ pub fn uevents() -> anyhow::Result<impl Stream<Item = anyhow::Result<UEvent>>> {
         .bind(&sa)
         .map_err(|e| anyhow!("Socket bind error: {}", e))?;
 
-    Ok(unfold(
-        (socket, bytes::BytesMut::with_capacity(1024 * 8)),
-        |(mut socket, mut buf)| async move {
-            buf.clear();
-            match socket.recv_from(&mut buf).await {
-                Ok(_addr) => {
-                    if buf.len() == 0 {
-                        return None;
-                    }
-                }
-                Err(e) => {
-                    return Some((Err(anyhow!("Socket receive error: {}", e)), (socket, buf)));
-                }
-            };
-
-            Some((UEvent::from_netlink_packet(&buf), (socket, buf)))
-        },
-    ))
+    Ok(try_stream! {
+        loop {
+            let (buf, _sock) = socket.recv_from_full().await?;
+            yield UEvent::from_netlink_packet(&buf)?;
+        }
+    })
 }