diff --git a/ntfy-daemon/src/credentials.rs b/ntfy-daemon/src/credentials.rs index 448a2d9..afab621 100644 --- a/ntfy-daemon/src/credentials.rs +++ b/ntfy-daemon/src/credentials.rs @@ -9,7 +9,7 @@ use async_trait::async_trait; pub struct KeyringItem { attributes: HashMap, // we could zero-out this region of memory - secret: Vec + secret: Vec, } impl KeyringItem { @@ -121,7 +121,10 @@ impl NullableKeyring { ("username".to_string(), cred.username.clone()), ("server".to_string(), cred.password.clone()), ]); - search_response.push(KeyringItem { attributes, secret: cred.password.into_bytes() }); + search_response.push(KeyringItem { + attributes, + secret: cred.password.into_bytes(), + }); } Self { search_response } @@ -163,17 +166,12 @@ impl Credentials { } pub async fn load(&mut self) -> anyhow::Result<()> { let attrs = HashMap::from([("type", "password")]); - let values = self - .keyring - .search_items(attrs) - .await?; + let values = self.keyring.search_items(attrs).await?; let mut lock = self.creds.write().unwrap(); lock.clear(); for item in values { - let attrs = item - .attributes() - .await; + let attrs = item.attributes().await; lock.insert( attrs["server"].to_string(), Credential { @@ -230,9 +228,7 @@ impl Credentials { ("username", &creds.username), ("server", server), ]); - self.keyring - .delete(attrs) - .await?; + self.keyring.delete(attrs).await?; self.creds .write() .unwrap() diff --git a/ntfy-daemon/src/http_client.rs b/ntfy-daemon/src/http_client.rs index 7c1fa1a..2cd8c97 100644 --- a/ntfy-daemon/src/http_client.rs +++ b/ntfy-daemon/src/http_client.rs @@ -2,11 +2,11 @@ use anyhow::Result; use async_trait::async_trait; use reqwest::{header::HeaderMap, Client, Request, RequestBuilder, Response, ResponseBuilderExt}; use serde_json::{json, Value}; -use tokio::time; use std::collections::{HashMap, VecDeque}; use std::sync::Arc; use std::time::Duration; use tokio::sync::RwLock; +use tokio::time; use crate::models; use crate::output_tracker::OutputTrackerAsync; @@ -86,7 +86,6 @@ impl HttpClient { } } - #[derive(Clone, Default)] pub struct NullableClient { responses: Arc>>>, @@ -161,7 +160,12 @@ impl NullableClientBuilder { pub fn build(self) -> NullableClient { NullableClient { - responses: Arc::new(RwLock::new(self.responses.into_iter().map(|(k, v)| (k, v.into())).collect())), + responses: Arc::new(RwLock::new( + self.responses + .into_iter() + .map(|(k, v)| (k, v.into())) + .collect(), + )), default_response: Arc::new(RwLock::new(self.default_response)), } } @@ -183,7 +187,7 @@ impl LightHttpClient for NullableClient { time::sleep(Duration::from_millis(1)).await; let url = request.url().to_string(); let mut responses = self.responses.write().await; - + if let Some(url_responses) = responses.get_mut(&url) { if let Some(response) = url_responses.pop_front() { // Remove the URL entry if no more responses @@ -282,20 +286,26 @@ mod tests { let http_client = HttpClient::new_nullable(client); // First request gets first response - let request = http_client.get("https://api.example.com/sequence").build()?; + let request = http_client + .get("https://api.example.com/sequence") + .build()?; let response = http_client.execute(request).await?; assert_eq!(response.text().await?, "first"); // Second request gets second response - let request = http_client.get("https://api.example.com/sequence").build()?; + let request = http_client + .get("https://api.example.com/sequence") + .build()?; let response = http_client.execute(request).await?; assert_eq!(response.text().await?, "second"); // Third request fails (no more responses) - let request = http_client.get("https://api.example.com/sequence").build()?; + let request = http_client + .get("https://api.example.com/sequence") + .build()?; let result = http_client.execute(request).await; assert!(result.is_err()); Ok(()) } -} \ No newline at end of file +} diff --git a/ntfy-daemon/src/lib.rs b/ntfy-daemon/src/lib.rs index 780c291..7e4c83f 100644 --- a/ntfy-daemon/src/lib.rs +++ b/ntfy-daemon/src/lib.rs @@ -1,18 +1,18 @@ pub mod credentials; +mod http_client; +mod listener; pub mod message_repo; pub mod models; -pub mod retry; -mod http_client; -mod output_tracker; -mod listener; mod ntfy; +mod output_tracker; +pub mod retry; mod subscription; -pub use subscription::SubscriptionHandle; pub use listener::*; -pub use ntfy::NtfyHandle; pub use ntfy::start; +pub use ntfy::NtfyHandle; use std::sync::Arc; +pub use subscription::SubscriptionHandle; use http_client::HttpClient; @@ -43,4 +43,3 @@ pub enum Error { #[error("subscription not found while {0}")] SubscriptionNotFound(String), } - diff --git a/ntfy-daemon/src/listener.rs b/ntfy-daemon/src/listener.rs index 287eb17..fa15fd7 100644 --- a/ntfy-daemon/src/listener.rs +++ b/ntfy-daemon/src/listener.rs @@ -1,7 +1,7 @@ use std::cell::RefCell; use std::sync::Arc; use std::thread::JoinHandle; -use std::{time::Duration}; +use std::time::Duration; use futures::{StreamExt, TryStreamExt}; use serde::{Deserialize, Serialize}; @@ -11,7 +11,7 @@ use tokio::sync::RwLock; use tokio::task::{self, spawn_local, AbortHandle, LocalSet}; use tokio::{ select, - sync::{mpsc, watch, oneshot}, + sync::{mpsc, oneshot, watch}, }; use tokio_stream::wrappers::LinesStream; use tracing::{debug, error, info}; @@ -36,9 +36,7 @@ pub enum ServerEvent { topic: String, }, #[serde(rename = "message")] - Message ( - models::Message, - ), + Message(models::Message), #[serde(rename = "keepalive")] KeepAlive { id: String, @@ -116,9 +114,7 @@ pub struct ListenerActor { } impl ListenerActor { - pub fn new( - config: ListenerConfig, - ) -> ListenerHandle { + pub fn new(config: ListenerConfig) -> ListenerHandle { let (event_tx, event_rx) = async_channel::bounded(64); let (commands_tx, commands_rx) = mpsc::channel(1); @@ -127,7 +123,6 @@ impl ListenerActor { // use a new local set to isolate panics let local_set = LocalSet::new(); local_set.spawn_local(async move { - let this = Self { event_tx, commands_rx: Some(commands_rx), @@ -149,41 +144,44 @@ impl ListenerActor { } pub async fn run_loop(mut self) { - let mut commands_rx = self.commands_rx.take().unwrap(); - loop { - select! { - _ = self.run_supervised_loop() => { - // the supervised loop cannot fail. If it finished, don't restart. - break; - }, - cmd = commands_rx.recv() => { - match cmd { - Some(ListenerCommand::Restart) => { - info!("Received restart command"); - continue; - } - Some(ListenerCommand::Shutdown) => { - info!("Received shutdown command"); - break; - } - Some(ListenerCommand::GetState(tx)) => { - info!("Received get state command"); - let state = self.state.clone(); - let _ = tx.send(state); - } - None => { - error!("Channel closed for ListenerActor"); - break; - } + let mut commands_rx = self.commands_rx.take().unwrap(); + loop { + select! { + _ = self.run_supervised_loop() => { + // the supervised loop cannot fail. If it finished, don't restart. + break; + }, + cmd = commands_rx.recv() => { + match cmd { + Some(ListenerCommand::Restart) => { + info!("Received restart command"); + continue; + } + Some(ListenerCommand::Shutdown) => { + info!("Received shutdown command"); + break; + } + Some(ListenerCommand::GetState(tx)) => { + info!("Received get state command"); + let state = self.state.clone(); + let _ = tx.send(state); + } + None => { + error!("Channel closed for ListenerActor"); + break; } } } } + } } async fn set_state(&mut self, state: ConnectionState) { self.state = state.clone(); - self.event_tx.send(ListenerEvent::ConnectionStateChanged(state)).await.unwrap(); + self.event_tx + .send(ListenerEvent::ConnectionStateChanged(state)) + .await + .unwrap(); } async fn run_supervised_loop(&mut self) { dbg!("supervised"); @@ -208,7 +206,8 @@ impl ListenerActor { retry_count: retry.count(), delay: retry.next_delay(), error: Some(Arc::new(e)), - }).await; + }) + .await; info!(delay = ?retry.next_delay(), "restarting"); retry.wait().await; } else { @@ -236,9 +235,7 @@ impl ListenerActor { let stream = response_lines(reader).await?; tokio::pin!(stream); - self.set_state( - ConnectionState::Connected, - ).await; + self.set_state(ConnectionState::Connected).await; info!(topic = %&self.config.topic, "listening"); while let Some(msg) = stream.next().await { @@ -254,7 +251,10 @@ impl ListenerActor { match event { ServerEvent::Message(msg) => { debug!("message event"); - self.event_tx.send(ListenerEvent::Message(msg)).await.unwrap(); + self.event_tx + .send(ListenerEvent::Message(msg)) + .await + .unwrap(); } ServerEvent::KeepAlive { .. } => { debug!("keepalive event"); @@ -283,7 +283,10 @@ impl ListenerHandle { // the response will be sent as an event in self.events pub async fn request_state(&self) -> ConnectionState { let (tx, rx) = oneshot::channel(); - self.commands.send(ListenerCommand::GetState(tx)).await.unwrap(); + self.commands + .send(ListenerCommand::GetState(tx)) + .await + .unwrap(); rx.await.unwrap() } } @@ -336,7 +339,6 @@ mod tests { let mut listener = ListenerActor::new(config.clone()); let items: Vec<_> = listener.events.take(3).collect().await; - dbg!(&items); assert!(matches!( @@ -355,7 +357,7 @@ mod tests { // ListenerEvent::Connected { .. }, // )); }); - local_set.await; + local_set.await; } #[tokio::test] @@ -400,23 +402,22 @@ mod tests { #[tokio::test] async fn integration_connects_sends_receives_simple() { let local_set = LocalSet::new(); - local_set - .spawn_local(async { - let http_client = HttpClient::new(reqwest::Client::new()); - let credentials = Credentials::new_nullable(vec![]).await.unwrap(); + local_set.spawn_local(async { + let http_client = HttpClient::new(reqwest::Client::new()); + let credentials = Credentials::new_nullable(vec![]).await.unwrap(); - let config = ListenerConfig { - http_client, - credentials, - endpoint: "http://localhost:8000".to_string(), - topic: "test".to_string(), - since: 0, - }; + let config = ListenerConfig { + http_client, + credentials, + endpoint: "http://localhost:8000".to_string(), + topic: "test".to_string(), + since: 0, + }; - let mut listener = ListenerActor::new(config.clone()); + let mut listener = ListenerActor::new(config.clone()); - // assert_event_matches!(listener, ListenerEvent::Connected { .. },); - }); - local_set.await; + // assert_event_matches!(listener, ListenerEvent::Connected { .. },); + }); + local_set.await; } } diff --git a/ntfy-daemon/src/message_repo/mod.rs b/ntfy-daemon/src/message_repo/mod.rs index 403c4e8..866689a 100644 --- a/ntfy-daemon/src/message_repo/mod.rs +++ b/ntfy-daemon/src/message_repo/mod.rs @@ -28,7 +28,8 @@ impl Db { } fn migrate(&mut self) -> Result<()> { self.conn - .read().unwrap() + .read() + .unwrap() .execute_batch(include_str!("./migrations/00.sql"))?; Ok(()) } diff --git a/ntfy-daemon/src/models.rs b/ntfy-daemon/src/models.rs index 85f4efc..586f717 100644 --- a/ntfy-daemon/src/models.rs +++ b/ntfy-daemon/src/models.rs @@ -323,7 +323,7 @@ impl From for u8 { #[derive(Clone, Debug)] pub struct Account { pub server: String, - pub username: String + pub username: String, } pub struct Notification { @@ -340,14 +340,11 @@ pub trait NetworkMonitorProxy: Sync + Send { fn listen(&self) -> Pin>>; } - -pub struct NullNotifier { - -} +pub struct NullNotifier {} impl NullNotifier { pub fn new() -> Self { - Self {} + Self {} } } impl NotificationProxy for NullNotifier { @@ -356,9 +353,7 @@ impl NotificationProxy for NullNotifier { } } -pub struct NullNetworkMonitor { - -} +pub struct NullNetworkMonitor {} impl NullNetworkMonitor { pub fn new() -> Self { @@ -370,4 +365,4 @@ impl NetworkMonitorProxy for NullNetworkMonitor { fn listen(&self) -> Pin>> { todo!() } -} \ No newline at end of file +} diff --git a/ntfy-daemon/src/ntfy.rs b/ntfy-daemon/src/ntfy.rs index 4547fa7..d3518ce 100644 --- a/ntfy-daemon/src/ntfy.rs +++ b/ntfy-daemon/src/ntfy.rs @@ -16,11 +16,9 @@ use crate::{ ListenerActor, ListenerCommand, ListenerConfig, ListenerHandle, SharedEnv, SubscriptionHandle, }; - const CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(15); const TIMEOUT: std::time::Duration = std::time::Duration::from_secs(240); // 4 minutes - pub fn build_client() -> anyhow::Result { Ok(reqwest::Client::builder() .connect_timeout(CONNECT_TIMEOUT) @@ -208,7 +206,11 @@ impl NtfyActor { password, respond_to, } => { - let result = self.env.credentials.insert(&server, &username, &password).await; + let result = self + .env + .credentials + .insert(&server, &username, &password) + .await; let _ = respond_to.send(result); } @@ -341,7 +343,12 @@ impl NtfyHandle { rx.await.map_err(|_| anyhow!("Actor response error"))? } - pub async fn add_account(&self, server: &str, username: &str, password: &str) -> anyhow::Result<()> { + pub async fn add_account( + &self, + server: &str, + username: &str, + password: &str, + ) -> anyhow::Result<()> { let (tx, rx) = oneshot::channel(); self.command_tx .send(NtfyMessage::AddAccount { diff --git a/ntfy-daemon/src/output_tracker.rs b/ntfy-daemon/src/output_tracker.rs index f3ed90b..00423f7 100644 --- a/ntfy-daemon/src/output_tracker.rs +++ b/ntfy-daemon/src/output_tracker.rs @@ -9,7 +9,9 @@ pub struct OutputTracker { impl Default for OutputTracker { fn default() -> Self { - Self { store: Default::default() } + Self { + store: Default::default(), + } } } @@ -41,7 +43,9 @@ pub struct OutputTrackerAsync { impl Default for OutputTrackerAsync { fn default() -> Self { - Self { store: Default::default() } + Self { + store: Default::default(), + } } } @@ -64,4 +68,4 @@ impl OutputTrackerAsync { vec![] } } -} \ No newline at end of file +} diff --git a/src/subscription.rs b/src/subscription.rs index ec82421..871dcd8 100644 --- a/src/subscription.rs +++ b/src/subscription.rs @@ -214,12 +214,17 @@ impl Subscription { async fn send_updated_info(&self) -> anyhow::Result<()> { let imp = self.imp(); - imp.client.get().unwrap().update_info( - models::Subscription::builder(self.topic()) - .display_name((imp.display_name.borrow().to_string())) - .muted(imp.muted.get()) - .build().map_err(|e| anyhow::anyhow!("invalid subscription data"))? - ).await?; + imp.client + .get() + .unwrap() + .update_info( + models::Subscription::builder(self.topic()) + .display_name((imp.display_name.borrow().to_string())) + .muted(imp.muted.get()) + .build() + .map_err(|e| anyhow::anyhow!("invalid subscription data"))?, + ) + .await?; Ok(()) } fn last_message(list: &gio::ListStore) -> Option { @@ -259,7 +264,12 @@ impl Subscription { }; let this = self.clone(); - this.imp().client.get().unwrap().update_read_until(value).await?; + this.imp() + .client + .get() + .unwrap() + .update_read_until(value) + .await?; this.imp().read_until.set(value); this.update_unread_count(); diff --git a/src/widgets/preferences.rs b/src/widgets/preferences.rs index d10cf6e..16ead84 100644 --- a/src/widgets/preferences.rs +++ b/src/widgets/preferences.rs @@ -78,7 +78,7 @@ glib::wrapper! { } impl NotifyPreferences { - pub fn new(notifier: ntfy_daemon::NtfyHandle ) -> Self { + pub fn new(notifier: ntfy_daemon::NtfyHandle) -> Self { let obj: Self = glib::Object::builder().build(); obj.imp() .notifier @@ -107,7 +107,6 @@ impl NotifyPreferences { imp.added_accounts.remove_all(); for a in accounts { - let row = adw::ActionRow::builder() .title(&a.server) .subtitle(&a.username) @@ -137,13 +136,22 @@ impl NotifyPreferences { let server = imp.server_entry.text(); let username = imp.username_entry.text(); - imp.notifier.get().unwrap().add_account(&server, &username, &password).await?; + imp.notifier + .get() + .unwrap() + .add_account(&server, &username, &password) + .await?; self.show_accounts().await?; Ok(()) } pub async fn remove_account(&self, server: &str) -> anyhow::Result<()> { - self.imp().notifier.get().unwrap().remove_account(server).await?; + self.imp() + .notifier + .get() + .unwrap() + .remove_account(server) + .await?; self.show_accounts().await?; Ok(()) }