refactors and more attempts to fix pipe closing issue

notification_provider/src/main.rs

@@ -34,12 +34,13 @@ fn execute_command(command: NotificationCommand) -> Result<()> {
 }
 
 fn read_notification_command(input: &mut Pipe<Client, Read>) -> Result<NotificationCommand> {
-    LOGGER.debug(format!("notification_provider: got input"));
+    LOGGER.debug(format!("Waiting for input..."));
     let object = input.read_to()?;
+    LOGGER.debug(format!("Input object:\n{}", object));
     Ok(object)
 }
 
-fn wait_for_input() -> Result<()> {
+fn input_loop() -> Result<()> {
     let mut input = Pipe::create_client_retrying(PIPE_NAME)
         .map_err(|e| format!("Could not create client pipe!\n{}", e))?;
     LOGGER.debug("notification_provider: waiting for input...");
@@ -61,7 +62,7 @@ fn run() -> Result<()> {
             .ok()
             .map_err(|e| format!("Could not init COM!\n{}", e))?
     };
-    wait_for_input().map_err(|e| format!("Error occured while waiting for input!\n{}", e))?;
+    input_loop().map_err(|e| format!("Error occured while waiting for input!\n{}", e))?;
     Ok(())
 }
 

shared/src/logging.rs

@@ -33,10 +33,16 @@ impl Logger {
         self.log(input, LogLevel::Error);
     }
 
+    #[cfg(debug_assertions)]
     pub fn debug<A: Display>(&self, input: A) {
         self.log(input, LogLevel::Debug);
     }
 
+    #[cfg(not(debug_assertions))]
+    pub fn debug<A: Display>(&self, input: A) {
+        drop(input)
+    }
+
     pub fn log<A: Display>(&self, input: A, level: LogLevel) {
         let log_path = self.log_path.get_or_init(|| {
             let mut log_root = PathBuf::from_str(TEMP_PATH).expect("Could not get debug path!");
@@ -63,8 +69,9 @@ impl Logger {
 
         let time_now = time::OffsetDateTime::now_utc();
         let mut msg = format!(
-            "[{} | {}] {}",
+            "[{} | ({} - {})] {}",
             time::PrimitiveDateTime::new(time_now.date(), time_now.time()),
+            self.process_name,
             self.source_name,
             input
         );

shared/src/notification_command.rs

@@ -1,7 +1,15 @@
+use std::fmt::Display;
+
 use serde::{Deserialize, Serialize};
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Debug)]
 pub struct NotificationCommand {
     pub name: String,
     pub content: String,
 }
+
+impl Display for NotificationCommand {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_fmt(format_args!("{}\n{}", &self.name, &self.content))
+    }
+}

shared/src/pipe/client.rs

@@ -13,7 +13,7 @@ use windows::Win32::{
 };
 
 const RETRYING_DELAY: u32 = 1000;
-const RETRYING_ATTEMPTS: u32 = 30;
+const RETRYING_ATTEMPTS: u32 = 15;
 
 pub struct Client;
 

shared/src/pipe/mod.rs

@@ -9,10 +9,14 @@ use crate::{
     logging::Logger,
     stream::{HandleStream, HandleStreamMode},
 };
-use std::io::{Read, Write};
+use std::{
+    fmt::Debug,
+    io::{Read, Write},
+};
 
 pub type Result<T> = std::result::Result<T, Box<dyn std::error::Error>>;
 
+pub const PIPE_BUFFER_SIZE: usize = 1024;
 pub const PIPE_PATH_ROOT: &str = "\\\\.\\pipe\\";
 pub const PIPE_NAME: &str = "AutoPowerNotificationPipe";
 
@@ -24,11 +28,12 @@ pub struct Pipe<M, S: HandleStreamMode> {
 }
 
 impl<M> Pipe<M, stream::Read> {
-    pub fn read_to<T: serde::de::DeserializeOwned>(&mut self) -> Result<T> {
-        let mut buf = Vec::with_capacity(1024);
-        self.stream.read_to_end(&mut buf)?;
+    pub fn read_to<T: serde::de::DeserializeOwned + Debug>(&mut self) -> Result<T> {
+        let mut buf = [0; PIPE_BUFFER_SIZE];
         let count = self.read(&mut buf)?;
+        LOGGER.debug(format!("Got {} bytes. Deserializing...", count));
         let obj = bincode::deserialize(&mut buf[..count])?;
+        LOGGER.debug(format!("Deserialized to {:?}", obj));
         Ok(obj)
     }
 }
@@ -49,6 +54,7 @@ impl<M> Pipe<M, stream::Write> {
 
 impl<M> std::io::Write for Pipe<M, stream::Write> {
     fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+        LOGGER.debug("Writing to pipe...");
         self.stream.write(buf)
     }
 
@@ -67,3 +73,10 @@ impl<M, S: HandleStreamMode> Pipe<M, S> {
         self.stream.close()
     }
 }
+
+impl<M, S: HandleStreamMode> Drop for Pipe<M, S> {
+    fn drop(&mut self) {
+        LOGGER.debug("Dropping pipe...");
+        self.close().unwrap();
+    }
+}

shared/src/pipe/server.rs

@@ -1,4 +1,4 @@
-use super::{Pipe, Result, PIPE_PATH_ROOT};
+use super::{Pipe, Result, PIPE_BUFFER_SIZE, PIPE_PATH_ROOT};
 use crate::{
     stream::{HandleStream, HandleStreamMode},
     util::get_last_win32_err,
@@ -16,8 +16,6 @@ use windows::Win32::{
     },
 };
 
-const PIPE_BUFFER_SIZE: u32 = 1024;
-
 pub struct Server;
 
 impl<S: HandleStreamMode> Pipe<Server, S> {
@@ -49,8 +47,8 @@ impl<S: HandleStreamMode> Pipe<Server, S> {
                 S::as_pipe_access_rights() | FILE_FLAG_FIRST_PIPE_INSTANCE,
                 PIPE_TYPE_MESSAGE | PIPE_READMODE_MESSAGE,
                 1,
-                PIPE_BUFFER_SIZE,
-                PIPE_BUFFER_SIZE,
+                PIPE_BUFFER_SIZE as u32,
+                PIPE_BUFFER_SIZE as u32,
                 0,
                 Some(&security),
             )

shared/src/stream/mod.rs

@@ -38,3 +38,9 @@ impl<M: HandleStreamMode> HandleStream<M> {
         Ok(())
     }
 }
+
+impl<M: HandleStreamMode> Drop for HandleStream<M> {
+    fn drop(&mut self) {
+        self.close().unwrap();
+    }
+}

shared/src/stream/reader.rs

@@ -1,9 +1,13 @@
+use crate::logging::Logger;
+
 use super::{HandleStream, HandleStreamMode};
 use windows::Win32::{
     Foundation::GENERIC_READ,
     Storage::FileSystem::{ReadFile, PIPE_ACCESS_INBOUND},
 };
 
+const LOGGER: Logger = Logger::new("stream_reader", "autopower_shared");
+
 pub struct Read;
 impl HandleStreamMode for Read {
     fn as_generic_access_rights() -> u32 {
@@ -20,6 +24,7 @@ impl HandleStream<Read> {}
 impl std::io::Read for HandleStream<Read> {
     fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
         let mut bytes_read = 0;
+        LOGGER.debug("Reading from file handle... (blocking)");
         unsafe { ReadFile(self.handle, Some(buf), Some(&mut bytes_read), None)? };
         Ok(bytes_read as usize)
     }

src/notification_provider.rs

@@ -24,16 +24,19 @@ impl NotificationProvider {
 
     pub fn send_display_command(&mut self, title: &str, description: &str) -> Result<()> {
         LOGGER.debug(format!("Sent command:\n{} | {}", title, description));
-        let command = NotificationCommand {
+        self.pipe.write_as(NotificationCommand {
             name: "display".to_owned(),
             content: format!("{}\n{}", title, description),
-        };
-        self.pipe.write_as(&command)?;
+        })?;
         Ok(())
     }
 
-    pub fn terminate(&self) -> Result<()> {
+    pub fn terminate(&mut self) -> Result<()> {
         LOGGER.debug("Terminating notification provider...");
+        self.pipe.write_as(NotificationCommand {
+            name: "terminate".to_owned(),
+            content: "".to_owned(),
+        })?;
         self.pipe.close()
     }
 }

src/services/power_service.rs

@@ -4,7 +4,7 @@ use crate::{
     power_scheme::set_power_scheme,
 };
 use autopower_shared::{logging::Logger, winstr::to_win32_wstr};
-use std::ffi::c_void;
+use std::{ffi::c_void, mem::ManuallyDrop};
 use windows::{
     core::PWSTR,
     Win32::{
@@ -149,7 +149,7 @@ impl PowerService {
         event_data: *mut c_void,
         _context: *mut c_void,
     ) -> u32 {
-        let mut me = _context.cast::<Self>().read();
+        let mut me = ManuallyDrop::new(_context.cast::<Self>().read());
         let data = HandlerData {
             event_type,
             event_data,
@@ -251,7 +251,7 @@ impl WindowsService for PowerService {
             LOGGER.error(format!("Could not set service status!\n{}", e));
         }
         me.notification_provider
-            .as_ref()
+            .as_mut()
             .unwrap()
             .terminate()
             .unwrap();