redis pubsub works!

crates/service/obs_gateway/src/main.rs

@@ -15,6 +15,7 @@
 pub mod endpoints;
 pub mod endpoints_ws;
 pub mod server_state;
+pub mod threads;
 pub mod twitch;
 pub mod util;
 
@@ -48,7 +49,7 @@ use sqlx::mysql::MySqlPoolOptions;
 use std::sync::Arc;
 use std::thread::sleep;
 use std::time::Duration;
-use tokio::runtime::Handle;
+use tokio::runtime::{Handle, Runtime};
 use twitch_api2::pubsub::Topic;
 use twitch_api2::pubsub;
 use twitch_oauth2::tokens::UserTokenBuilder;
@@ -126,13 +127,6 @@ async fn main() -> AnyhowResult<()> {
   let redis_pool = r2d2::Pool::builder()
       .build(redis_manager)?;
 
-  info!("Connecting to pubsub redis...");
-
-  let redis_pubsub_manager = RedisConnectionManager::new(redis_connection_string.clone())?;
-
-  let redis_pubsub_pool = r2d2::Pool::builder()
-      .build(redis_pubsub_manager)?;
-
   let server_state = ObsGatewayServerState {
     env_config: EnvConfig {
       num_workers,
@@ -156,20 +150,23 @@ async fn main() -> AnyhowResult<()> {
     backends: BackendsConfig {
       mysql_pool: pool,
       redis_pool,
-      redis_pubsub_pool,
     }
   };
 
-  info!("Starting thread...");
+  // Works, but can't Ctrl-C
+  //let handle = Handle::current();
+  //handle.spawn_blocking(|| {
+  //  loop {
+  //    info!("...thread...");
+  //    sleep(Duration::from_millis(2000));
+  //  }
+  //});
 
-  let handle = Handle::current();
+  //let tokio_runtime = Runtime::new()?;
+  //tokio_runtime.spawn(async {
+  //  poll_ip_bans(ip_banlist2, mysql_pool3).await;
+  //});
 
-  handle.spawn_blocking(|| {
-    loop {
-      info!("...thread...");
-      sleep(Duration::from_millis(2000));
-    }
-  });
 
   info!("Starting server...");
 

crates/service/obs_gateway/src/server_state.rs

@@ -20,8 +20,6 @@ pub struct ObsGatewayServerState {
 pub struct BackendsConfig {
   pub mysql_pool: MySqlPool,
   pub redis_pool: r2d2::Pool<RedisConnectionManager>,
-  /// For pubsub only (the connection can only be used for pubsub.)
-  pub redis_pubsub_pool: r2d2::Pool<RedisConnectionManager>,
 }
 
 // TODO: Many of these do not need to be passed around past server init.

crates/service/obs_gateway/src/twitch_pubsub_job.rs

@@ -1,6 +1,21 @@
-use config::shared_constants::DEFAULT_RUST_LOG;
+#![forbid(private_in_public)]
+#![forbid(unused_must_use)]
+//#![forbid(warnings)]
+
+#![allow(dead_code)]
+#![allow(non_snake_case)]
+#![allow(unused_imports)]
+#![allow(unused_mut)]
+#![allow(unused_variables)]
+
+#[macro_use] extern crate lazy_static;
+#[macro_use] extern crate magic_crypt;
+#[macro_use] extern crate serde_derive;
+
+use config::shared_constants::{DEFAULT_RUST_LOG, DEFAULT_REDIS_DATABASE_1_CONNECTION_STRING};
 use container_common::anyhow_result::AnyhowResult;
 use futures::executor::{ThreadPool, ThreadPoolBuilder};
+use r2d2_redis::r2d2;
 use log::info;
 use std::thread::sleep;
 use std::time::Duration;
@@ -8,7 +23,12 @@ use crate::twitch::twitch_client_wrapper::TwitchClientWrapper;
 use crate::twitch::websocket_client::TwitchWebsocketClient;
 use futures::task::SpawnExt;
 use tokio::runtime::Builder;
+use r2d2_redis::RedisConnectionManager;
+use std::sync::Arc;
+use r2d2_redis::r2d2::Pool;
+use redis_common::redis_keys::RedisKeys;
 
+pub mod threads;
 pub mod twitch;
 pub mod util;
 
@@ -16,6 +36,20 @@ pub mod util;
 pub async fn main() -> AnyhowResult<()> {
   easyenv::init_all_with_default_logging(Some(DEFAULT_RUST_LOG));
 
+  let redis_connection_string =
+      easyenv::get_env_string_or_default(
+        "REDIS_1_URL",
+        DEFAULT_REDIS_DATABASE_1_CONNECTION_STRING);
+
+  info!("Connecting to pubsub redis...");
+
+  let redis_pubsub_manager = RedisConnectionManager::new(redis_connection_string.clone())?;
+
+  let redis_pubsub_pool = r2d2::Pool::builder()
+      .build(redis_pubsub_manager)?;
+
+  let redis_pubsub_pool = Arc::new(redis_pubsub_pool);
+
   info!("Creating thread pool...");
 
   // "The thread pool multiplexes any number of tasks onto a fixed number
@@ -28,19 +62,19 @@ pub async fn main() -> AnyhowResult<()> {
   // https://docs.rs/tokio/latest/tokio/runtime/struct.Builder.html
   let runtime = Builder::new_multi_thread()
       .worker_threads(4)
-      .thread_name("my-custom-name")
+      .thread_name("twitch-pubsub-")
       .thread_stack_size(3 * 1024 * 1024)
       .enable_all()
       .build()?;
 
   info!("Thread pool created");
 
-  //thread_pool.spawn(watch_user_thread())?;
-  runtime.spawn(watch_user_thread(10));
-  runtime.spawn(watch_user_thread(9999));
+  //runtime.spawn(watch_user_thread(10));
+  //runtime.spawn(watch_user_thread(9999));
+  runtime.spawn(listen_for_subscriptions_thread(redis_pubsub_pool));
 
   loop {
-    sleep(Duration::from_millis(1000));
+    sleep(Duration::from_millis(10_000));
   }
 }
 
@@ -55,4 +89,25 @@ pub async fn watch_user_thread(user_id: u32) {
     client.send_ping().await.unwrap();
     sleep(Duration::from_millis(10_000));
   }
-}
+}
+
+pub async fn listen_for_subscriptions_thread(redis_pool: Arc<Pool<RedisConnectionManager>>) {
+
+  // TODO: ERROR HANDLING
+  let mut pool = redis_pool.get().unwrap();
+  let mut pubsub = pool.as_pubsub();
+  let channel = RedisKeys::obs_session_active_topic();
+  pubsub.subscribe(channel).unwrap();
+
+  //loop {
+  //  let payload : String = try!(msg.get_payload());
+  //  println!("channel '{}': {}", msg.get_channel_name(), payload);
+  //}
+
+  loop {
+    info!("PubSub");
+    let message = pubsub.get_message().unwrap();
+    let payload : String = message.get_payload().unwrap();
+    info!("Message: {}", payload);
+  }
+}