From 982a4ea0f6dc9c8fb92da1dc33eaf5b82da0103b Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Wed, 28 Apr 2021 23:23:05 -0400 Subject: [PATCH] wip dispatcher --- Cargo.lock | 1 + Cargo.toml | 1 + src/dispatcher.rs | 39 +++++++++++++++++++++++++++++++++++++++ src/main.rs | 30 ++++++++++++++++++------------ src/redis_subscriber.rs | 9 +++++++-- 5 files changed, 66 insertions(+), 14 deletions(-) create mode 100644 src/dispatcher.rs diff --git a/Cargo.lock b/Cargo.lock index 3cc975206b..0aac66c347 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1158,6 +1158,7 @@ dependencies = [ "anyhow", "easyenv", "google_maps", + "lazy_static", "log", "prost", "prost-build", diff --git a/Cargo.toml b/Cargo.toml index e738d52881..b2fc29ce31 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,6 +14,7 @@ path = "src/main.rs" [dependencies] anyhow = "1.0.38" easyenv = "0.0.4" +lazy_static = "1.4.0" log = "0.4.14" prost = "0.7.0" regex = "1.4.6" diff --git a/src/dispatcher.rs b/src/dispatcher.rs new file mode 100644 index 0000000000..d44370a705 --- /dev/null +++ b/src/dispatcher.rs @@ -0,0 +1,39 @@ +use crate::protos::protos; +use lazy_static::lazy_static; +use log::{info, warn}; +use regex::Regex; + +pub struct Dispatcher { +} + +impl Dispatcher { + pub fn new() -> Self { + Self {} + } + + pub fn handle_message(&self, message: protos::TwitchMessage) { + lazy_static! { + static ref COMMAND_REGEX : Regex = Regex::new(r"^\s*(\w+)\s+(.*)$").expect("Regex should work"); + } + info!("Handling Proto: {:?}", message); + + //for cap in re.captures_iter(text) { + // println!("Month: {} Day: {} Year: {}", &cap[2], &cap[3], &cap[1]); + //} + + let message_ref = match message.message_contents { + None => return, + Some(ref m) => m, + }; + + let captures = COMMAND_REGEX.captures(&message_ref).unwrap(); + info!("Captures: {:?}", captures); + } +} + +#[cfg(test)] +mod tests { + #[test] + fn test_types() { + } +} diff --git a/src/main.rs b/src/main.rs index 73d836b378..845ff04f7f 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,25 +1,25 @@ #[macro_use] extern crate serde_derive; -use anyhow::{Context, Error}; -use anyhow::anyhow; - -use redis::aio::Connection; -use redis::{AsyncCommands, RedisResult}; - -pub type AnyhowResult = anyhow::Result; - -use std::thread; -use std::time::Duration; - +mod dispatcher; mod protos; mod redis_client; mod redis_subscriber; mod secrets; +use anyhow::anyhow; +use anyhow::{Context, Error}; +use log::{warn, info}; +use redis::aio::Connection; +use redis::{AsyncCommands, RedisResult}; +use std::thread; +use std::time::Duration; use crate::secrets::Secrets; use crate::redis_client::RedisClient; use crate::redis_subscriber::RedisSubscribeClient; +use crate::dispatcher::Dispatcher; + +pub type AnyhowResult = anyhow::Result; const ENV_SUBSCRIBE_TOPIC : &'static str = "SUBSCRIBE_TOPIC"; const ENV_SUBSCRIBE_TOPIC_DEFAULT : &'static str = "twitch"; @@ -33,7 +33,13 @@ async fn main() -> anyhow::Result<()> { let secrets = Secrets::from_file("secrets.toml")?; - let mut redis_client = RedisSubscribeClient::new(&secrets.redis); + let dispatcher = Dispatcher::new(); + + let mut redis_client = RedisSubscribeClient::new( + &secrets.redis, + dispatcher + ); + redis_client.connect().await?; loop { diff --git a/src/redis_subscriber.rs b/src/redis_subscriber.rs index c2674233e8..584a7cda25 100644 --- a/src/redis_subscriber.rs +++ b/src/redis_subscriber.rs @@ -1,5 +1,6 @@ use anyhow::anyhow; use crate::AnyhowResult; +use crate::dispatcher::Dispatcher; use crate::protos::protos; use crate::secrets::RedisSecrets; use log::{info, warn}; @@ -18,14 +19,18 @@ pub struct RedisSubscribeClient { connection: Option, connection_failure_count: u32, + + /// Event dispatcher that handles routing and passing to handlers. + dispatcher: Dispatcher, } impl RedisSubscribeClient { - pub fn new(secrets: &RedisSecrets) -> Self { + pub fn new(secrets: &RedisSecrets, dispatcher: Dispatcher) -> Self { Self { secrets: secrets.clone(), connection: None, connection_failure_count: 0, + dispatcher, } } @@ -85,7 +90,7 @@ impl RedisSubscribeClient { }, }; - info!("Proto: {:?}", message_proto); + self.dispatcher.handle_message(message_proto); } thread::sleep(Duration::from_millis(250));