WIP: osu: asyncify caches

youmubot-osu/src/discord/announcer.rs

@@ -22,50 +22,61 @@ use youmubot_prelude::*;
 /// osu! announcer's unique announcer key.
 pub const ANNOUNCER_KEY: &'static str = "osu";
 
-/// Announce osu! top scores.
-pub fn updates(c: Arc<CacheAndHttp>, d: AppData, channels: MemberToChannels) -> CommandResult {
-    let osu = d.get_cloned::<OsuClient>();
-    let cache = d.get_cloned::<BeatmapMetaCache>();
-    let oppai = d.get_cloned::<BeatmapCache>();
-    // For each user...
-    let mut data = OsuSavedUsers::open(&*d.read()).borrow()?.clone();
-    for (user_id, osu_user) in data.iter_mut() {
-        let channels = channels.channels_of(c.clone(), *user_id);
-        if channels.is_empty() {
-            continue; // We don't wanna update an user without any active server
-        }
-        osu_user.pp = match (&[Mode::Std, Mode::Taiko, Mode::Catch, Mode::Mania])
-            .par_iter()
-            .map(|m| {
-                handle_user_mode(
-                    c.clone(),
-                    &osu,
-                    &cache,
-                    &oppai,
-                    &osu_user,
-                    *user_id,
-                    &channels[..],
-                    *m,
-                    d.clone(),
-                )
-            })
-            .collect::<Result<_, _>>()
-        {
-            Ok(v) => v,
-            Err(e) => {
-                eprintln!("osu: Cannot update {}: {}", osu_user.id, e.0);
-                continue;
+/// The announcer struct implementing youmubot_prelude::Announcer
+pub struct Announcer;
+
+#[async_trait]
+impl youmubot_prelude::Announcer for Announcer {
+    async fn updates(
+        &mut self,
+        c: Arc<CacheAndHttp>,
+        d: AppData,
+        channels: MemberToChannels,
+    ) -> Result<()> {
+        let d = d.read().await;
+        let osu = d.get::<OsuClient>().unwrap();
+        let cache = d.get::<BeatmapMetaCache>().unwrap();
+        let oppai = d.get::<BeatmapCache>().unwrap();
+        // For each user...
+        let mut data = OsuSavedUsers::open(&*d).borrow()?.clone();
+        for (user_id, osu_user) in data.iter_mut() {
+            let channels = channels.channels_of(c.clone(), *user_id).await;
+            if channels.is_empty() {
+                continue; // We don't wanna update an user without any active server
             }
-        };
-        osu_user.last_update = chrono::Utc::now();
+            osu_user.pp = match (&[Mode::Std, Mode::Taiko, Mode::Catch, Mode::Mania])
+                .par_iter()
+                .map(|m| {
+                    handle_user_mode(
+                        c.clone(),
+                        &osu,
+                        &cache,
+                        &oppai,
+                        &osu_user,
+                        *user_id,
+                        &channels[..],
+                        *m,
+                        &*d,
+                    )
+                })
+                .collect::<Result<_, _>>()
+            {
+                Ok(v) => v,
+                Err(e) => {
+                    eprintln!("osu: Cannot update {}: {}", osu_user.id, e.0);
+                    continue;
+                }
+            };
+            osu_user.last_update = chrono::Utc::now();
+        }
+        // Update users
+        *OsuSavedUsers::open(&*d.read()).borrow_mut()? = data;
+        Ok(())
     }
-    // Update users
-    *OsuSavedUsers::open(&*d.read()).borrow_mut()? = data;
-    Ok(())
 }
 
 /// Handles an user/mode scan, announces all possible new scores, return the new pp value.
-fn handle_user_mode(
+async fn handle_user_mode(
     c: Arc<CacheAndHttp>,
     osu: &Osu,
     cache: &BeatmapMetaCache,
@@ -74,15 +85,16 @@ fn handle_user_mode(
     user_id: UserId,
     channels: &[ChannelId],
     mode: Mode,
-    d: AppData,
+    d: &TypeMap,
 ) -> Result<Option<f64>, Error> {
     let scores = scan_user(osu, osu_user, mode)?;
     let user = osu
-        .user(UserID::ID(osu_user.id), |f| f.mode(mode))?
+        .user(UserID::ID(osu_user.id), |f| f.mode(mode))
+        .await?
         .ok_or(Error::from("user not found"))?;
     scores
-        .into_par_iter()
-        .map(|(rank, score)| -> Result<_, Error> {
+        .into_iter()
+        .map(|(rank, score)| -> Result<_> {
             let beatmap = cache.get_beatmap_default(score.beatmap_id)?;
             let content = oppai.get_beatmap(beatmap.beatmap_id)?;
             Ok((rank, score, BeatmapWithMode(beatmap, mode), content))

youmubot-osu/src/discord/beatmap_cache.rs

@@ -3,16 +3,14 @@ use crate::{
     Client,
 };
 use dashmap::DashMap;
-use serenity::framework::standard::CommandError;
-use std::sync::Arc;
-use youmubot_prelude::TypeMapKey;
+use youmubot_prelude::*;
 
 /// BeatmapMetaCache intercepts beatmap-by-id requests and caches them for later recalling.
 /// Does not cache non-Ranked beatmaps.
-#[derive(Clone, Debug)]
+#[derive(Debug)]
 pub struct BeatmapMetaCache {
     client: Client,
-    cache: Arc<DashMap<(u64, Mode), Beatmap>>,
+    cache: DashMap<(u64, Mode), Beatmap>,
 }
 
 impl TypeMapKey for BeatmapMetaCache {
@@ -24,10 +22,10 @@ impl BeatmapMetaCache {
     pub fn new(client: Client) -> Self {
         BeatmapMetaCache {
             client,
-            cache: Arc::new(DashMap::new()),
+            cache: DashMap::new(),
         }
     }
-    fn insert_if_possible(&self, id: u64, mode: Option<Mode>) -> Result<Beatmap, CommandError> {
+    async fn insert_if_possible(&self, id: u64, mode: Option<Mode>) -> Result<Beatmap> {
         let beatmap = self
             .client
             .beatmaps(crate::BeatmapRequestKind::Beatmap(id), |f| {
@@ -36,35 +34,37 @@ impl BeatmapMetaCache {
                 }
                 f
             })
-            .and_then(|v| {
-                v.into_iter()
-                    .next()
-                    .ok_or(CommandError::from("beatmap not found"))
-            })?;
+            .await
+            .and_then(|v| v.into_iter().next().ok_or(Error::msg("beatmap not found")))?;
         if let ApprovalStatus::Ranked(_) = beatmap.approval {
             self.cache.insert((id, beatmap.mode), beatmap.clone());
         };
         Ok(beatmap)
     }
     /// Get the given beatmap
-    pub fn get_beatmap(&self, id: u64, mode: Mode) -> Result<Beatmap, CommandError> {
-        self.cache
-            .get(&(id, mode))
-            .map(|b| Ok(b.clone()))
-            .unwrap_or_else(|| self.insert_if_possible(id, Some(mode)))
+    pub async fn get_beatmap(&self, id: u64, mode: Mode) -> Result<Beatmap> {
+        match self.cache.get(&(id, mode)).map(|v| v.clone()) {
+            Some(v) => Ok(v),
+            None => self.insert_if_possible(id, Some(mode)).await,
+        }
     }
 
     /// Get a beatmap without a mode...
-    pub fn get_beatmap_default(&self, id: u64) -> Result<Beatmap, CommandError> {
-        (&[Mode::Std, Mode::Taiko, Mode::Catch, Mode::Mania])
-            .iter()
-            .filter_map(|&mode| {
-                self.cache
-                    .get(&(id, mode))
-                    .filter(|b| b.mode == mode)
-                    .map(|b| Ok(b.clone()))
-            })
-            .next()
-            .unwrap_or_else(|| self.insert_if_possible(id, None))
+    pub async fn get_beatmap_default(&self, id: u64) -> Result<Beatmap> {
+        Ok(
+            match (&[Mode::Std, Mode::Taiko, Mode::Catch, Mode::Mania])
+                .iter()
+                .filter_map(|&mode| {
+                    self.cache
+                        .get(&(id, mode))
+                        .filter(|b| b.mode == mode)
+                        .map(|b| b.clone())
+                })
+                .next()
+            {
+                Some(v) => v,
+                None => self.insert_if_possible(id, None).await?,
+            },
+        )
     }
 }