From 446dd31f819aa39e7124fae0ffa108b02492cb39 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Sun, 16 Jan 2022 18:59:55 -0500 Subject: [PATCH] redis pubsub works! --- crates/service/obs_gateway/src/main.rs | 31 ++++----- .../service/obs_gateway/src/server_state.rs | 2 - crates/service/obs_gateway/src/threads/mod.rs | 0 .../obs_gateway/src/twitch_pubsub_job.rs | 69 +++++++++++++++++-- 4 files changed, 76 insertions(+), 26 deletions(-) create mode 100644 crates/service/obs_gateway/src/threads/mod.rs diff --git a/crates/service/obs_gateway/src/main.rs b/crates/service/obs_gateway/src/main.rs index df3355ea1d..7e685f50f4 100644 --- a/crates/service/obs_gateway/src/main.rs +++ b/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..."); diff --git a/crates/service/obs_gateway/src/server_state.rs b/crates/service/obs_gateway/src/server_state.rs index 1364539522..79bdb94610 100644 --- a/crates/service/obs_gateway/src/server_state.rs +++ b/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, - /// For pubsub only (the connection can only be used for pubsub.) - pub redis_pubsub_pool: r2d2::Pool, } // TODO: Many of these do not need to be passed around past server init. diff --git a/crates/service/obs_gateway/src/threads/mod.rs b/crates/service/obs_gateway/src/threads/mod.rs new file mode 100644 index 0000000000..e69de29bb2 diff --git a/crates/service/obs_gateway/src/twitch_pubsub_job.rs b/crates/service/obs_gateway/src/twitch_pubsub_job.rs index 21e4ec5018..c25b6e95c3 100644 --- a/crates/service/obs_gateway/src/twitch_pubsub_job.rs +++ b/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)); } -} \ No newline at end of file +} + +pub async fn listen_for_subscriptions_thread(redis_pool: Arc>) { + + // 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); + } +}