mirror of
https://github.com/storytold/artcraft.git
synced 2026-11-04 01:18:34 +00:00
use redis connection manager (theoretically reconnects)
This commit is contained in:
Brandon Thomas committed
1 parent
c620cf5b18
commit
71758aef4e
4 files changed
+94
-24
No files matched your search
Generated
+63
@@ -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",
|
||||
|
||||
+1
-1
@@ -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"
|
||||
|
||||
@@ -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;
|
||||
|
||||
|
||||
+23
-23
@@ -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>,
|
||||
connection_manager: Option<ConnectionManager>,
|
||||
|
||||
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<u32> {
|
||||
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<u32> {
|
||||
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)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user