diff --git a/build.rs b/build.rs index c881ef718a..d9ac9fbbc4 100644 --- a/build.rs +++ b/build.rs @@ -5,7 +5,7 @@ fn main() -> anyhow::Result<()> { prost_build::compile_protos( &[ // TODO: This needs to respect glob - "protos/storyteller-protos/protos/twitch-gateway/twitch-gateway.proto", + "protos/storyteller-protos/protos/pubsub-events/pubsub-events.proto", "protos/storyteller-protos/protos/unreal/unreal.proto", ], &["protos/"])?; diff --git a/protos/storyteller-protos b/protos/storyteller-protos index e86757d405..faf4d85ac6 160000 --- a/protos/storyteller-protos +++ b/protos/storyteller-protos @@ -1 +1 @@ -Subproject commit e86757d405aab3e9c5120b62c1eae76c03b75f1e +Subproject commit faf4d85ac62186624d2c4921e42fec889f152e7a diff --git a/src/dispatcher.rs b/src/dispatcher.rs index 585502590b..887518e4ce 100644 --- a/src/dispatcher.rs +++ b/src/dispatcher.rs @@ -5,10 +5,10 @@ use regex::Regex; use std::collections::HashMap; pub trait Handler { - fn handle_message(&self, + /*fn handle_message(&self, command: &str, unparsed_command_args: &str, - twitch_message: protos::TwitchMessage); + twitch_message: protos::TwitchMessage);*/ } pub struct Dispatcher { @@ -26,13 +26,28 @@ impl Dispatcher { self.handlers.insert(command.to_string(), handler); } - pub fn handle_message(&self, twitch_message: protos::TwitchMessage) { - lazy_static! { - static ref COMMAND_REGEX : Regex = Regex::new(r"^\s*(\w+)\s+(.*)$").expect("Regex should work"); - } - info!("Handling Proto: {:?}", twitch_message); + pub fn handle_pubsub_event(&self, message: protos::PubsubEventPayloadV1) { + info!("Handling Proto: {:?}", message); - let message = match twitch_message.message_contents { + let payload_type = message.ingestion_payload_type + .and_then(|num| protos::pubsub_event_payload_v1::IngestionPayloadType::from_i32(num)); + + let payload_type = match payload_type { + Some(p) => p, + None => { + warn!("No payload type; skipping."); + return; + } + }; + + match payload_type { + protos::pubsub_event_payload_v1::IngestionPayloadType::TwitchMessage => { + + }, + _ => {}, + } + + /*let message = match twitch_message.message_contents { None => return, Some(ref m) => m.clone(), }; @@ -58,7 +73,15 @@ impl Dispatcher { if let Some(ref handler) = self.handlers.get(&command) { handler.handle_message(&command, unparsed_payload, twitch_message); + }*/ + } + + fn parse_message(&self, message: &str) -> String { + lazy_static! { + static ref COMMAND_REGEX : Regex = Regex::new(r"^\s*(\w+)\s+(.*)$").expect("Regex should work"); } + + } } diff --git a/src/handlers/coordinate_and_geocode_handler.rs b/src/handlers/coordinate_and_geocode_handler.rs index eaf9e270e1..3ed75410ea 100644 --- a/src/handlers/coordinate_and_geocode_handler.rs +++ b/src/handlers/coordinate_and_geocode_handler.rs @@ -19,7 +19,7 @@ impl CoordinateAndGeocodeHandler { } } - fn handle_lat_long(&self, lat_long: LatLong, twitch_message: protos::TwitchMessage) { + /*fn handle_lat_long(&self, lat_long: LatLong, twitch_message: protos::TwitchMessage) { let mut cesium_proto = protos::CesiumWarpRequest::default(); // Cesium @@ -51,7 +51,7 @@ impl CoordinateAndGeocodeHandler { return; } Ok(_) => { - unreal_proto.payload = buffer; + unreal_proto.payload_data = buffer; } } @@ -80,12 +80,12 @@ impl CoordinateAndGeocodeHandler { }, Err(_) => {}, } - } + }*/ } impl Handler for CoordinateAndGeocodeHandler { - /// Command payload contains the following example instruction: + /*/// Command payload contains the following example instruction: /// * "34.0659° N, 84.6769° W" /// * "33.753746, -84.386330" /// * "Obihiro, Hokkaido, Japan" @@ -100,7 +100,7 @@ impl Handler for CoordinateAndGeocodeHandler { // TODO: Geocoding. - } + }*/ } #[derive(Clone,Debug,PartialEq)] diff --git a/src/main.rs b/src/main.rs index 0935e174e3..41e3124754 100644 --- a/src/main.rs +++ b/src/main.rs @@ -3,6 +3,7 @@ extern crate serde_derive; mod dispatcher; mod handlers; +mod message_to_command_parser; mod protos; mod redis_client; mod redis_subscriber; @@ -26,13 +27,16 @@ use std::sync::{Arc, Mutex}; pub type AnyhowResult = anyhow::Result; const ENV_SUBSCRIBE_TOPIC : &'static str = "SUBSCRIBE_TOPIC"; -const ENV_SUBSCRIBE_TOPIC_DEFAULT : &'static str = "twitch"; +const ENV_SUBSCRIBE_TOPIC_DEFAULT : &'static str = "firehose"; + +const ENV_PUBLISH_TOPIC : &'static str = "PUBLISH_TOPIC"; +const ENV_PUBLISH_TOPIC_DEFAULT : &'static str = "unreal"; #[tokio::main] async fn main() -> anyhow::Result<()> { easyenv::init_env_logger(None); - let redis_publish_topic = easyenv::get_env_string_or_default( + let redis_subscribe_topic = easyenv::get_env_string_or_default( ENV_SUBSCRIBE_TOPIC, ENV_SUBSCRIBE_TOPIC_DEFAULT); let secrets = Secrets::from_file("secrets.toml")?; @@ -57,7 +61,7 @@ async fn main() -> anyhow::Result<()> { loop { redis_pubsub_client.connect().await?; - redis_pubsub_client.subscribe(&redis_publish_topic).await?; + redis_pubsub_client.subscribe(&redis_subscribe_topic).await?; redis_pubsub_client.start_stream().await?; thread::sleep(Duration::from_secs(5)); } diff --git a/src/message_to_command_parser.rs b/src/message_to_command_parser.rs new file mode 100644 index 0000000000..03adca7055 --- /dev/null +++ b/src/message_to_command_parser.rs @@ -0,0 +1,23 @@ +use crate::AnyhowResult; +use regex::Regex; + +pub struct MessageToCommandParser {} + +pub struct ParsedCommandFirstPass { + pub command: String, + pub raw_arguments: String, +} + +impl MessageToCommandParser { + + pub fn parse_raw_command(message: &str) -> AnyhowResult { + lazy_static! { + static ref COMMAND_REGEX : Regex = Regex::new(r"^\s*(\w+)\s+(.*)$").expect("Regex should work"); + } + + Ok(ParsedCommandFirstPass { + command: "".to_string(), + raw_arguments: "".to_string() + }) + } +} diff --git a/src/protos.rs b/src/protos.rs index 01e9d5e0af..6e719271fc 100644 --- a/src/protos.rs +++ b/src/protos.rs @@ -1,5 +1,5 @@ // Include the protos pub mod protos { - include!(concat!(env!("OUT_DIR"), "/storyteller.twitchgateway.rs")); + include!(concat!(env!("OUT_DIR"), "/storyteller.pubsub.rs")); include!(concat!(env!("OUT_DIR"), "/storyteller.unreal.rs")); } \ No newline at end of file diff --git a/src/redis_subscriber.rs b/src/redis_subscriber.rs index 584a7cda25..ef50d073ba 100644 --- a/src/redis_subscriber.rs +++ b/src/redis_subscriber.rs @@ -82,7 +82,7 @@ impl RedisSubscribeClient { for message in connection.on_message().next().await { let bytes = message.get_payload_bytes(); - let mut message_proto = match protos::TwitchMessage::decode(bytes) { + let mut message_proto = match protos::PubsubEventPayloadV1::decode(bytes) { Ok(m) => m, Err(e) => { warn!("Error decoding proto: {:?}", e); @@ -90,7 +90,7 @@ impl RedisSubscribeClient { }, }; - self.dispatcher.handle_message(message_proto); + self.dispatcher.handle_pubsub_event(message_proto); } thread::sleep(Duration::from_millis(250));