diff --git a/Cargo.lock b/Cargo.lock index 7d57e1d8..85684c98 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -138,6 +138,23 @@ dependencies = [ "event-listener", ] +[[package]] +name = "async-process" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b21b63ab5a0db0369deb913540af2892750e42d949faacc7a61495ac418a1692" +dependencies = [ + "async-io", + "blocking", + "cfg-if 1.0.0", + "event-listener", + "futures-lite", + "libc", + "once_cell", + "signal-hook", + "winapi 0.3.9", +] + [[package]] name = "async-std" version = "1.10.0" @@ -149,6 +166,7 @@ dependencies = [ "async-global-executor", "async-io", "async-lock", + "async-process", "crossbeam-utils", "futures-channel", "futures-core", @@ -492,6 +510,20 @@ dependencies = [ "bitflags", ] +[[package]] +name = "combine" +version = "4.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a909e4d93292cd8e9c42e189f61681eff9d67b6541f96b8a1a737f23737bd001" +dependencies = [ + "bytes 1.1.0", + "futures-core", + "memchr", + "pin-project-lite 0.2.7", + "tokio 1.11.0", + "tokio-util 0.6.8", +] + [[package]] name = "concurrent-queue" version = "1.2.2" @@ -733,6 +765,12 @@ version = "0.15.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "77c90badedccf4105eca100756a0b1289e191f6fcbdadd3cee1d2f614f97da8f" +[[package]] +name = "dtoa" +version = "0.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "56899898ce76aaf4a0f24d914c97ea6ed976d42fec6ad33fcbb0a1103e07b2b0" + [[package]] name = "either" version = "1.6.1" @@ -978,6 +1016,12 @@ version = "0.3.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d3d00f4eddb73e498a54394f228cd55853bdf059259e8e7bc6e69d408892e99" +[[package]] +name = "futures-timer" +version = "3.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e64b03909df88034c26dc1547e8970b91f98bdb65165d6a4e9110d94263dbb2c" + [[package]] name = "futures-util" version = "0.3.17" @@ -1163,12 +1207,14 @@ checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70" [[package]] name = "hive_pubsub" -version = "0.4.4" -source = "git+https://gitlab.insrt.uk/insert/hive?rev=a89826df2b30166220e68a6ed01a58b751456604#a89826df2b30166220e68a6ed01a58b751456604" +version = "0.5.0" +source = "git+https://gitlab.insrt.uk/insert/hive?rev=b0f3db9d33990530d7640d4bbb309c8d9eb5c0cf#b0f3db9d33990530d7640d4bbb309c8d9eb5c0cf" dependencies = [ "futures", "many-to-many", - "mongodb", + "mobc", + "mobc-redis", + "redis 0.21.2", "serde", "serde_json", "ulid", @@ -1692,6 +1738,32 @@ dependencies = [ "winapi 0.3.9", ] +[[package]] +name = "mobc" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f76d2f2e2dcbb00a8d3b2b09f026a74a82693ea52cd071647aa6cfa7f1ff37e" +dependencies = [ + "async-std", + "async-trait", + "futures-channel", + "futures-core", + "futures-timer", + "futures-util", + "log", + "tokio 1.11.0", +] + +[[package]] +name = "mobc-redis" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7b5db77b37c9224d5b9949b214041ea3e1c15b6b1e5dd24a5acb8e73975d6d6" +dependencies = [ + "mobc", + "redis 0.19.0", +] + [[package]] name = "mongodb" version = "1.2.3" @@ -2516,6 +2588,48 @@ dependencies = [ "rand_core 0.3.1", ] +[[package]] +name = "redis" +version = "0.19.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1a6ddfecac9391fed21cce10e83c65fa4abafd77df05c98b1c647c65374ce9b3" +dependencies = [ + "async-std", + "async-trait", + "bytes 1.1.0", + "combine", + "dtoa", + "futures-util", + "itoa", + "percent-encoding", + "pin-project-lite 0.2.7", + "sha1", + "tokio 1.11.0", + "tokio-util 0.6.8", + "url", +] + +[[package]] +name = "redis" +version = "0.21.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "202c5bf92cad3d57605c366e644a7fbf305a83f19754fc66678c6265dcc9b8b4" +dependencies = [ + "async-std", + "async-trait", + "bytes 1.1.0", + "combine", + "dtoa", + "futures-util", + "itoa", + "percent-encoding", + "pin-project-lite 0.2.7", + "sha1", + "tokio 1.11.0", + "tokio-util 0.6.8", + "url", +] + [[package]] name = "redox_syscall" version = "0.2.10" @@ -2673,12 +2787,15 @@ dependencies = [ "linkify", "log", "many-to-many", + "mobc", + "mobc-redis", "mongodb", "nanoid", "num_enum", "once_cell", "phf", "rauth", + "redis 0.21.2", "regex", "reqwest 0.11.4", "rmp-serde", @@ -3094,6 +3211,16 @@ dependencies = [ "opaque-debug 0.3.0", ] +[[package]] +name = "signal-hook" +version = "0.3.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9c98891d737e271a2954825ef19e46bd16bdb98e2746f2eec4f7a4ef7946efd1" +dependencies = [ + "libc", + "signal-hook-registry", +] + [[package]] name = "signal-hook-registry" version = "1.4.0" diff --git a/Cargo.toml b/Cargo.toml index 3667ddc3..55b5f19b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -49,7 +49,12 @@ web-push = "0.7.2" many-to-many = "0.1.2" lettre = "0.10.0-alpha.4" rauth = { git = "https://github.com/insertish/rauth", rev = "775a6f3ffb9ffc0e382af32fac2924e9a707e890" } -hive_pubsub = { git = "https://gitlab.insrt.uk/insert/hive", rev = "a89826df2b30166220e68a6ed01a58b751456604", features = ["mongo"] } +hive_pubsub = { git = "https://gitlab.insrt.uk/insert/hive", rev = "b0f3db9d33990530d7640d4bbb309c8d9eb5c0cf", features = ["redis-backend"] } + +# redis +redis = { version = "0.21.2", features = ["async-std-comp"] } +mobc = { version = "0.7.3" } +mobc-redis = { version = "0.7.0", default-features = false, features = ["async-std-comp"] } # web async-tungstenite = { version = "0.10.0", features = ["async-std-runtime"] } diff --git a/set_version.sh b/set_version.sh index 6ddded04..d4fc3d4a 100755 --- a/set_version.sh +++ b/set_version.sh @@ -1,3 +1,3 @@ #!/bin/bash -export version=0.5.3-alpha.4-patch.0 +export version=0.5.3-alpha.5 echo "pub const VERSION: &str = \"${version}\";" > src/version.rs diff --git a/src/main.rs b/src/main.rs index 60c69196..2beb3f59 100644 --- a/src/main.rs +++ b/src/main.rs @@ -17,6 +17,7 @@ extern crate ctrlc; pub mod database; pub mod notifications; pub mod routes; +pub mod redis; pub mod util; pub mod version; @@ -47,6 +48,7 @@ async fn main() { util::variables::preflight_checks(); database::connect().await; + redis::connect().await; notifications::hive::init_hive().await; ctrlc::set_handler(move || { @@ -56,7 +58,7 @@ async fn main() { .expect("Error setting Ctrl-C handler"); let web_task = task::spawn(launch_web()); - let hive_task = task::spawn(notifications::hive::listen()); + let hive_task = task::spawn_local(notifications::hive::listen()); join!( web_task, diff --git a/src/notifications/events.rs b/src/notifications/events.rs index 314359e5..6660ebaf 100644 --- a/src/notifications/events.rs +++ b/src/notifications/events.rs @@ -187,7 +187,7 @@ impl ClientboundNotification { pub fn publish(self, topic: String) { async_std::task::spawn(async move { prehandle_hook(&self).await.ok(); // ! FIXME: this should be moved to pubsub - hive_pubsub::backend::mongo::publish(get_hive(), &topic, self) + hive_pubsub::backend::redis::publish(get_hive(), topic, self) .await .ok(); }); diff --git a/src/notifications/hive.rs b/src/notifications/hive.rs index 591c6d85..80652a83 100644 --- a/src/notifications/hive.rs +++ b/src/notifications/hive.rs @@ -1,18 +1,23 @@ +use std::sync::{Arc, Mutex}; + use super::{events::ClientboundNotification, websocket}; -use crate::database::*; +use crate::redis::get_pool; +use crate::util::variables::REDIS_URI; use futures::FutureExt; -use hive_pubsub::backend::mongo::MongodbPubSub; +use hive_pubsub::backend::redis::RedisPubSub; use hive_pubsub::PubSub; use log::{debug, error}; use once_cell::sync::OnceCell; use serde_json::to_string; -type Hive = MongodbPubSub; -static HIVE: OnceCell = OnceCell::new(); +type Hive<'a> = RedisPubSub<'a, String, String, ClientboundNotification>; +static HIVE: OnceCell> = OnceCell::new(); pub async fn init_hive() { - let hive = MongodbPubSub::new( + let pubsub_con = redis::Client::open(REDIS_URI.to_string()).unwrap().get_async_connection().await.unwrap().into_pubsub(); + + let hive = RedisPubSub::new( |ids, notification: ClientboundNotification| { let notif = notification.clone(); async_std::task::spawn(async move { @@ -26,7 +31,8 @@ pub async fn init_hive() { error!("Failed to serialise notification."); } }, - get_collection("pubsub"), + get_pool(), + Arc::new(Mutex::new(pubsub_con)) ); if HIVE.set(hive).is_err() { @@ -37,6 +43,7 @@ pub async fn init_hive() { pub async fn listen() { HIVE.get() .unwrap() + .clone() .listen() .fuse() .await @@ -61,6 +68,6 @@ pub fn subscribe_if_exists(user: String, topic: String) -> Result<(), String> { Ok(()) } -pub fn get_hive() -> &'static Hive { +pub fn get_hive() -> &'static Hive<'static> { HIVE.get().unwrap() } diff --git a/src/redis/mod.rs b/src/redis/mod.rs new file mode 100644 index 00000000..912d3500 --- /dev/null +++ b/src/redis/mod.rs @@ -0,0 +1,18 @@ +use crate::util::variables::REDIS_URI; + +use mobc::Pool; +use mobc_redis::RedisConnectionManager; +use once_cell::sync::OnceCell; + +static REDISPOOL: OnceCell> = OnceCell::new(); + +pub async fn connect() { + let client = mobc_redis::redis::Client::open(REDIS_URI.to_string()).unwrap(); + let manager = mobc_redis::RedisConnectionManager::new(client); + let pool = mobc::Pool::builder().max_open(100).build(manager); + REDISPOOL.set(pool).ok().unwrap(); +} + +pub fn get_pool() -> &'static Pool { + REDISPOOL.get().unwrap() +} diff --git a/src/util/variables.rs b/src/util/variables.rs index 198b0ab9..278afa7d 100644 --- a/src/util/variables.rs +++ b/src/util/variables.rs @@ -7,6 +7,8 @@ lazy_static! { // Application Settings pub static ref MONGO_URI: String = env::var("REVOLT_MONGO_URI").expect("Missing REVOLT_MONGO_URI environment variable."); + pub static ref REDIS_URI: String = + env::var("REVOLT_REDIS_URI").expect("Missing REVOLT_REDIS_URI environment variable."); pub static ref WS_HOST: String = env::var("REVOLT_WS_HOST").unwrap_or_else(|_| "0.0.0.0:9000".to_string()); pub static ref PUBLIC_URL: String = diff --git a/src/version.rs b/src/version.rs index 0d8a24a0..2e6f4203 100644 --- a/src/version.rs +++ b/src/version.rs @@ -1 +1 @@ -pub const VERSION: &str = "0.5.3-alpha.4-patch.0"; +pub const VERSION: &str = "0.5.3-alpha.5";