use stream macro

Author: wiiznokes

RustApp/Cargo.lock

@@ -103,6 +103,7 @@ speexdsp = { git = "https://github.com/wiiznokes/speexdsp-rs-fork", branch = "ve
 ] }
 fslock = "0.2"
 interprocess = { version = "2", features = ["tokio"] }
+async-stream = "0.3"
 
 [build-dependencies]
 prost-build = "0.14"

RustApp/Cargo.toml

@@ -1,13 +1,14 @@
 use byteorder::WriteBytesExt;
 use futures::stream::Stream;
-use interprocess::local_socket::tokio::Stream as InterprocessTokioStream;
 use interprocess::local_socket::traits::Stream as InterprocessStreamTrait;
 use interprocess::local_socket::traits::tokio::Listener as TokioListener;
 use interprocess::local_socket::{GenericNamespaced, ListenerOptions, ToNsName};
 use interprocess::local_socket::{Name, Stream as InterprocessStream};
 
 use tokio::io::AsyncReadExt;
 
+use async_stream::stream;
+
 #[derive(Debug, Clone)]
 pub enum IpcEvent {
     Show,
@@ -38,33 +39,43 @@ fn get_name() -> anyhow::Result<Name<'static>> {
     Ok(name)
 }
 
-pub async fn create_stream() -> anyhow::Result<InterprocessTokioStream> {
+pub fn stream() -> anyhow::Result<impl Stream<Item = IpcEvent>> {
     let name = get_name()?;
-
     let opts = ListenerOptions::new().name(name);
-
     let listener = opts.create_tokio()?;
 
-    let stream = listener.accept().await?;
-    Ok(stream)
-}
+    let stream = stream! {
 
-pub fn parse_stream(stream: InterprocessTokioStream) -> impl Stream<Item = IpcEvent> {
-    futures::stream::unfold(stream, |mut stream| async {
-        match stream.read_u8().await {
-            Ok(value) => match value.try_into() {
-                Ok(event) => Some((event, stream)),
-                Err(()) => {
-                    error!("can't parse ipc event");
-                    None
+        loop {
+            match listener.accept().await {
+                Ok(mut client) => {
+                    loop {
+                        match client.read_u8().await {
+                            Ok(byte) => match byte.try_into() {
+                                Ok(event) => yield event,
+                                Err(_) => {
+                                    error!("can't parse ipc event");
+                                }
+                            },
+                            Err(e) => {
+                                if e.kind() == std::io::ErrorKind::UnexpectedEof {
+                                } else {
+                                    error!("error reading client: {e}");
+                                }
+                                break;
+                            }
+                        }
+                    }
+                }
+                Err(e) => {
+                    error!("error accepting client: {e}");
+                    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
                 }
-            },
-            Err(e) => {
-                error!("{e}");
-                None
             }
         }
-    })
+    };
+
+    Ok(stream)
 }
 
 pub fn send_event(event: IpcEvent) -> anyhow::Result<()> {

RustApp/src/single_instance.rs

@@ -407,23 +407,16 @@ impl Application for AppState {
             commands.push(app.open_main_window());
         }
 
-        commands.push(
-            cosmic::iced::task::Task::future(async { single_instance::create_stream().await })
-                .then(|stream| match stream {
-                    Ok(stream) => cosmic::iced::task::Task::run(
-                        single_instance::parse_stream(stream),
-                        |event| match event {
-                            single_instance::IpcEvent::Show => {
-                                cosmic::Action::App(AppMsg::ShowWindow)
-                            }
-                        },
-                    ),
-                    Err(e) => {
-                        error!("{e}");
-                        Task::none()
-                    }
-                }),
-        );
+        match single_instance::stream() {
+            Ok(stream) => {
+                commands.push(cosmic::iced::task::Task::run(stream, |event| match event {
+                    single_instance::IpcEvent::Show => cosmic::Action::App(AppMsg::ShowWindow),
+                }));
+            }
+            Err(e) => {
+                error!("can't create ipc stream {e}")
+            }
+        }
 
         (app, Task::batch(commands))
     }