Switch to Redis from MongoDB for event broker.

Closes #1.

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"
+version = "0.5.0"
-source = "git+https://gitlab.insrt.uk/insert/hive?rev=a89826df2b30166220e68a6ed01a58b751456604#a89826df2b30166220e68a6ed01a58b751456604"
+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"

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"] }

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

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,

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();
         });

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<String, String, ClientboundNotification>;
+type Hive<'a> = RedisPubSub<'a, String, String, ClientboundNotification>;
-static HIVE: OnceCell<Hive> = OnceCell::new();
+static HIVE: OnceCell<Hive<'static>> = 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()
 }

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<Pool<RedisConnectionManager>> = 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<RedisConnectionManager> {
+    REDISPOOL.get().unwrap()
+}

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 =

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";