From 281ef2bd0e349a32c4b8ac69161b2fe2b957638d Mon Sep 17 00:00:00 2001
From: Kubat <mael.martin31@gmail.com>
Date: Sat, 28 Oct 2023 15:15:30 +0200
Subject: [PATCH] AMADEUS: Query the queue and history content each second...
---
amadeus/src/app.rs | 253 ++++++++++++----------
amadeus/src/components/mainpanel/queue.rs | 30 +--
2 files changed, 157 insertions(+), 126 deletions(-)
diff --git a/amadeus/src/app.rs b/amadeus/src/app.rs
index 2d929548..d5c135ef 100644
--- a/amadeus/src/app.rs
+++ b/amadeus/src/app.rs
@@ -22,7 +22,7 @@ use iced::{
};
use iced_aw::{native::Split, split};
use lektor_lib::{requests::*, ConnectConfigPtr};
-use lektor_payloads::{KId, Kara, KaraFilter, PlayState, PlayStateWithCurrent};
+use lektor_payloads::{KId, Kara, KaraFilter, PlayState, PlayStateWithCurrent, Priority};
use lektor_utils::{config::UserConfig, log};
use std::{future::Future, sync::Arc};
@@ -123,6 +123,63 @@ impl Amadeus {
)
}
+ fn get_history(&self) -> impl Future<Output = Option<Vec<Arc<Kara>>>> {
+ let cfg = self.connect_config.clone();
+ let store = self.kara_store.clone();
+ async move {
+ stream::iter(
+ get_history(cfg)
+ .await
+ .map_err(|err| log::error!("{err}"))
+ .unwrap_or_default(),
+ )
+ .then(|id| KaraStore::into_get(store.clone(), id))
+ .try_collect()
+ .await
+ .map_err(|err| log::error!("{err}"))
+ .ok()
+ }
+ }
+
+ fn get_playlist_content(&self, plt: Arc<str>) -> impl Future<Output = Option<Vec<Arc<Kara>>>> {
+ let cfg = self.connect_config.clone();
+ let store = self.kara_store.clone();
+ async move {
+ stream::iter(
+ get_playlist_content(cfg, plt)
+ .await
+ .map_err(|err| log::error!("{err}"))
+ .unwrap_or_default(),
+ )
+ .then(|id| KaraStore::into_get(store.clone(), id))
+ .try_collect()
+ .await
+ .map_err(|err| log::error!("{err}"))
+ .ok()
+ }
+ }
+
+ fn get_queue(&self) -> impl Future<Output = Option<Vec<(Priority, Arc<Kara>)>>> {
+ let cfg = self.connect_config.clone();
+ let store = self.kara_store.clone();
+ async move {
+ stream::iter(
+ get_queue(cfg)
+ .await
+ .map_err(|err| log::error!("{err}"))
+ .unwrap_or_default(),
+ )
+ .then(move |(p, id)| {
+ let store = store.clone();
+ async move { store.get(id).await.map(|kara| (p, kara)) }
+ })
+ .try_collect()
+ .await
+ .map_err(|err| log::error!("{err}"))
+ .ok()
+ }
+ }
+
/// Handle a config message. This is ugly so we put it in another function.
fn handle_config_message(&mut self, config: config::Message) -> Command<Message> {
if let config::Message::TryConnect = config {
@@ -197,13 +254,13 @@ impl Amadeus {
.map(move |_| RefreshRequest::QueueLevel(prio).into()),
queue::Request::ClearQueueLevel(prio) => Command::batch([
self.mainpanel
- .update(queue::Message::ClearQueueLevel(prio).into())
+ .update(queue::Message::ClearLevel(prio).into())
.map(Message::from),
send(remove_level_from_queue(cfg, prio)),
]),
queue::Request::ClearQueue => Command::batch([
self.mainpanel
- .update(queue::Message::ClearQueue.into())
+ .update(queue::Message::Clear.into())
.map(Message::from),
send(remove_range_from_queue(cfg, ..)),
]),
@@ -219,7 +276,7 @@ impl Amadeus {
}
queue::Request::ToggleQueueLevel(prio, show) => self
.mainpanel
- .update(queue::Message::ToggleQueueLevel(prio, show).into())
+ .update(queue::Message::ToggleLevel(prio, show).into())
.map(Message::from),
},
@@ -297,13 +354,11 @@ impl Amadeus {
match (plt, req) {
(_, AddToQueue(prio, id)) => Command::batch([
send(add_kid_to_queue(cfg, prio, id.clone())),
- self.with_kara(id, move |kara| {
- queue::Message::AddKaraToQueue(prio, kara).into()
- }),
+ self.with_kara(id, move |kara| queue::Message::AddKara(prio, kara).into()),
]),
(_, RemoveFromQueue(id)) => Command::batch([
self.mainpanel
- .update(queue::Message::RemoveKaraFromQueue(id.clone()).into())
+ .update(queue::Message::RemoveKara(id.clone()).into())
.map(Message::from),
send(remove_kid_from_queue(cfg, id)),
]),
@@ -407,94 +462,50 @@ impl Amadeus {
&mut self,
req: RefreshRequest,
) -> Command<<Self as Application>::Message> {
- let cfg = self.connect_config.clone();
- let store = self.kara_store.clone();
match req {
- RefreshRequest::Playlists => Command::perform(get_playlists(cfg), |res| {
- res.map_err(|err| log::error!("{err}"))
- .map(|plts| {
- let main_keep = Message::from(playlists::Message::KeepPlaylists(
- plts.iter().map(|(name, _)| name.as_ref().into()).collect(),
- ));
- let side_keep =
- sidebar::Message::from_iter(plts.iter().map(|(name, _)| name.clone()))
- .into();
- let update_each = plts.into_iter().map(|(name, infos)| {
- let name = name.as_ref().into();
- Message::from(playlists::Message::UpdatePlaylistInfos(name, infos))
- });
- Message::from_iter([main_keep, side_keep].into_iter().chain(update_each))
- })
- .unwrap_or_default()
- }),
+ RefreshRequest::Playlists => {
+ Command::perform(get_playlists(self.connect_config.clone()), |res| {
+ res.map_err(|err| log::error!("{err}"))
+ .map(|plts| {
+ let main_keep = Message::from(playlists::Message::KeepPlaylists(
+ plts.iter().map(|(name, _)| name.as_ref().into()).collect(),
+ ));
+ let side_keep = sidebar::Message::from_iter(
+ plts.iter().map(|(name, _)| name.clone()),
+ )
+ .into();
+ let update_each = plts.into_iter().map(|(name, infos)| {
+ let name = name.as_ref().into();
+ Message::from(playlists::Message::UpdatePlaylistInfos(name, infos))
+ });
+ Message::from_iter(
+ [main_keep, side_keep].into_iter().chain(update_each),
+ )
+ })
+ .unwrap_or_default()
+ })
+ }
RefreshRequest::Playlist(plt) => {
- let plt2 = plt.clone();
- Command::perform(
- async move {
- stream::iter(
- get_playlist_content(cfg, plt)
- .await
- .map_err(|err| log::error!("{err}"))
- .unwrap_or_default(),
- )
- .then(|id| KaraStore::into_get(store.clone(), id))
- .try_collect()
- .await
- },
- move |karas| {
- karas
- .map_err(|err| log::error!("{err}"))
- .map(|karas| Message::from(playlists::Message::Reload(plt2, karas)))
- .unwrap_or_default()
- },
- )
+ Command::perform(self.get_playlist_content(plt.clone()), move |k| {
+ k.map(|k| Message::from(playlists::Message::Reload(plt, k)))
+ .unwrap_or_default()
+ })
}
- RefreshRequest::History => Command::perform(
- async move {
- stream::iter(
- get_history(cfg)
- .await
- .map_err(|err| log::error!("{err}"))
- .unwrap_or_default(),
- )
- .then(|id| KaraStore::into_get(store.clone(), id))
- .try_collect()
- .await
- },
- |karas| {
- karas
- .map(|karas| Message::from(history::Message::Reload(karas)))
- .map_err(|err| log::error!("{err}"))
- .unwrap_or_default()
- },
- ),
+ RefreshRequest::History => Command::perform(self.get_history(), |k| {
+ k.map(|k| Message::from(history::Message::Reload(k)))
+ .unwrap_or_default()
+ }),
// Fow now we always update the whole queue, see later to only update one part of the
// queue as an optimization.
- RefreshRequest::Queue | RefreshRequest::QueueLevel(_) => Command::perform(
- async move {
- stream::iter(
- get_queue(cfg)
- .await
- .map_err(|err| log::error!("{err}"))
- .unwrap_or_default(),
- )
- .then(move |(p, id)| {
- let store = store.clone();
- async move { store.get(id).await.map(|kara| (p, kara)) }
- })
- .try_collect()
- .await
- },
- |karas| {
- karas
- .map(|karas| Message::from(queue::Message::ReloadQueue(karas)))
- .map_err(|err| log::error!("{err}"))
+ RefreshRequest::Queue | RefreshRequest::QueueLevel(_) => {
+ Command::perform(self.get_queue(), |k| {
+ k.map(|k| Message::from(queue::Message::Reload(k)))
.unwrap_or_default()
- },
- ),
+ })
+ }
}
}
@@ -601,38 +612,58 @@ impl Application for Amadeus {
}
// Kill lektord & exit the application
+ Message::ExitApplication => iced::window::close(),
Message::ShutdownLektord => send(shutdown_lektord(self.connect_config.clone()))
.map(|_| Message::ExitApplication),
- Message::ExitApplication => iced::window::close(),
// Messages got from subscriptions.
Message::Tick(instant) => {
let delta = instant.saturating_duration_since(self.last_instant);
self.last_instant = instant;
log::debug!("duration since last instant: {delta:?}");
- use PlayState::*;
- Command::perform(get_status(self.connect_config.clone()), |res| match res {
- Ok(PlayStateWithCurrent {
- state: s @ Play | s @ Pause,
- current: Some((id, elapsed, duration)),
- }) => Message::from_iter([
- Message::ChangedPlayback(s),
- Message::ChangedKaraId(id),
- Message::TimeUpdate(elapsed, duration),
- ]),
- Ok(PlayStateWithCurrent {
- state: Stop,
- current: None,
- }) => Message::ChangedPlayback(Stop),
- Ok(state) => {
- log::error!("got incoherent state from the server: {state:?}");
- Message::ChangedPlayback(PlayState::Stop)
- }
- Err(err) => {
- log::error!("{err}");
- Message::None
- }
- })
+
+ let queue = Command::perform(self.get_queue(), |res| {
+ res.map(|queue| {
+ Message::MainPanelMessage(mainpanel::Message::Queue(
+ queue::Message::Reload(queue),
+ ))
+ })
+ .unwrap_or_default()
+ });
+
+ let history = Command::perform(self.get_history(), |res| {
+ res.map(|history| {
+ Message::MainPanelMessage(mainpanel::Message::History(
+ history::Message::Reload(history),
+ ))
+ })
+ .unwrap_or_default()
+ });
+
+ let status = Command::perform(get_status(self.connect_config.clone()), |res| {
+ res.map_err(|err| log::error!("{err}"))
+ .map(|res| match res {
+ PlayStateWithCurrent {
+ state: s @ PlayState::Play | s @ PlayState::Pause,
+ current: Some((id, elapsed, duration)),
+ } => Message::from_iter([
+ Message::ChangedPlayback(s),
+ Message::ChangedKaraId(id),
+ Message::TimeUpdate(elapsed, duration),
+ ]),
+ PlayStateWithCurrent {
+ state: PlayState::Stop,
+ current: None,
+ } => Message::ChangedPlayback(PlayState::Stop),
+ state => {
+ log::error!("got incoherent state from the server: {state:?}");
+ Message::ChangedPlayback(PlayState::Stop)
+ }
+ })
+ .unwrap_or_default()
+ });
+
+ Command::batch([queue, history, status])
}
// Config changed
diff --git a/amadeus/src/components/mainpanel/queue.rs b/amadeus/src/components/mainpanel/queue.rs
index 8bc1d45f..8b8ef734 100644
--- a/amadeus/src/components/mainpanel/queue.rs
+++ b/amadeus/src/components/mainpanel/queue.rs
@@ -19,13 +19,13 @@ pub struct State([(bool, karalist::State); PRIORITY_LENGTH]);
/// Messages for the queue.
#[derive(Debug, Clone)]
pub enum Message {
- ClearQueue,
- ClearQueueLevel(Priority),
- ReloadQueue(Vec<(Priority, Arc<Kara>)>),
- ReloadQueueLevel(Priority, Vec<Arc<Kara>>),
- RemoveKaraFromQueue(KId),
- AddKaraToQueue(Priority, Arc<Kara>),
- ToggleQueueLevel(Priority, Option<bool>),
+ Clear,
+ ClearLevel(Priority),
+ Reload(Vec<(Priority, Arc<Kara>)>),
+ ReloadLevel(Priority, Vec<Arc<Kara>>),
+ RemoveKara(KId),
+ AddKara(Priority, Arc<Kara>),
+ ToggleLevel(Priority, Option<bool>),
}
/// Request something.
@@ -59,46 +59,46 @@ pub enum Request {
impl State {
pub fn update(&mut self, message: Message) -> Command<Request> {
match message {
- Message::ClearQueue => {
+ Message::Clear => {
self.0.iter_mut().for_each(|(show, lvl)| {
*show = false;
lvl.update(karalist::Message::Clear);
});
Command::none()
}
- Message::ClearQueueLevel(prio) => {
+ Message::ClearLevel(prio) => {
let (show, level) = &mut self.0[prio.index()];
*show = false;
level.update(karalist::Message::Clear);
Command::none()
}
- Message::ReloadQueueLevel(prio, karas) => {
+ Message::ReloadLevel(prio, karas) => {
let (show, level) = &mut self.0[prio.index()];
level.update(karalist::Message::Reload(karas));
*show = !level.is_empty();
Command::none()
}
- Message::ReloadQueue(karas) => {
+ Message::Reload(karas) => {
let get_prio = |prio: Priority| {
move |(p, kara): &(_, Arc<Kara>)| prio.eq(p).then_some(kara.clone())
};
Command::batch(PRIORITY_VALUES.iter().copied().map(|prio| {
let karas = karas.iter().filter_map(get_prio(prio));
- self.update(Message::ReloadQueueLevel(prio, karas.collect()))
+ self.update(Message::ReloadLevel(prio, karas.collect()))
}))
}
- Message::AddKaraToQueue(prio, kara) => {
+ Message::AddKara(prio, kara) => {
self.0[prio.index()].1.update(karalist::Message::Add(kara));
Command::none()
}
- Message::RemoveKaraFromQueue(id) => {
+ Message::RemoveKara(id) => {
self.0.iter_mut().for_each(|(show, lvl)| {
lvl.update(karalist::Message::RemoveId(id.clone()));
*show = !lvl.is_empty();
});
Command::none()
}
- Message::ToggleQueueLevel(prio, show) => {
+ Message::ToggleLevel(prio, show) => {
let flag = &mut self.0[prio.index()].0;
match show {
Some(show) => *flag = show,
--
GitLab