diff --git a/src/dispatcher.rs b/src/dispatcher.rs index 13edbd1432..915fc910a1 100644 --- a/src/dispatcher.rs +++ b/src/dispatcher.rs @@ -1,9 +1,11 @@ use crate::protos::protos; use lazy_static::lazy_static; +use anyhow::anyhow; use log::{info, warn}; use regex::Regex; use std::collections::HashMap; -use crate::proto_utils::get_payload_type; +use crate::proto_utils::{get_payload_type, get_twitch_message, get_twitch_metadata}; +use crate::AnyhowResult; pub trait Handler { /*fn handle_message(&self, @@ -27,7 +29,7 @@ impl Dispatcher { self.handlers.insert(command.to_string(), handler); } - pub fn handle_pubsub_event(&self, message: protos::PubsubEventPayloadV1) { + pub fn handle_pubsub_event(&self, message: protos::PubsubEventPayloadV1) -> AnyhowResult<()> { info!("Handling Proto: {:?}", message); let maybe_payload_type = get_payload_type(&message); @@ -36,16 +38,22 @@ impl Dispatcher { Some(p) => p, None => { warn!("No payload type; skipping."); - return; + return Err(anyhow!("No payload type; skipping.")); } }; match payload_type { protos::pubsub_event_payload_v1::IngestionPayloadType::TwitchMessage => { - info!("TWITCH MESSAGE!!!"); + let twitch_metadata = get_twitch_metadata(&message)?; + info!("Twitch metadata: {:?}", twitch_metadata); + + let twitch_message = get_twitch_message(&message)?; + info!("Twitch message: {:?}", twitch_message); }, _ => {}, } + + Ok(()) } } diff --git a/src/proto_utils.rs b/src/proto_utils.rs index 901ad1741f..fa70b50785 100644 --- a/src/proto_utils.rs +++ b/src/proto_utils.rs @@ -1,20 +1,52 @@ use anyhow::anyhow; use crate::AnyhowResult; use crate::protos::protos::IngestionTwitchMessage; +use crate::protos::protos::IngestionTwitchMetadata; +use crate::protos::protos::PubsubEventPayloadV1; use crate::protos::protos::pubsub_event_payload_v1::IngestionPayloadType; +use crate::protos::protos::pubsub_event_payload_v1::IngestionSourceType; use crate::protos::protos; use prost::Message; // TODO: add some type safety w/ generics and type bounds -pub fn get_payload_type(event_payload: &protos::PubsubEventPayloadV1) - -> Option +pub fn get_source_type(event_payload: &PubsubEventPayloadV1) + -> Option +{ + event_payload.ingestion_payload_type + .and_then(|num| IngestionSourceType::from_i32(num)) +} + +pub fn get_payload_type(event_payload: &PubsubEventPayloadV1) + -> Option { event_payload.ingestion_payload_type .and_then(|num| IngestionPayloadType::from_i32(num)) } -pub fn get_twitch_message(event_payload: &protos::PubsubEventPayloadV1) + +// ================= TWITCH ================= // + +// TODO: make an enum wrapper for each type, then return that +pub fn get_twitch_metadata(event_payload: &PubsubEventPayloadV1) + -> AnyhowResult +{ + let source_type = match get_source_type(event_payload) { + None => return Err(anyhow!("no source type!")), + Some(p) => p, + }; + + match source_type { + IngestionSourceType::IstTwitch => {}, + _ => return Err(anyhow!("wrong source type!")), + } + + IngestionTwitchMetadata::decode(event_payload.ingestion_source_data()) + .map_err(|_| anyhow!("error decoding source proto")) +} + +// TODO: make an enum wrapper for each type, then return that +pub fn get_twitch_message(event_payload: &PubsubEventPayloadV1) -> AnyhowResult { let payload_type = match get_payload_type(event_payload) { @@ -28,5 +60,5 @@ pub fn get_twitch_message(event_payload: &protos::PubsubEventPayloadV1) } IngestionTwitchMessage::decode(event_payload.ingestion_payload_data()) - .map_err(|_| anyhow!("error decoding proto")) + .map_err(|_| anyhow!("error decoding payload proto")) }