WIP on pubsub enricher port to better protos

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/"])?;

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");
     }
+
+
   }
 }
 

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)]

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<T> = anyhow::Result<T>;
 
 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));
   }

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<ParsedCommandFirstPass> {
+    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()
+    })
+  }
+}

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"));
 }

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));