chore: updates after iced-rebase
This commit is contained in:
Ashley Wulber
GitHub
parent
bd0d180482
commit
71d9d6d5bb
41 changed files with 1786 additions and 2396 deletions
|
|
@ -33,73 +33,75 @@ pub enum Output {
|
|||
pub fn proxy() -> Subscription<Output> {
|
||||
struct SomeWorker;
|
||||
|
||||
Subscription::run_with_id(
|
||||
std::any::TypeId::of::<SomeWorker>(),
|
||||
stream::channel(50, |mut output| async move {
|
||||
let mut state = State::Ready;
|
||||
Subscription::run_with(std::any::TypeId::of::<SomeWorker>(), |_| {
|
||||
stream::channel(
|
||||
50,
|
||||
|mut output: futures::channel::mpsc::Sender<Output>| async move {
|
||||
let mut state = State::Ready;
|
||||
|
||||
loop {
|
||||
match &mut state {
|
||||
State::Ready => {
|
||||
let (sender, receiver) = channel(10);
|
||||
let Ok(conn) = Connection::session().await else {
|
||||
error!("Failed to connect to session bus");
|
||||
state = State::Finished;
|
||||
continue;
|
||||
};
|
||||
loop {
|
||||
match &mut state {
|
||||
State::Ready => {
|
||||
let (sender, receiver) = channel(10);
|
||||
let Ok(conn) = Connection::session().await else {
|
||||
error!("Failed to connect to session bus");
|
||||
state = State::Finished;
|
||||
continue;
|
||||
};
|
||||
|
||||
let Ok(proxy) = NotificationsProxy::new(&conn).await else {
|
||||
error!("Failed to create proxy from session connection");
|
||||
state = State::Finished;
|
||||
continue;
|
||||
};
|
||||
let tx = sender.clone();
|
||||
if let Err(err) = output.send(Output::Ready(sender)).await {
|
||||
error!("Failed to send sender: {}", err);
|
||||
state = State::Finished;
|
||||
continue;
|
||||
}
|
||||
state = match proxy.receive_notification_closed().await {
|
||||
Ok(mut s) => {
|
||||
tokio::spawn(async move {
|
||||
while let Some(msg) = s.next().await {
|
||||
let Ok(id) = msg.args().map(|args| args.id) else {
|
||||
continue;
|
||||
};
|
||||
_ = tx.send(Input::CloseEvent(id)).await;
|
||||
}
|
||||
});
|
||||
State::WaitingForNotificationEvent(proxy, receiver)
|
||||
let Ok(proxy) = NotificationsProxy::new(&conn).await else {
|
||||
error!("Failed to create proxy from session connection");
|
||||
state = State::Finished;
|
||||
continue;
|
||||
};
|
||||
let tx = sender.clone();
|
||||
if let Err(err) = output.send(Output::Ready(sender)).await {
|
||||
error!("Failed to send sender: {}", err);
|
||||
state = State::Finished;
|
||||
continue;
|
||||
}
|
||||
Err(err) => {
|
||||
error!(
|
||||
"failed to get a stream of signals for notifications. {}",
|
||||
err
|
||||
);
|
||||
State::Finished
|
||||
state = match proxy.receive_notification_closed().await {
|
||||
Ok(mut s) => {
|
||||
tokio::spawn(async move {
|
||||
while let Some(msg) = s.next().await {
|
||||
let Ok(id) = msg.args().map(|args| args.id) else {
|
||||
continue;
|
||||
};
|
||||
_ = tx.send(Input::CloseEvent(id)).await;
|
||||
}
|
||||
});
|
||||
State::WaitingForNotificationEvent(proxy, receiver)
|
||||
}
|
||||
Err(err) => {
|
||||
error!(
|
||||
"failed to get a stream of signals for notifications. {}",
|
||||
err
|
||||
);
|
||||
State::Finished
|
||||
}
|
||||
};
|
||||
}
|
||||
State::WaitingForNotificationEvent(proxy, rx) => match rx.recv().await {
|
||||
Some(Input::Dismiss(id)) => {
|
||||
if let Err(err) = proxy.close_notification(id).await {
|
||||
error!("Failed to close notification: {}", err);
|
||||
}
|
||||
}
|
||||
};
|
||||
}
|
||||
State::WaitingForNotificationEvent(proxy, rx) => match rx.recv().await {
|
||||
Some(Input::Dismiss(id)) => {
|
||||
if let Err(err) = proxy.close_notification(id).await {
|
||||
error!("Failed to close notification: {}", err);
|
||||
Some(Input::CloseEvent(id)) => {
|
||||
_ = output.send(Output::CloseEvent(id)).await;
|
||||
}
|
||||
None => {
|
||||
warn!("Notification event channel closed");
|
||||
state = State::Finished;
|
||||
continue;
|
||||
}
|
||||
},
|
||||
State::Finished => {
|
||||
let () = futures::future::pending().await;
|
||||
}
|
||||
Some(Input::CloseEvent(id)) => {
|
||||
_ = output.send(Output::CloseEvent(id)).await;
|
||||
}
|
||||
None => {
|
||||
warn!("Notification event channel closed");
|
||||
state = State::Finished;
|
||||
continue;
|
||||
}
|
||||
},
|
||||
State::Finished => {
|
||||
let () = futures::future::pending().await;
|
||||
}
|
||||
}
|
||||
}
|
||||
}),
|
||||
)
|
||||
},
|
||||
)
|
||||
})
|
||||
}
|
||||
|
|
|
|||
|
|
@ -12,6 +12,7 @@ use cosmic_notifications_util::Notification;
|
|||
use futures_util::{SinkExt, StreamExt};
|
||||
use std::{
|
||||
collections::HashMap,
|
||||
hash::Hash,
|
||||
os::unix::io::{FromRawFd, RawFd},
|
||||
pin::pin,
|
||||
};
|
||||
|
|
@ -37,89 +38,97 @@ pub enum Output {
|
|||
}
|
||||
|
||||
pub fn notifications(proxy: NotificationsAppletProxy<'static>) -> Subscription<Output> {
|
||||
struct SomeWorker;
|
||||
struct Wrapper(NotificationsAppletProxy<'static>);
|
||||
|
||||
Subscription::run_with_id(
|
||||
std::any::TypeId::of::<SomeWorker>(),
|
||||
stream::channel(50, |mut output| async move {
|
||||
let mut state = State::WaitingForNotificationEvent;
|
||||
let (sender, mut receiver) = mpsc::channel(10);
|
||||
_ = output.send(Output::Ready(sender)).await;
|
||||
impl Hash for Wrapper {
|
||||
fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
|
||||
std::any::TypeId::of::<NotificationsAppletProxy<'static>>().hash(state);
|
||||
}
|
||||
}
|
||||
Subscription::run_with(Wrapper(proxy), |Wrapper(proxy)| {
|
||||
let proxy = proxy.clone();
|
||||
stream::channel(
|
||||
50,
|
||||
move |mut output: futures::channel::mpsc::Sender<Output>| async move {
|
||||
let mut state = State::WaitingForNotificationEvent;
|
||||
let (sender, mut receiver) = mpsc::channel(10);
|
||||
_ = output.send(Output::Ready(sender)).await;
|
||||
|
||||
let mut signal;
|
||||
let mut fail_count: u8 = 0;
|
||||
loop {
|
||||
match proxy.receive_notify().await {
|
||||
Ok(s) => {
|
||||
signal = s;
|
||||
break;
|
||||
}
|
||||
Err(err) => {
|
||||
error!(
|
||||
"failed to get a stream of signals for notifications. {}",
|
||||
err
|
||||
);
|
||||
fail_count = fail_count.saturating_add(1);
|
||||
if fail_count > 5 {
|
||||
error!("Failed to receive notification events");
|
||||
// exit because the applet needs the notifications daemon in order to work properly
|
||||
std::process::exit(0);
|
||||
} else {
|
||||
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
|
||||
let mut signal;
|
||||
let mut fail_count: u8 = 0;
|
||||
loop {
|
||||
match proxy.receive_notify().await {
|
||||
Ok(s) => {
|
||||
signal = s;
|
||||
break;
|
||||
}
|
||||
Err(err) => {
|
||||
error!(
|
||||
"failed to get a stream of signals for notifications. {}",
|
||||
err
|
||||
);
|
||||
fail_count = fail_count.saturating_add(1);
|
||||
if fail_count > 5 {
|
||||
error!("Failed to receive notification events");
|
||||
// exit because the applet needs the notifications daemon in order to work properly
|
||||
std::process::exit(0);
|
||||
} else {
|
||||
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
}
|
||||
}
|
||||
loop {
|
||||
match &mut state {
|
||||
State::WaitingForNotificationEvent => {
|
||||
trace!("Waiting for notification events...");
|
||||
let mut next_signal = signal.next();
|
||||
let mut next_input = pin!(receiver.recv().fuse());
|
||||
cosmic::iced::futures::select! {
|
||||
v = next_signal => {
|
||||
if let Some(msg) = v {
|
||||
let Ok(args) = msg.args() else {
|
||||
break;
|
||||
};
|
||||
let notification = Notification::new(
|
||||
args.app_name,
|
||||
args.id,
|
||||
args.app_icon,
|
||||
args.summary,
|
||||
args.body,
|
||||
args.actions,
|
||||
args.hints,
|
||||
args.expire_timeout,
|
||||
);
|
||||
_ = output.send(Output::Notification(notification)).await;
|
||||
} else {
|
||||
tracing::error!("Signal stream closed, ending notifications subscription");
|
||||
state = State::Finished;
|
||||
}
|
||||
}
|
||||
v = next_input => {
|
||||
if let Some(Input::Activated(id, action)) = v {
|
||||
if proxy.invoke_action(id, action.clone()).await.is_err() {
|
||||
tracing::error!("Failed to invoke action {id} {action}");
|
||||
loop {
|
||||
match &mut state {
|
||||
State::WaitingForNotificationEvent => {
|
||||
trace!("Waiting for notification events...");
|
||||
let mut next_signal = signal.next();
|
||||
let mut next_input = pin!(receiver.recv().fuse());
|
||||
cosmic::iced::futures::select! {
|
||||
v = next_signal => {
|
||||
if let Some(msg) = v {
|
||||
let Ok(args) = msg.args() else {
|
||||
break;
|
||||
};
|
||||
let notification = Notification::new(
|
||||
args.app_name,
|
||||
args.id,
|
||||
args.app_icon,
|
||||
args.summary,
|
||||
args.body,
|
||||
args.actions,
|
||||
args.hints,
|
||||
args.expire_timeout,
|
||||
);
|
||||
_ = output.send(Output::Notification(notification)).await;
|
||||
} else {
|
||||
tracing::error!("Invoked {action} for {id}");
|
||||
tracing::error!("Signal stream closed, ending notifications subscription");
|
||||
state = State::Finished;
|
||||
}
|
||||
}
|
||||
v = next_input => {
|
||||
if let Some(Input::Activated(id, action)) = v {
|
||||
if proxy.invoke_action(id, action.clone()).await.is_err() {
|
||||
tracing::error!("Failed to invoke action {id} {action}");
|
||||
} else {
|
||||
tracing::error!("Invoked {action} for {id}");
|
||||
}
|
||||
} else {
|
||||
tracing::error!("Channel closed, ending notifications subscription");
|
||||
state = State::Finished;
|
||||
}
|
||||
} else {
|
||||
tracing::error!("Channel closed, ending notifications subscription");
|
||||
state = State::Finished;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
State::Finished => {
|
||||
let () = futures::future::pending().await;
|
||||
State::Finished => {
|
||||
let () = futures::future::pending().await;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}),
|
||||
)
|
||||
},
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
#[proxy(
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue