daniel543x/artcraft
1
0 Fork 0
mirror of https://github.com/storytold/artcraft.git synced 2026-11-05 17:58:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

break out redis pubsub, add google geocoder crate

Browse Source
This commit is contained in:
Brandon Thomas committed 2021-04-28 04:49:23 -04:00
1 parent 88960124fd
commit ef937d8097
6 files changed
+545 -96
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

Cargo.lock
Generated
+437 -10
View File
@@ -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"
Cargo.toml
+3
View File
@@ -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]
src/main.rs
+3 -1
View File
@@ -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";
src/redis_client.rs
-85
View File
@@ -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<PubSub>,
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(())
}
}
src/redis_subscriber.rs
+96
View File
@@ -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<PubSub>,
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(())
}
}
src/secrets.rs
+6
View File
@@ -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