From 2bacf887c0f3aff01d60dd1255b2d0e673f550ea Mon Sep 17 00:00:00 2001 From: ashelyn ghost Date: Mon, 8 Jul 2024 20:36:06 -0600 Subject: Separate out listener and event handling --- src/system.rs | 130 +++++++++++++++++++--------------------------------------- 1 file changed, 43 insertions(+), 87 deletions(-) (limited to 'src/system.rs') diff --git a/src/system.rs b/src/system.rs index 77df36a..9a8b327 100644 --- a/src/system.rs +++ b/src/system.rs @@ -1,126 +1,82 @@ -use std::{num::NonZeroUsize, str::FromStr}; +use std::{collections::HashMap, num::NonZeroUsize, str::FromStr}; -use std::sync::Arc; -use tokio::task::JoinSet; -use tokio::sync::Mutex; +use tokio::sync::mpsc::channel; use twilight_http::Client; use twilight_model::id::{Id, marker::{MessageMarker, UserMarker}}; -use twilight_gateway::{Intents, Shard, ShardId}; use twilight_model::util::Timestamp; -#[derive(Clone)] +use crate::listener::{Listener, ClientEvent}; + pub struct System { + pub name: String, pub config: crate::config::System, - pub message_dedup_cache: Arc, Timestamp>>>, + pub message_dedup_cache: lru::LruCache, Timestamp>, + pub clients: HashMap, } impl System { - pub fn new(system_config: crate::config::System) -> Self { + pub fn new(system_name: String, system_config: crate::config::System) -> Self { System { + name: system_name, config: system_config, - message_dedup_cache: Arc::new(Mutex::new(lru::LruCache::new(NonZeroUsize::new(100).unwrap()))) + message_dedup_cache: lru::LruCache::new(NonZeroUsize::new(100).unwrap()), + clients: HashMap::new(), } } pub async fn start_clients(&mut self) { - println!("Starting clients for system"); + println!("Starting clients for system {}", self.name); + + let (tx, mut rx) = channel::(100); let reference_user_id : Id = Id::from_str(self.config.reference_user_id.as_str()) .expect(format!("Invalid user ID: {}", self.config.reference_user_id).as_str()); - let mut set = JoinSet::new(); - for member in self.config.members.iter() { - let token = member.discord_token.clone(); - let self_clone = self.clone(); - set.spawn(async move { - println!("Starting client with token: {}", token); - - let intents = Intents::GUILD_MEMBERS | Intents::GUILD_PRESENCES | Intents::GUILD_MESSAGES | Intents::MESSAGE_CONTENT; - - let mut shard = Shard::new(ShardId::ONE, token.clone(), intents); - let mut client = Client::new(token.clone()); - - loop { - let event = match shard.next_event().await { - Ok(event) => event, - Err(source) => { - println!("error receiving event"); - - if source.is_fatal() { - break; - } - - continue; - } - }; - - match event { - twilight_gateway::Event::Ready(client) => { - println!("Bot started for {}#{}", client.user.name, client.user.discriminator); - }, - - twilight_gateway::Event::MessageCreate(message) => { - if message.author.id != reference_user_id { - continue - } - - if self_clone.is_new_message(message.id, message.timestamp).await { - self_clone.handle_message_create(message, &mut client).await; - } - }, - - twilight_gateway::Event::MessageUpdate(message) => { - if message.author.is_none() || message.author.as_ref().unwrap().id != reference_user_id { - continue - } - - if self_clone.is_new_message(message.id, message.edited_timestamp.unwrap_or(message.timestamp.expect("No message creation timestamp"))).await { - self_clone.handle_message_update(message, &mut client).await; - } - }, - - _ => (), - } - } + let client = twilight_http::Client::new(member.discord_token.clone()); + self.clients.insert(member.name.clone(), client); + + let tx = tx.clone(); + let member = member.clone(); + tokio::spawn(async move { + let mut listener = Listener::new(member, reference_user_id); + listener.start_listening(tx).await; }); } - while let Some(join_result) = set.join_next().await { - if let Err(join_error) = join_result { - println!("Task encountered error: {}", join_error); - } else { - println!("Task joined cleanly"); + loop { + match rx.recv().await { + Some(event) => match event { + ClientEvent::Message { event_time, message_id, content, author: _ } => { + if self.is_new_message(message_id, event_time) { + self.handle_message(content).await; + } + } + ClientEvent::Error(_err) => { + println!("Client ran into an error for system {}", self.name); + return + }, + }, + None => { + return + }, } } } - async fn is_new_message(&self, message_id: Id, timestamp: Timestamp) -> bool { - let mut message_cache = self.message_dedup_cache.lock().await; - - let last_seen_timestamp = message_cache.get(&message_id); + fn is_new_message(&mut self, message_id: Id, timestamp: Timestamp) -> bool { + let last_seen_timestamp = self.message_dedup_cache.get(&message_id); let current_timestamp = timestamp; if last_seen_timestamp.is_none() || last_seen_timestamp.unwrap().as_micros() < current_timestamp.as_micros() { - message_cache.put(message_id, timestamp); + self.message_dedup_cache.put(message_id, timestamp); true } else { false } } - async fn handle_message_create(&self, message: Box, client: &mut Client) { - println!("Message created: {}", message.content); - - if let Err(err) = client.create_message(message.channel_id) - .reply(message.id) - .content(&format!("Recognized message from authorized user {}", message.author.name)).expect("Error: reply too long") - .await { - println!("Error: {}", err); - } - } - - async fn handle_message_update(&self, message: Box, _client: &mut Client) { - println!("Message updated: {}", message.content.unwrap_or("".to_string())) + async fn handle_message(&mut self, content: String) { + println!("Message: {}", content); } } -- cgit 1.4.1