secstate/Notify
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fmt all

Browse Source
This commit is contained in:
ranfdev
2024-11-20 20:41:46 +01:00
parent 812e335c41
commit 9c7f701557
10 changed files with 146 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
ntfy-daemon/src/credentials.rs
View File

@ -9,7 +9,7 @@ use async_trait::async_trait;
pub struct KeyringItem { pub struct KeyringItem {
attributes: HashMap<String, String>, attributes: HashMap<String, String>,
// we could zero-out this region of memory // we could zero-out this region of memory
secret: Vec<u8> secret: Vec<u8>,
} }
impl KeyringItem { impl KeyringItem {
@ -121,7 +121,10 @@ impl NullableKeyring {
("username".to_string(), cred.username.clone()), ("username".to_string(), cred.username.clone()),
("server".to_string(), cred.password.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 } Self { search_response }
@ -163,17 +166,12 @@ impl Credentials {
} }
pub async fn load(&mut self) -> anyhow::Result<()> { pub async fn load(&mut self) -> anyhow::Result<()> {
let attrs = HashMap::from([("type", "password")]); let attrs = HashMap::from([("type", "password")]);
let values = self let values = self.keyring.search_items(attrs).await?;
.keyring
.search_items(attrs)
.await?;
let mut lock = self.creds.write().unwrap(); let mut lock = self.creds.write().unwrap();
lock.clear(); lock.clear();
for item in values { for item in values {
let attrs = item let attrs = item.attributes().await;
.attributes()
.await;
lock.insert( lock.insert(
attrs["server"].to_string(), attrs["server"].to_string(),
Credential { Credential {
@ -230,9 +228,7 @@ impl Credentials {
("username", &creds.username), ("username", &creds.username),
("server", server), ("server", server),
]); ]);
self.keyring self.keyring.delete(attrs).await?;
.delete(attrs)
.await?;
self.creds self.creds
.write() .write()
.unwrap() .unwrap()

26
ntfy-daemon/src/http_client.rs
View File

@ -2,11 +2,11 @@ use anyhow::Result;
use async_trait::async_trait; use async_trait::async_trait;
use reqwest::{header::HeaderMap, Client, Request, RequestBuilder, Response, ResponseBuilderExt}; use reqwest::{header::HeaderMap, Client, Request, RequestBuilder, Response, ResponseBuilderExt};
use serde_json::{json, Value}; use serde_json::{json, Value};
use tokio::time;
use std::collections::{HashMap, VecDeque}; use std::collections::{HashMap, VecDeque};
use std::sync::Arc; use std::sync::Arc;
use std::time::Duration; use std::time::Duration;
use tokio::sync::RwLock; use tokio::sync::RwLock;
use tokio::time;
use crate::models; use crate::models;
use crate::output_tracker::OutputTrackerAsync; use crate::output_tracker::OutputTrackerAsync;
@ -86,7 +86,6 @@ impl HttpClient {
} }
} }
#[derive(Clone, Default)] #[derive(Clone, Default)]
pub struct NullableClient { pub struct NullableClient {
responses: Arc<RwLock<HashMap<String, VecDeque<Response>>>>, responses: Arc<RwLock<HashMap<String, VecDeque<Response>>>>,
@ -161,7 +160,12 @@ impl NullableClientBuilder {
pub fn build(self) -> NullableClient { pub fn build(self) -> NullableClient {
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)), default_response: Arc::new(RwLock::new(self.default_response)),
} }
} }
@ -183,7 +187,7 @@ impl LightHttpClient for NullableClient {
time::sleep(Duration::from_millis(1)).await; time::sleep(Duration::from_millis(1)).await;
let url = request.url().to_string(); let url = request.url().to_string();
let mut responses = self.responses.write().await; let mut responses = self.responses.write().await;
if let Some(url_responses) = responses.get_mut(&url) { if let Some(url_responses) = responses.get_mut(&url) {
if let Some(response) = url_responses.pop_front() { if let Some(response) = url_responses.pop_front() {
// Remove the URL entry if no more responses // Remove the URL entry if no more responses
@ -282,20 +286,26 @@ mod tests {
let http_client = HttpClient::new_nullable(client); let http_client = HttpClient::new_nullable(client);
// First request gets first response // 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?; let response = http_client.execute(request).await?;
assert_eq!(response.text().await?, "first"); assert_eq!(response.text().await?, "first");
// Second request gets second response // 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?; let response = http_client.execute(request).await?;
assert_eq!(response.text().await?, "second"); assert_eq!(response.text().await?, "second");
// Third request fails (no more responses) // 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; let result = http_client.execute(request).await;
assert!(result.is_err()); assert!(result.is_err());
Ok(()) Ok(())
} }
} }

13
ntfy-daemon/src/lib.rs
View File

@ -1,18 +1,18 @@
pub mod credentials; pub mod credentials;
mod http_client;
mod listener;
pub mod message_repo; pub mod message_repo;
pub mod models; pub mod models;
pub mod retry;
mod http_client;
mod output_tracker;
mod listener;
mod ntfy; mod ntfy;
mod output_tracker;
pub mod retry;
mod subscription; mod subscription;
pub use subscription::SubscriptionHandle;
pub use listener::*; pub use listener::*;
pub use ntfy::NtfyHandle;
pub use ntfy::start; pub use ntfy::start;
pub use ntfy::NtfyHandle;
use std::sync::Arc; use std::sync::Arc;
pub use subscription::SubscriptionHandle;
use http_client::HttpClient; use http_client::HttpClient;
@ -43,4 +43,3 @@ pub enum Error {
#[error("subscription not found while {0}")] #[error("subscription not found while {0}")]
SubscriptionNotFound(String), SubscriptionNotFound(String),
} }

119
ntfy-daemon/src/listener.rs
View File

@ -1,7 +1,7 @@
use std::cell::RefCell; use std::cell::RefCell;
use std::sync::Arc; use std::sync::Arc;
use std::thread::JoinHandle; use std::thread::JoinHandle;
use std::{time::Duration}; use std::time::Duration;
use futures::{StreamExt, TryStreamExt}; use futures::{StreamExt, TryStreamExt};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
@ -11,7 +11,7 @@ use tokio::sync::RwLock;
use tokio::task::{self, spawn_local, AbortHandle, LocalSet}; use tokio::task::{self, spawn_local, AbortHandle, LocalSet};
use tokio::{ use tokio::{
select, select,
sync::{mpsc, watch, oneshot}, sync::{mpsc, oneshot, watch},
}; };
use tokio_stream::wrappers::LinesStream; use tokio_stream::wrappers::LinesStream;
use tracing::{debug, error, info}; use tracing::{debug, error, info};
@ -36,9 +36,7 @@ pub enum ServerEvent {
topic: String, topic: String,
}, },
#[serde(rename = "message")] #[serde(rename = "message")]
Message ( Message(models::Message),
models::Message,
),
#[serde(rename = "keepalive")] #[serde(rename = "keepalive")]
KeepAlive { KeepAlive {
id: String, id: String,
@ -116,9 +114,7 @@ pub struct ListenerActor {
} }
impl ListenerActor { impl ListenerActor {
pub fn new( pub fn new(config: ListenerConfig) -> ListenerHandle {
config: ListenerConfig,
) -> ListenerHandle {
let (event_tx, event_rx) = async_channel::bounded(64); let (event_tx, event_rx) = async_channel::bounded(64);
let (commands_tx, commands_rx) = mpsc::channel(1); let (commands_tx, commands_rx) = mpsc::channel(1);
@ -127,7 +123,6 @@ impl ListenerActor {
// use a new local set to isolate panics // use a new local set to isolate panics
let local_set = LocalSet::new(); let local_set = LocalSet::new();
local_set.spawn_local(async move { local_set.spawn_local(async move {
let this = Self { let this = Self {
event_tx, event_tx,
commands_rx: Some(commands_rx), commands_rx: Some(commands_rx),
@ -149,41 +144,44 @@ impl ListenerActor {
} }
pub async fn run_loop(mut self) { pub async fn run_loop(mut self) {
let mut commands_rx = self.commands_rx.take().unwrap(); let mut commands_rx = self.commands_rx.take().unwrap();
loop { loop {
select! { select! {
_ = self.run_supervised_loop() => { _ = self.run_supervised_loop() => {
// the supervised loop cannot fail. If it finished, don't restart. // the supervised loop cannot fail. If it finished, don't restart.
break; break;
}, },
cmd = commands_rx.recv() => { cmd = commands_rx.recv() => {
match cmd { match cmd {
Some(ListenerCommand::Restart) => { Some(ListenerCommand::Restart) => {
info!("Received restart command"); info!("Received restart command");
continue; continue;
} }
Some(ListenerCommand::Shutdown) => { Some(ListenerCommand::Shutdown) => {
info!("Received shutdown command"); info!("Received shutdown command");
break; break;
} }
Some(ListenerCommand::GetState(tx)) => { Some(ListenerCommand::GetState(tx)) => {
info!("Received get state command"); info!("Received get state command");
let state = self.state.clone(); let state = self.state.clone();
let _ = tx.send(state); let _ = tx.send(state);
} }
None => { None => {
error!("Channel closed for ListenerActor"); error!("Channel closed for ListenerActor");
break; break;
}
} }
} }
} }
} }
}
} }
async fn set_state(&mut self, state: ConnectionState) { async fn set_state(&mut self, state: ConnectionState) {
self.state = state.clone(); 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) { async fn run_supervised_loop(&mut self) {
dbg!("supervised"); dbg!("supervised");
@ -208,7 +206,8 @@ impl ListenerActor {
retry_count: retry.count(), retry_count: retry.count(),
delay: retry.next_delay(), delay: retry.next_delay(),
error: Some(Arc::new(e)), error: Some(Arc::new(e)),
}).await; })
.await;
info!(delay = ?retry.next_delay(), "restarting"); info!(delay = ?retry.next_delay(), "restarting");
retry.wait().await; retry.wait().await;
} else { } else {
@ -236,9 +235,7 @@ impl ListenerActor {
let stream = response_lines(reader).await?; let stream = response_lines(reader).await?;
tokio::pin!(stream); tokio::pin!(stream);
self.set_state( self.set_state(ConnectionState::Connected).await;
ConnectionState::Connected,
).await;
info!(topic = %&self.config.topic, "listening"); info!(topic = %&self.config.topic, "listening");
while let Some(msg) = stream.next().await { while let Some(msg) = stream.next().await {
@ -254,7 +251,10 @@ impl ListenerActor {
match event { match event {
ServerEvent::Message(msg) => { ServerEvent::Message(msg) => {
debug!("message event"); debug!("message event");
self.event_tx.send(ListenerEvent::Message(msg)).await.unwrap(); self.event_tx
.send(ListenerEvent::Message(msg))
.await
.unwrap();
} }
ServerEvent::KeepAlive { .. } => { ServerEvent::KeepAlive { .. } => {
debug!("keepalive event"); debug!("keepalive event");
@ -283,7 +283,10 @@ impl ListenerHandle {
// the response will be sent as an event in self.events // the response will be sent as an event in self.events
pub async fn request_state(&self) -> ConnectionState { pub async fn request_state(&self) -> ConnectionState {
let (tx, rx) = oneshot::channel(); let (tx, rx) = oneshot::channel();
self.commands.send(ListenerCommand::GetState(tx)).await.unwrap(); self.commands
.send(ListenerCommand::GetState(tx))
.await
.unwrap();
rx.await.unwrap() rx.await.unwrap()
} }
} }
@ -336,7 +339,6 @@ mod tests {
let mut listener = ListenerActor::new(config.clone()); let mut listener = ListenerActor::new(config.clone());
let items: Vec<_> = listener.events.take(3).collect().await; let items: Vec<_> = listener.events.take(3).collect().await;
dbg!(&items); dbg!(&items);
assert!(matches!( assert!(matches!(
@ -355,7 +357,7 @@ mod tests {
// ListenerEvent::Connected { .. }, // ListenerEvent::Connected { .. },
// )); // ));
}); });
local_set.await; local_set.await;
} }
#[tokio::test] #[tokio::test]
@ -400,23 +402,22 @@ mod tests {
#[tokio::test] #[tokio::test]
async fn integration_connects_sends_receives_simple() { async fn integration_connects_sends_receives_simple() {
let local_set = LocalSet::new(); let local_set = LocalSet::new();
local_set local_set.spawn_local(async {
.spawn_local(async { let http_client = HttpClient::new(reqwest::Client::new());
let http_client = HttpClient::new(reqwest::Client::new()); let credentials = Credentials::new_nullable(vec![]).await.unwrap();
let credentials = Credentials::new_nullable(vec![]).await.unwrap();
let config = ListenerConfig { let config = ListenerConfig {
http_client, http_client,
credentials, credentials,
endpoint: "http://localhost:8000".to_string(), endpoint: "http://localhost:8000".to_string(),
topic: "test".to_string(), topic: "test".to_string(),
since: 0, since: 0,
}; };
let mut listener = ListenerActor::new(config.clone()); let mut listener = ListenerActor::new(config.clone());
// assert_event_matches!(listener, ListenerEvent::Connected { .. },); // assert_event_matches!(listener, ListenerEvent::Connected { .. },);
}); });
local_set.await; local_set.await;
} }
} }

3
ntfy-daemon/src/message_repo/mod.rs
View File

@ -28,7 +28,8 @@ impl Db {
} }
fn migrate(&mut self) -> Result<()> { fn migrate(&mut self) -> Result<()> {
self.conn self.conn
.read().unwrap() .read()
.unwrap()
.execute_batch(include_str!("./migrations/00.sql"))?; .execute_batch(include_str!("./migrations/00.sql"))?;
Ok(()) Ok(())
} }

15
ntfy-daemon/src/models.rs
View File

@ -323,7 +323,7 @@ impl From<Status> for u8 {
#[derive(Clone, Debug)] #[derive(Clone, Debug)]
pub struct Account { pub struct Account {
pub server: String, pub server: String,
pub username: String pub username: String,
} }
pub struct Notification { pub struct Notification {
@ -340,14 +340,11 @@ pub trait NetworkMonitorProxy: Sync + Send {
fn listen(&self) -> Pin<Box<dyn Stream<Item = ()>>>; fn listen(&self) -> Pin<Box<dyn Stream<Item = ()>>>;
} }
pub struct NullNotifier {}
pub struct NullNotifier {
}
impl NullNotifier { impl NullNotifier {
pub fn new() -> Self { pub fn new() -> Self {
Self {} Self {}
} }
} }
impl NotificationProxy for NullNotifier { impl NotificationProxy for NullNotifier {
@ -356,9 +353,7 @@ impl NotificationProxy for NullNotifier {
} }
} }
pub struct NullNetworkMonitor { pub struct NullNetworkMonitor {}
}
impl NullNetworkMonitor { impl NullNetworkMonitor {
pub fn new() -> Self { pub fn new() -> Self {
@ -370,4 +365,4 @@ impl NetworkMonitorProxy for NullNetworkMonitor {
fn listen(&self) -> Pin<Box<dyn Stream<Item = ()>>> { fn listen(&self) -> Pin<Box<dyn Stream<Item = ()>>> {
todo!() todo!()
} }
} }

15
ntfy-daemon/src/ntfy.rs
View File

@ -16,11 +16,9 @@ use crate::{
ListenerActor, ListenerCommand, ListenerConfig, ListenerHandle, SharedEnv, SubscriptionHandle, ListenerActor, ListenerCommand, ListenerConfig, ListenerHandle, SharedEnv, SubscriptionHandle,
}; };
const CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(15); 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 const TIMEOUT: std::time::Duration = std::time::Duration::from_secs(240); // 4 minutes
pub fn build_client() -> anyhow::Result<reqwest::Client> { pub fn build_client() -> anyhow::Result<reqwest::Client> {
Ok(reqwest::Client::builder() Ok(reqwest::Client::builder()
.connect_timeout(CONNECT_TIMEOUT) .connect_timeout(CONNECT_TIMEOUT)
@ -208,7 +206,11 @@ impl NtfyActor {
password, password,
respond_to, 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); let _ = respond_to.send(result);
} }
@ -341,7 +343,12 @@ impl NtfyHandle {
rx.await.map_err(|_| anyhow!("Actor response error"))? 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(); let (tx, rx) = oneshot::channel();
self.command_tx self.command_tx
.send(NtfyMessage::AddAccount { .send(NtfyMessage::AddAccount {

10
ntfy-daemon/src/output_tracker.rs
View File

@ -9,7 +9,9 @@ pub struct OutputTracker<T> {
impl<T> Default for OutputTracker<T> { impl<T> Default for OutputTracker<T> {
fn default() -> Self { fn default() -> Self {
Self { store: Default::default() } Self {
store: Default::default(),
}
} }
} }
@ -41,7 +43,9 @@ pub struct OutputTrackerAsync<T> {
impl<T> Default for OutputTrackerAsync<T> { impl<T> Default for OutputTrackerAsync<T> {
fn default() -> Self { fn default() -> Self {
Self { store: Default::default() } Self {
store: Default::default(),
}
} }
} }
@ -64,4 +68,4 @@ impl<T: Clone> OutputTrackerAsync<T> {
vec![] vec![]
} }
} }
} }

24
src/subscription.rs
View File

@ -214,12 +214,17 @@ impl Subscription {
async fn send_updated_info(&self) -> anyhow::Result<()> { async fn send_updated_info(&self) -> anyhow::Result<()> {
let imp = self.imp(); let imp = self.imp();
imp.client.get().unwrap().update_info( imp.client
models::Subscription::builder(self.topic()) .get()
.display_name((imp.display_name.borrow().to_string())) .unwrap()
.muted(imp.muted.get()) .update_info(
.build().map_err(|e| anyhow::anyhow!("invalid subscription data"))? models::Subscription::builder(self.topic())
).await?; .display_name((imp.display_name.borrow().to_string()))
.muted(imp.muted.get())
.build()
.map_err(|e| anyhow::anyhow!("invalid subscription data"))?,
)
.await?;
Ok(()) Ok(())
} }
fn last_message(list: &gio::ListStore) -> Option<models::Message> { fn last_message(list: &gio::ListStore) -> Option<models::Message> {
@ -259,7 +264,12 @@ impl Subscription {
}; };
let this = self.clone(); 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.imp().read_until.set(value);
this.update_unread_count(); this.update_unread_count();

16
src/widgets/preferences.rs
View File

@ -78,7 +78,7 @@ glib::wrapper! {
} }
impl NotifyPreferences { 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(); let obj: Self = glib::Object::builder().build();
obj.imp() obj.imp()
.notifier .notifier
@ -107,7 +107,6 @@ impl NotifyPreferences {
imp.added_accounts.remove_all(); imp.added_accounts.remove_all();
for a in accounts { for a in accounts {
let row = adw::ActionRow::builder() let row = adw::ActionRow::builder()
.title(&a.server) .title(&a.server)
.subtitle(&a.username) .subtitle(&a.username)
@ -137,13 +136,22 @@ impl NotifyPreferences {
let server = imp.server_entry.text(); let server = imp.server_entry.text();
let username = imp.username_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?; self.show_accounts().await?;
Ok(()) Ok(())
} }
pub async fn remove_account(&self, server: &str) -> anyhow::Result<()> { 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?; self.show_accounts().await?;
Ok(()) Ok(())
} }