diff --git a/Cargo.lock b/Cargo.lock index e747c911a2..bc01b56b9c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -17,6 +17,12 @@ version = "1.0.38" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "afddf7f520a80dbf76e6f50a35bca42a2331ef227a28b3b6dc5c2e2338d114b1" +[[package]] +name = "arc-swap" +version = "0.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dabe5a181f83789739c194cbe5a897dde195078fac08568d09221fd6137a7ba8" + [[package]] name = "async-channel" version = "1.6.1" @@ -417,6 +423,21 @@ version = "0.3.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3dcaa9ae7725d12cdb85b3ad99a434db70b468c09ded17e012d86b5c1010f7a7" +[[package]] +name = "futures" +version = "0.3.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f55667319111d593ba876406af7c409c0ebb44dc4be6132a783ccf163ea14c1" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + [[package]] name = "futures-channel" version = "0.3.13" @@ -424,6 +445,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8c2dd2df839b57db9ab69c2c9d8f3e8c81984781937fe2807dc6dcf3b2ad2939" dependencies = [ "futures-core", + "futures-sink", ] [[package]] @@ -432,6 +454,17 @@ version = "0.3.13" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "15496a72fabf0e62bdc3df11a59a3787429221dd0710ba8ef163d6f7a9112c94" +[[package]] +name = "futures-executor" +version = "0.3.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "891a4b7b96d84d5940084b2a37632dd65deeae662c114ceaa2c879629c9c0ad1" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + [[package]] name = "futures-io" version = "0.3.13" @@ -453,6 +486,18 @@ dependencies = [ "waker-fn", ] +[[package]] +name = "futures-macro" +version = "0.3.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ea405816a5139fb39af82c2beb921d52143f556038378d6db21183a5c37fbfb7" +dependencies = [ + "proc-macro-hack", + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "futures-sink" version = "0.3.13" @@ -477,13 +522,17 @@ version = "0.3.13" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1812c7ab8aedf8d6f2701a43e1243acdbcc2b36ab26e2ad421eb99ac963d96d1" dependencies = [ + "futures-channel", "futures-core", "futures-io", + "futures-macro", "futures-sink", "futures-task", "memchr", "pin-project-lite 0.2.4", "pin-utils", + "proc-macro-hack", + "proc-macro-nested", "slab", ] @@ -907,6 +956,18 @@ version = "0.2.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ac74c624d6b2d21f425f752262f42188365d7b8ff1aff74c82e45136510a4857" +[[package]] +name = "proc-macro-hack" +version = "0.5.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dbf0c48bc1d91375ae5c3cd81e3722dff1abcf81a30960240640d223f59fe0e5" + +[[package]] +name = "proc-macro-nested" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bc881b2c22681370c6a780e47af9840ef841837bc98118431d4e1868bd0c1086" + [[package]] name = "proc-macro2" version = "1.0.24" @@ -1028,12 +1089,14 @@ version = "0.17.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "95357caf2640abc54651b93c98a8df4fe1ccbf44b8e601ccdf43d5c1451f29ac" dependencies = [ + "arc-swap", "async-native-tls", "async-std", "async-trait", "bytes 0.5.6", "combine", "dtoa", + "futures", "futures-util", "itoa", "native-tls", diff --git a/Cargo.toml b/Cargo.toml index c87af41d11..b2406efa0f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -22,7 +22,7 @@ serde_derive = "1.0.123" tokio = { version = "0.2", features = ["full"] } toml = "0.5.8" -redis = { version = "0.17.0", features = ["tls", "async-std-tls-comp", "tokio-tls-comp"] } +redis = { version = "0.17.0", features = ["tls", "async-std-tls-comp", "tokio-tls-comp", "connection-manager"] } # Twitch Helix etc. API, but no PubSub just yet #twitch_api2 = "0.4.1" diff --git a/src/main.rs b/src/main.rs index 082fff4903..89ce24812f 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,3 +1,10 @@ +#![allow(dead_code)] +#![allow(unused_mut)] +#![allow(unused_imports)] +#![warn(unused_must_use)] +//#![allow(warnings)] + + #[macro_use] extern crate serde_derive; diff --git a/src/redis_client.rs b/src/redis_client.rs index 82c1eaa074..1f00adcdd8 100644 --- a/src/redis_client.rs +++ b/src/redis_client.rs @@ -5,7 +5,7 @@ use anyhow::anyhow; use std::thread; use std::time::Duration; use redis::AsyncCommands; -use redis::aio::Connection; +use redis::aio::{Connection, ConnectionManager}; use redis; pub struct RedisClient { @@ -13,8 +13,7 @@ pub struct RedisClient { /// 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_manager: Option, connection_failure_count: u32, } @@ -23,16 +22,17 @@ impl RedisClient { pub fn new(secrets: &RedisSecrets) -> Self { Self { secrets: secrets.clone(), - connection: None, + connection_manager: 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?; - self.connection = Some(connection); + let connection_manager = client.get_tokio_connection_manager().await?; + self.connection_manager = Some(connection_manager); + self.connection_failure_count = 0; Ok(()) @@ -53,35 +53,35 @@ impl RedisClient { } pub async fn publish(&mut self, channel: &str, message: &str) -> AnyhowResult { - let connection = match &mut self.connection { + let connection_manager = match &mut self.connection_manager { None => { return Err(anyhow!("Not connected")); }, - Some(connection) => connection, + Some(cm) => cm, }; - // TODO: Smart reconnect - /*let result = connection.publish(channel, message).await?; - let result = match result { - Ok(r) => r, - Err(err) => { - match err.kind() { - IoError | TryAgain | ClusterDown | MasterDown => {}, - _ => return Err(anyhow!("Redis error: {:?}", err)), - } - } - };*/ + // // TODO: Smart reconnect + // /*let result = connection.publish(channel, message).await?; + // let result = match result { + // Ok(r) => r, + // Err(err) => { + // match err.kind() { + // IoError | TryAgain | ClusterDown | MasterDown => {}, + // _ => return Err(anyhow!("Redis error: {:?}", err)), + // } + // } + // };*/ - let result = connection.publish(channel, message).await?; + let result = connection_manager.publish(channel, message).await?; Ok(result) } pub async fn publish_bytes(&mut self, channel: &str, message: &[u8]) -> AnyhowResult { - let connection = match &mut self.connection { + let connection_manager = match &mut self.connection_manager { None => { return Err(anyhow!("Not connected")); }, - Some(connection) => connection, + Some(cm) => cm, }; // TODO: Smart reconnect @@ -96,7 +96,7 @@ impl RedisClient { } };*/ - let result = connection.publish(channel, message).await?; + let result = connection_manager.publish(channel, message).await?; Ok(result) } }