From ef937d8097230fbb02ce8834db60db8137d6a04a Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Wed, 28 Apr 2021 04:49:23 -0400 Subject: [PATCH] break out redis pubsub, add google geocoder crate --- Cargo.lock | 447 +++++++++++++++++++++++++++++++++++++++- Cargo.toml | 3 + src/main.rs | 4 +- src/redis_client.rs | 85 -------- src/redis_subscriber.rs | 96 +++++++++ src/secrets.rs | 6 + 6 files changed, 545 insertions(+), 96 deletions(-) create mode 100644 src/redis_subscriber.rs diff --git a/Cargo.lock b/Cargo.lock index 3e62019c90..3cc975206b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -17,6 +17,12 @@ version = "1.0.40" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "28b2cd92db5cbd74e8e5028f7e27dd7aa3090e89e4f2a197cc7c8dfb69c7063b" +[[package]] +name = "arrayvec" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "23b62fc65de8e4e7f52534fb52b0f3ed04746ae267519eef2a83941e8085068b" + [[package]] name = "async-channel" version = "1.6.1" @@ -181,6 +187,12 @@ version = "0.12.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3441f0f7b02788e948e47f457ca01f1d7e6d92c693bc132c22b087d3141c03ff" +[[package]] +name = "base64" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "904dfeac50f3cdaba28fc6f57fdcddb75f49ed61346676a78c4ffe55877802fd" + [[package]] name = "bitflags" version = "1.2.1" @@ -243,6 +255,31 @@ version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd" +[[package]] +name = "chrono" +version = "0.4.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "670ad68c9088c2a963aaa298cb369688cf3f9465ce5e2d4ca10e6e0098a1ce73" +dependencies = [ + "libc", + "num-integer", + "num-traits", + "serde", + "time", + "winapi 0.3.9", +] + +[[package]] +name = "chrono-tz" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2554a3155fec064362507487171dcc4edc3df60cb10f3a1fb10ed8094822b120" +dependencies = [ + "chrono", + "parse-zoneinfo", + "serde", +] + [[package]] name = "combine" version = "4.5.2" @@ -254,7 +291,7 @@ dependencies = [ "futures-util", "memchr", "pin-project-lite 0.2.6", - "tokio", + "tokio 0.2.25", ] [[package]] @@ -325,6 +362,15 @@ version = "1.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e78d4f1cc4ae33bbfc157ed5d5a5ef3bc29227303d595861deb238fcec4e9457" +[[package]] +name = "encoding_rs" +version = "0.8.28" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "80df024fbc5ac80f87dfef0d9f5209a252f2a497f7f42944cff24d8253cac065" +dependencies = [ + "cfg-if 1.0.0", +] + [[package]] name = "env_logger" version = "0.7.1" @@ -494,6 +540,43 @@ dependencies = [ "web-sys", ] +[[package]] +name = "google_maps" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb31e07d9172813158c681923ac044846a5fad9d1fbba035f87851af45b4235a" +dependencies = [ + "chrono", + "chrono-tz", + "log", + "percent-encoding", + "rand", + "reqwest", + "rust_decimal", + "rust_decimal_macros", + "serde", + "serde_json", +] + +[[package]] +name = "h2" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc018e188373e2777d0ef2467ebff62a08e66c3f5857b23c8fbec3018210dc00" +dependencies = [ + "bytes 1.0.1", + "fnv", + "futures-core", + "futures-sink", + "futures-util", + "http", + "indexmap", + "slab", + "tokio 1.5.0", + "tokio-util 0.6.6", + "tracing", +] + [[package]] name = "hashbrown" version = "0.9.1" @@ -518,6 +601,40 @@ dependencies = [ "libc", ] +[[package]] +name = "http" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "527e8c9ac747e28542699a951517aa9a6945af506cd1f2e1b53a576c17b6cc11" +dependencies = [ + "bytes 1.0.1", + "fnv", + "itoa", +] + +[[package]] +name = "http-body" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5dfb77c123b4e2f72a2069aeae0b4b4949cc7e966df277813fc16347e7549737" +dependencies = [ + "bytes 1.0.1", + "http", + "pin-project-lite 0.2.6", +] + +[[package]] +name = "httparse" +version = "1.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4a1ce40d6fc9764887c2fdc7305c3dcc429ba11ff981c1509416afd5697e4437" + +[[package]] +name = "httpdate" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05842d0d43232b23ccb7060ecb0f0626922c21f30012e97b767b30afd4a5d4b9" + [[package]] name = "humantime" version = "1.3.0" @@ -527,6 +644,43 @@ dependencies = [ "quick-error", ] +[[package]] +name = "hyper" +version = "0.14.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e5f105c494081baa3bf9e200b279e27ec1623895cd504c7dbef8d0b080fcf54" +dependencies = [ + "bytes 1.0.1", + "futures-channel", + "futures-core", + "futures-util", + "h2", + "http", + "http-body", + "httparse", + "httpdate", + "itoa", + "pin-project", + "socket2", + "tokio 1.5.0", + "tower-service", + "tracing", + "want", +] + +[[package]] +name = "hyper-tls" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6183ddfa99b85da61a140bea0efc93fdf56ceaa041b37d553518030827f9905" +dependencies = [ + "bytes 1.0.1", + "hyper", + "native-tls", + "tokio 1.5.0", + "tokio-native-tls", +] + [[package]] name = "idna" version = "0.2.3" @@ -566,6 +720,12 @@ dependencies = [ "libc", ] +[[package]] +name = "ipnet" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47be2f14c678be2fdcab04ab1171db51b2762ce6f0a8ee87c8dd4a04ed216135" + [[package]] name = "itertools" version = "0.9.0" @@ -643,6 +803,12 @@ version = "2.3.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0ee1c47aaa256ecabcaea351eae4a9b01ef39ed810004e298d2511ed284b1525" +[[package]] +name = "mime" +version = "0.3.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a60c7ce501c71e03a9c9c0d35b861413ae925bd979cc7a4e30d060069aaac8d" + [[package]] name = "mio" version = "0.6.23" @@ -662,6 +828,19 @@ dependencies = [ "winapi 0.2.8", ] +[[package]] +name = "mio" +version = "0.7.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf80d3e903b34e0bd7282b218398aec54e082c840d9baf8339e0080a0c542956" +dependencies = [ + "libc", + "log", + "miow 0.3.7", + "ntapi", + "winapi 0.3.9", +] + [[package]] name = "mio-named-pipes" version = "0.1.7" @@ -669,7 +848,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0840c1c50fd55e521b247f949c241c9997709f23bd7f023b9762cd561e935656" dependencies = [ "log", - "mio", + "mio 0.6.23", "miow 0.3.7", "winapi 0.3.9", ] @@ -682,7 +861,7 @@ checksum = "afcb699eb26d4332647cc848492bbc15eafb26f08d0304550d5aa1f612e066f0" dependencies = [ "iovec", "libc", - "mio", + "mio 0.6.23", ] [[package]] @@ -741,6 +920,34 @@ dependencies = [ "winapi 0.3.9", ] +[[package]] +name = "ntapi" +version = "0.3.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f6bb902e437b6d86e03cce10a7e2af662292c5dfef23b65899ea3ac9354ad44" +dependencies = [ + "winapi 0.3.9", +] + +[[package]] +name = "num-integer" +version = "0.1.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2cc698a63b549a70bc047073d2949cce27cd1c7b0a4a862d08a8031bc2801db" +dependencies = [ + "autocfg", + "num-traits", +] + +[[package]] +name = "num-traits" +version = "0.2.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a64b1ec5cda2586e284722486d802acf1f7dbdc623e2bfc57e65ca1cd099290" +dependencies = [ + "autocfg", +] + [[package]] name = "num_cpus" version = "1.13.0" @@ -796,6 +1003,15 @@ version = "2.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "427c3892f9e783d91cc128285287e70a59e206ca452770ece88a76f7a3eddd72" +[[package]] +name = "parse-zoneinfo" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c705f256449c60da65e11ff6626e0c16a0a0b96aaa348de61376b249bc340f41" +dependencies = [ + "regex", +] + [[package]] name = "percent-encoding" version = "2.1.0" @@ -812,6 +1028,26 @@ dependencies = [ "indexmap", ] +[[package]] +name = "pin-project" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7509cc106041c40a4518d2af7a61530e1eed0e6285296a3d8c5472806ccc4a4" +dependencies = [ + "pin-project-internal", +] + +[[package]] +name = "pin-project-internal" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "48c950132583b500556b1efd71d45b319029f2b71518d979fcc208e16b42426f" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "pin-project-lite" version = "0.1.12" @@ -921,14 +1157,16 @@ version = "0.0.0" dependencies = [ "anyhow", "easyenv", + "google_maps", "log", "prost", "prost-build", "redis", + "regex", "rustls", "serde", "serde_derive", - "tokio", + "tokio 0.2.25", "toml", ] @@ -1005,9 +1243,9 @@ dependencies = [ "percent-encoding", "pin-project-lite 0.1.12", "sha1", - "tokio", + "tokio 0.2.25", "tokio-tls", - "tokio-util", + "tokio-util 0.3.1", "url", ] @@ -1046,6 +1284,40 @@ dependencies = [ "winapi 0.3.9", ] +[[package]] +name = "reqwest" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2296f2fac53979e8ccbc4a1136b25dcefd37be9ed7e4a1f6b05a6029c84ff124" +dependencies = [ + "base64 0.13.0", + "bytes 1.0.1", + "encoding_rs", + "futures-core", + "futures-util", + "http", + "http-body", + "hyper", + "hyper-tls", + "ipnet", + "js-sys", + "lazy_static", + "log", + "mime", + "native-tls", + "percent-encoding", + "pin-project-lite 0.2.6", + "serde", + "serde_urlencoded", + "tokio 1.5.0", + "tokio-native-tls", + "url", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", + "winreg", +] + [[package]] name = "ring" version = "0.16.20" @@ -1061,19 +1333,46 @@ dependencies = [ "winapi 0.3.9", ] +[[package]] +name = "rust_decimal" +version = "1.12.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d64bd27418c091ebd539b00f50ad059fc41421c87d5033c217cd76c7506d0d71" +dependencies = [ + "arrayvec", + "num-traits", + "serde", +] + +[[package]] +name = "rust_decimal_macros" +version = "1.12.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "81dbf822f50d527a9ba351f392aadca0ac47375c3bfad5d48ecfad53cc067833" +dependencies = [ + "quote", + "rust_decimal", +] + [[package]] name = "rustls" version = "0.18.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5d1126dcf58e93cee7d098dbda643b5f92ed724f1f6a63007c1116eed6700c81" dependencies = [ - "base64", + "base64 0.12.3", "log", "ring", "sct", "webpki", ] +[[package]] +name = "ryu" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "71d301d4193d031abdd79ff7e3dd721168a9572ef3fe51a1517aba235bd8f86e" + [[package]] name = "schannel" version = "0.1.19" @@ -1122,6 +1421,9 @@ name = "serde" version = "1.0.125" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "558dc50e1a5a5fa7112ca2ce4effcb321b0300c0d4ccf0776a9f60cd89031171" +dependencies = [ + "serde_derive", +] [[package]] name = "serde_derive" @@ -1134,6 +1436,29 @@ dependencies = [ "syn", ] +[[package]] +name = "serde_json" +version = "1.0.64" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "799e97dc9fdae36a5c8b8f2cae9ce2ee9fdce2058c57a93e6099d919fd982f79" +dependencies = [ + "itoa", + "ryu", + "serde", +] + +[[package]] +name = "serde_urlencoded" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "edfa57a7f8d9c1d260a549e7224100f6c43d43f9103e06dd8b4095a9b2b43ce9" +dependencies = [ + "form_urlencoded", + "itoa", + "ryu", + "serde", +] + [[package]] name = "sha1" version = "0.6.0" @@ -1225,6 +1550,16 @@ dependencies = [ "syn", ] +[[package]] +name = "time" +version = "0.1.43" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ca8a50ef2360fbd1eeb0ecd46795a87a19024eb4b53c5dc916ca1fd95fe62438" +dependencies = [ + "libc", + "winapi 0.3.9", +] + [[package]] name = "tinyvec" version = "1.2.0" @@ -1253,7 +1588,7 @@ dependencies = [ "lazy_static", "libc", "memchr", - "mio", + "mio 0.6.23", "mio-named-pipes", "mio-uds", "num_cpus", @@ -1264,6 +1599,21 @@ dependencies = [ "winapi 0.3.9", ] +[[package]] +name = "tokio" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "83f0c8e7c0addab50b663055baf787d0af7f413a46e6e7fb9559a4e4db7137a5" +dependencies = [ + "autocfg", + "bytes 1.0.1", + "libc", + "memchr", + "mio 0.7.11", + "num_cpus", + "pin-project-lite 0.2.6", +] + [[package]] name = "tokio-macros" version = "0.2.6" @@ -1275,6 +1625,16 @@ dependencies = [ "syn", ] +[[package]] +name = "tokio-native-tls" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f7d995660bd2b7f8c1568414c1126076c13fbb725c40112dc0120b78eb9b717b" +dependencies = [ + "native-tls", + "tokio 1.5.0", +] + [[package]] name = "tokio-tls" version = "0.3.1" @@ -1282,7 +1642,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9a70f4fcd7b3b24fb194f837560168208f669ca8cb70d0c4b862944452396343" dependencies = [ "native-tls", - "tokio", + "tokio 0.2.25", ] [[package]] @@ -1296,7 +1656,21 @@ dependencies = [ "futures-sink", "log", "pin-project-lite 0.1.12", - "tokio", + "tokio 0.2.25", +] + +[[package]] +name = "tokio-util" +version = "0.6.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "940a12c99365c31ea8dd9ba04ec1be183ffe4920102bb7122c2f515437601e8e" +dependencies = [ + "bytes 1.0.1", + "futures-core", + "futures-sink", + "log", + "pin-project-lite 0.2.6", + "tokio 1.5.0", ] [[package]] @@ -1308,6 +1682,38 @@ dependencies = [ "serde", ] +[[package]] +name = "tower-service" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "360dfd1d6d30e05fda32ace2c8c70e9c0a9da713275777f5a4dbb8a1893930c6" + +[[package]] +name = "tracing" +version = "0.1.25" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "01ebdc2bb4498ab1ab5f5b73c5803825e60199229ccba0698170e3be0e7f959f" +dependencies = [ + "cfg-if 1.0.0", + "pin-project-lite 0.2.6", + "tracing-core", +] + +[[package]] +name = "tracing-core" +version = "0.1.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f50de3927f93d202783f4513cda820ab47ef17f624b03c096e86ef00c67e6b5f" +dependencies = [ + "lazy_static", +] + +[[package]] +name = "try-lock" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59547bce71d9c38b83d9c0e92b6066c4253371f15005def0c30d9657f50c7642" + [[package]] name = "unicode-bidi" version = "0.3.5" @@ -1377,6 +1783,16 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9d5b2c62b4012a3e1eca5a7e077d13b3bf498c4073e33ccd58626607748ceeca" +[[package]] +name = "want" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ce8a968cb1cd110d136ff8b819a556d6fb6d919363c61534f6860c7eb172ba0" +dependencies = [ + "log", + "try-lock", +] + [[package]] name = "wasi" version = "0.10.2+wasi-snapshot-preview1" @@ -1390,6 +1806,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "83240549659d187488f91f33c0f8547cbfef0b2088bc470c116d1d260ef623d9" dependencies = [ "cfg-if 1.0.0", + "serde", + "serde_json", "wasm-bindgen-macro", ] @@ -1531,6 +1949,15 @@ version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" +[[package]] +name = "winreg" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0120db82e8a1e0b9fb3345a539c478767c0048d842860994d96113d5b667bd69" +dependencies = [ + "winapi 0.3.9", +] + [[package]] name = "ws2_32-sys" version = "0.2.1" diff --git a/Cargo.toml b/Cargo.toml index edbf0b2ffd..e738d52881 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,12 +16,15 @@ anyhow = "1.0.38" easyenv = "0.0.4" log = "0.4.14" prost = "0.7.0" +regex = "1.4.6" rustls = "0.18" serde = "1.0.123" serde_derive = "1.0.123" tokio = { version = "0.2", features = ["full"] } toml = "0.5.8" +google_maps = "1.0.3" + redis = { version = "0.17.0", features = ["tls", "async-std-tls-comp", "tokio-tls-comp"] } [build-dependencies] diff --git a/src/main.rs b/src/main.rs index bd98c4c007..73d836b378 100644 --- a/src/main.rs +++ b/src/main.rs @@ -14,10 +14,12 @@ use std::time::Duration; mod protos; mod redis_client; +mod redis_subscriber; mod secrets; use crate::secrets::Secrets; -use crate::redis_client::{RedisClient, RedisSubscribeClient}; +use crate::redis_client::RedisClient; +use crate::redis_subscriber::RedisSubscribeClient; const ENV_SUBSCRIBE_TOPIC : &'static str = "SUBSCRIBE_TOPIC"; const ENV_SUBSCRIBE_TOPIC_DEFAULT : &'static str = "twitch"; diff --git a/src/redis_client.rs b/src/redis_client.rs index 4cb792d088..2721c285fd 100644 --- a/src/redis_client.rs +++ b/src/redis_client.rs @@ -23,17 +23,6 @@ pub struct RedisClient { connection_failure_count: u32, } -pub struct RedisSubscribeClient { - secrets: RedisSecrets, - - /// This manages our connection. - /// It's meant to be absent when not connected. - /// TODO: Probably need to reset this when we disconnect. - connection: Option, - - connection_failure_count: u32, -} - impl RedisClient { pub fn new(secrets: &RedisSecrets) -> Self { Self { @@ -116,77 +105,3 @@ impl RedisClient { } } -impl RedisSubscribeClient { - pub fn new(secrets: &RedisSecrets) -> Self { - Self { - secrets: secrets.clone(), - connection: None, - connection_failure_count: 0, - } - } - - pub async fn connect(&mut self) -> AnyhowResult<()> { - let client = redis::Client::open(self.secrets.connection_url())?; - let mut connection = client.get_async_connection().await?.into_pubsub(); - - self.connection = Some(connection); - self.connection_failure_count = 0; - - Ok(()) - } - - pub async fn failure_notify_maybe_reconnect(&mut self) -> AnyhowResult<()> { - self.connection_failure_count += 1; - - if self.connection_failure_count > 5 { - println!("Attempting Redis Reconnect in 3 sec..."); - thread::sleep(Duration::from_secs(3)); - self.connect().await?; - self.connection_failure_count = 0; - } else { - println!("Not ready to reconnect. Fail Count = {}", self.connection_failure_count); - } - Ok(()) - } - - pub async fn subscribe(&mut self, topic: &str) -> AnyhowResult<()> { - let mut connection = match &mut self.connection { - None => { - return Err(anyhow!("Not connected")); - }, - Some(connection) => connection, - }; - - connection.subscribe(topic).await?; - Ok(()) - } - - pub async fn start_stream(&mut self) -> AnyhowResult<()> { - let connection = match &mut self.connection { - None => { - return Err(anyhow!("Not connected")); - }, - Some(connection) => connection, - }; - - loop { - for message in connection.on_message().next().await { - let bytes = message.get_payload_bytes(); - - let mut message_proto = match protos::TwitchMessage::decode(bytes) { - Ok(m) => m, - Err(e) => { - warn!("Error decoding proto: {:?}", e); - continue; - }, - }; - - info!("Proto: {:?}", message_proto); - } - - thread::sleep(Duration::from_millis(250)); - } - - Ok(()) - } -} diff --git a/src/redis_subscriber.rs b/src/redis_subscriber.rs new file mode 100644 index 0000000000..c2674233e8 --- /dev/null +++ b/src/redis_subscriber.rs @@ -0,0 +1,96 @@ +use anyhow::anyhow; +use crate::AnyhowResult; +use crate::protos::protos; +use crate::secrets::RedisSecrets; +use log::{info, warn}; +use prost::Message; +use redis::aio::PubSub; +use std::thread; +use std::time::Duration; +use tokio::stream::{Stream, StreamExt}; + +pub struct RedisSubscribeClient { + secrets: RedisSecrets, + + /// This manages our connection. + /// It's meant to be absent when not connected. + /// TODO: Probably need to reset this when we disconnect. + connection: Option, + + connection_failure_count: u32, +} + +impl RedisSubscribeClient { + pub fn new(secrets: &RedisSecrets) -> Self { + Self { + secrets: secrets.clone(), + connection: None, + connection_failure_count: 0, + } + } + + pub async fn connect(&mut self) -> AnyhowResult<()> { + let client = redis::Client::open(self.secrets.connection_url())?; + let mut connection = client.get_async_connection().await?.into_pubsub(); + + self.connection = Some(connection); + self.connection_failure_count = 0; + + Ok(()) + } + + pub async fn failure_notify_maybe_reconnect(&mut self) -> AnyhowResult<()> { + self.connection_failure_count += 1; + + if self.connection_failure_count > 5 { + println!("Attempting Redis Reconnect in 3 sec..."); + thread::sleep(Duration::from_secs(3)); + self.connect().await?; + self.connection_failure_count = 0; + } else { + println!("Not ready to reconnect. Fail Count = {}", self.connection_failure_count); + } + Ok(()) + } + + pub async fn subscribe(&mut self, topic: &str) -> AnyhowResult<()> { + let mut connection = match &mut self.connection { + None => { + return Err(anyhow!("Not connected")); + }, + Some(connection) => connection, + }; + + connection.subscribe(topic).await?; + Ok(()) + } + + pub async fn start_stream(&mut self) -> AnyhowResult<()> { + let connection = match &mut self.connection { + None => { + return Err(anyhow!("Not connected")); + }, + Some(connection) => connection, + }; + + loop { + for message in connection.on_message().next().await { + let bytes = message.get_payload_bytes(); + + let mut message_proto = match protos::TwitchMessage::decode(bytes) { + Ok(m) => m, + Err(e) => { + warn!("Error decoding proto: {:?}", e); + continue; + }, + }; + + info!("Proto: {:?}", message_proto); + } + + thread::sleep(Duration::from_millis(250)); + } + + Ok(()) + } +} diff --git a/src/secrets.rs b/src/secrets.rs index 2f738531aa..801af0ba2d 100644 --- a/src/secrets.rs +++ b/src/secrets.rs @@ -5,6 +5,7 @@ use std::io::Read; pub struct Secrets { pub redis: RedisSecrets, pub twitch: TwitchSecrets, + pub google: GoogleSecrets, } #[derive(Deserialize, Clone)] @@ -15,6 +16,11 @@ pub struct RedisSecrets { pub port: u32, } +#[derive(Deserialize, Clone)] +pub struct GoogleSecrets { + pub api_key: String, +} + #[derive(Deserialize, Clone)] pub struct TwitchSecrets { /// Twitch username