more proto handling

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(())
   }
 }
 

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<protos::pubsub_event_payload_v1::IngestionPayloadType>
+pub fn get_source_type(event_payload: &PubsubEventPayloadV1)
+  -> Option<IngestionSourceType>
+{
+  event_payload.ingestion_payload_type
+    .and_then(|num| IngestionSourceType::from_i32(num))
+}
+
+pub fn get_payload_type(event_payload: &PubsubEventPayloadV1)
+  -> Option<IngestionPayloadType>
 {
   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<IngestionTwitchMetadata>
+{
+  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<IngestionTwitchMessage>
 {
   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"))
 }