Merge branch 'reaction-watcher'

Cargo.lock

@@ -183,6 +183,14 @@ dependencies = [
  "cfg-if 0.1.10 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
+[[package]]
+name = "crossbeam-channel"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "crossbeam-utils 0.7.0 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
 [[package]]
 name = "crossbeam-deque"
 version = "0.7.2"
@@ -1731,6 +1739,7 @@ dependencies = [
 name = "youmubot-prelude"
 version = "0.1.0"
 dependencies = [
+ "crossbeam-channel 0.4.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "reqwest 0.10.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "serenity 0.8.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "youmubot-db 0.1.0",
@@ -1763,6 +1772,7 @@ dependencies = [
 "checksum core-foundation 0.6.4 (registry+https://github.com/rust-lang/crates.io-index)" = "25b9e03f145fd4f2bf705e07b900cd41fc636598fe5dc452fd0db1441c3f496d"
 "checksum core-foundation-sys 0.6.2 (registry+https://github.com/rust-lang/crates.io-index)" = "e7ca8a5221364ef15ce201e8ed2f609fc312682a8f4e0e3d4aa5879764e0fa3b"
 "checksum crc32fast 1.2.0 (registry+https://github.com/rust-lang/crates.io-index)" = "ba125de2af0df55319f41944744ad91c71113bf74a4646efff39afe1f6842db1"
+"checksum crossbeam-channel 0.4.0 (registry+https://github.com/rust-lang/crates.io-index)" = "acec9a3b0b3559f15aee4f90746c4e5e293b701c0f7d3925d24e01645267b68c"
 "checksum crossbeam-deque 0.7.2 (registry+https://github.com/rust-lang/crates.io-index)" = "c3aa945d63861bfe624b55d153a39684da1e8c0bc8fba932f7ee3a3c16cea3ca"
 "checksum crossbeam-epoch 0.8.0 (registry+https://github.com/rust-lang/crates.io-index)" = "5064ebdbf05ce3cb95e45c8b086f72263f4166b29b97f6baff7ef7fe047b55ac"
 "checksum crossbeam-queue 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)" = "c695eeca1e7173472a32221542ae469b3e9aac3a4fc81f7696bcad82029493db"

youmubot-core/src/community/votes.rs

@@ -2,13 +2,12 @@ use serenity::framework::standard::CommandError as Error;
 use serenity::{
     framework::standard::{macros::command, Args, CommandResult},
     model::{
-        channel::{Message, MessageReaction, ReactionType},
+        channel::{Message,Reaction, ReactionType},
         id::UserId,
     },
     utils::MessageBuilder,
 };
 use std::collections::HashMap as Map;
-use std::thread;
 use std::time::Duration;
 use youmubot_prelude::{Duration as ParseDuration, *};
 
@@ -93,10 +92,50 @@ pub fn vote(ctx: &mut Context, msg: &Message, mut args: Args) -> CommandResult {
         .iter()
         .try_for_each(|(v, _)| panel.react(&ctx, *v))?;
 
-    // Start sleeping
-    thread::sleep(*duration);
+    let reaction_to_choice: Map<_, _> = choices.iter().map(|r| (r.0, &r.1)).collect();
+    let mut user_reactions: Map<UserId, Vec<&str>> = Map::new();
+
+    ctx.data.get_cloned::<ReactionWatcher>().handle_reactions(
+        |reaction: &Reaction, is_add| {
+            if reaction.message_id != panel.id {
+                return Ok(());
+            }
+            if reaction.user(&ctx)?.bot {
+                return Ok(());
+            }
+            let choice = if let ReactionType::Unicode(ref s) = reaction.emoji {
+                if let Some(choice) = reaction_to_choice.get(s.as_str()) {
+                    choice
+                } else {
+                    return Ok(());
+                }
+            } else {
+                return Ok(());
+            };
+            if is_add {
+                user_reactions
+                    .entry(reaction.user_id)
+                    .or_default()
+                    .push(choice);
+            } else {
+                user_reactions.entry(reaction.user_id).and_modify(|v| {
+                    v.retain(|f| &f != choice);
+                });
+            }
+            Ok(())
+        },
+        *duration,
+    )?;
+    let result: Vec<(&str, Vec<UserId>)> = {
+        let mut res: Map<&str, Vec<UserId>> = Map::new();
+        for (u, r) in user_reactions {
+            for t in r {
+                res.entry(t).or_default().push(u);
+            }
+        }
+        res.into_iter().collect()
+    };
 
-    let result = collect_reactions(ctx, &panel, &choices)?;
     if result.len() == 0 {
         msg.reply(
             &ctx,
@@ -141,61 +180,6 @@ pub fn vote(ctx: &mut Context, msg: &Message, mut args: Args) -> CommandResult {
     // unimplemented!();
 }
 
-// Collect reactions and store them as a map from choice to
-fn collect_reactions<'a>(
-    ctx: &mut Context,
-    msg: &Message,
-    choices: &'a [(&'static str, String)],
-) -> Result<Vec<(&'a str, Vec<UserId>)>, Error> {
-    // Get a brand new version of the Message
-    let reactions = msg.channel_id.message(&ctx, msg.id)?.reactions;
-    let reaction_to_choice: Map<_, _> = choices.into_iter().map(|r| (r.0, &r.1)).collect();
-    let mut vec: Vec<(&str, Vec<UserId>)> = Vec::new();
-    reactions
-        .into_iter()
-        .filter_map(|r| {
-            if let ReactionType::Unicode(ref v) = r.reaction_type {
-                reaction_to_choice
-                    .get(&&v[..])
-                    .cloned()
-                    .filter(|_| r.count > 1)
-                    .map(|choice| (r.clone(), choice))
-            } else {
-                None
-            }
-        })
-        .try_for_each(|(r, choice)| -> Result<_, Error> {
-            let users = collect_reaction_users(ctx, &msg, &r)?;
-            vec.push((choice, users));
-            Ok(())
-        })?;
-    vec.sort_by(|(_, b): &(_, Vec<_>), (_, d)| d.len().cmp(&b.len()));
-    Ok(vec)
-}
-
-fn collect_reaction_users(
-    ctx: &mut Context,
-    msg: &Message,
-    reaction: &MessageReaction,
-) -> Result<Vec<UserId>, Error> {
-    let mut res = Vec::with_capacity(reaction.count as usize);
-    (0..reaction.count)
-        .step_by(100)
-        .try_for_each(|_| -> Result<_, Error> {
-            let user_ids = msg
-                .reaction_users(
-                    &ctx,
-                    reaction.reaction_type.clone(),
-                    Some(100),
-                    res.last().cloned(),
-                )?
-                .into_iter()
-                .map(|i| i.id);
-            res.extend(user_ids);
-            Ok(())
-        })?;
-    Ok(res)
-}
 // Pick a set of random n reactions!
 fn pick_n_reactions(n: usize) -> Result<Vec<&'static str>, Error> {
     use rand::seq::SliceRandom;

youmubot-prelude/Cargo.toml

@@ -9,4 +9,5 @@ edition = "2018"
 [dependencies]
 serenity = "0.8"
 youmubot-db = { path = "../youmubot-db" }
+crossbeam-channel = "0.4"
 reqwest = "0.10"

youmubot-prelude/src/lib.rs

@@ -3,10 +3,12 @@ use std::sync::Arc;
 
 pub mod announcer;
 pub mod args;
+pub mod reaction_watch;
 pub mod setup;
 
 pub use announcer::Announcer;
 pub use args::Duration;
+pub use reaction_watch::{ReactionHandler, ReactionWatcher};
 
 /// The global app data.
 pub type AppData = Arc<RwLock<ShareMap>>;

youmubot-prelude/src/reaction_watch.rs

@@ -0,0 +1,75 @@
+use crossbeam_channel::{after, bounded, select, Sender};
+use serenity::{framework::standard::CommandResult, model::channel::Reaction, prelude::*};
+use std::sync::{Arc, Mutex};
+
+/// Handles a reaction.
+///
+/// Every handler needs an expire time too.
+pub trait ReactionHandler {
+    /// Handle a reaction. This is fired on EVERY reaction.
+    /// You do the filtering yourself.
+    ///
+    /// If `is_added` is false, the reaction was removed instead of added.
+    fn handle_reaction(&mut self, reaction: &Reaction, is_added: bool) -> CommandResult;
+}
+
+impl<T> ReactionHandler for T
+where
+    T: FnMut(&Reaction, bool) -> CommandResult,
+{
+    fn handle_reaction(&mut self, reaction: &Reaction, is_added: bool) -> CommandResult {
+        self(reaction, is_added)
+    }
+}
+
+/// The store for a set of dynamic reaction handlers.
+#[derive(Debug, Clone)]
+pub struct ReactionWatcher {
+    channels: Arc<Mutex<Vec<Sender<(Arc<Reaction>, bool)>>>>,
+}
+
+impl TypeMapKey for ReactionWatcher {
+    type Value = ReactionWatcher;
+}
+
+impl ReactionWatcher {
+    /// Create a new ReactionWatcher.
+    pub fn new() -> Self {
+        Self {
+            channels: Arc::new(Mutex::new(vec![])),
+        }
+    }
+    /// Send a reaction.
+    /// If `is_added` is false, the reaction was removed.
+    pub fn send(&self, r: Reaction, is_added: bool) {
+        let r = Arc::new(r);
+        self.channels
+            .lock()
+            .expect("Poisoned!")
+            .retain(|e| e.send((r.clone(), is_added)).is_ok());
+    }
+    /// React! to a series of reaction
+    ///
+    /// The reactions stop after `duration`.
+    pub fn handle_reactions(
+        &self,
+        mut h: impl ReactionHandler,
+        duration: std::time::Duration,
+    ) -> CommandResult {
+        let (send, reactions) = bounded(0);
+        {
+            self.channels.lock().expect("Poisoned!").push(send);
+        }
+        let timeout = after(duration);
+        loop {
+            let r = select! {
+                recv(reactions) -> r => { let (r, is_added) = r.unwrap(); h.handle_reaction(&*r, is_added) },
+                recv(timeout) -> _ => break,
+            };
+            if let Err(v) = r {
+                return Err(v);
+            }
+        }
+        Ok(())
+    }
+}

youmubot-prelude/src/setup.rs

@@ -9,5 +9,9 @@ pub fn setup_prelude(db_path: &Path, data: &mut ShareMap, _: &mut StandardFramew
     crate::announcer::AnnouncerChannels::insert_into(data, db_path.join("announcers.yaml"))
         .expect("Announcers DB set up");
 
+    // Set up the HTTP client.
     data.insert::<crate::HTTPClient>(reqwest::blocking::Client::new());
+
+    // Set up the reaction watcher.
+    data.insert::<crate::ReactionWatcher>(crate::ReactionWatcher::new());
 }

youmubot/src/main.rs

@@ -2,7 +2,10 @@ use dotenv;
 use dotenv::var;
 use serenity::{
     framework::standard::{DispatchError, StandardFramework},
-    model::{channel::Message, gateway},
+    model::{
+        channel::{Message, Reaction},
+        gateway,
+    },
 };
 use youmubot_prelude::*;
 
@@ -25,6 +28,18 @@ impl EventHandler for Handler {
         println!("{:?}", message);
         self.hooks.iter().for_each(|f| f(&mut ctx, &message));
     }
+
+    fn reaction_add(&self, ctx: Context, reaction: Reaction) {
+        ctx.data
+            .get_cloned::<ReactionWatcher>()
+            .send(reaction, true);
+    }
+
+    fn reaction_remove(&self, ctx: Context, reaction: Reaction) {
+        ctx.data
+            .get_cloned::<ReactionWatcher>()
+            .send(reaction, false);
+    }
 }
 
 fn main() {