diff --git a/crates/service/obs_gateway/src/threads/listen_for_active_obs_sessions_thread.rs b/crates/service/obs_gateway/src/threads/listen_for_active_obs_sessions_thread.rs index a8242df1d5..28b0b9945f 100644 --- a/crates/service/obs_gateway/src/threads/listen_for_active_obs_sessions_thread.rs +++ b/crates/service/obs_gateway/src/threads/listen_for_active_obs_sessions_thread.rs @@ -19,66 +19,87 @@ use container_common::thread::thread_id::ThreadId; // TODO: Apart from error handling, this looks mostly good. -pub async fn listen_for_active_obs_session_thread( + +pub struct ListenForActiveObsSessionThread { + server_hostname: String, mysql_pool: Arc>, redis_pool: Arc>, redis_pubsub_pool: Arc>, runtime: Arc, - hostname: String, -) { - // TODO: ERROR HANDLING - // TODO: ERROR HANDLING - // TODO: ERROR HANDLING +} - // Setup PubSub. - - let mut pubsub_pool = redis_pubsub_pool.get().unwrap(); - let mut pubsub = pubsub_pool.as_pubsub(); - - pubsub.subscribe(RedisKeys::obs_active_session_topic()).unwrap(); - - loop { - let message = pubsub.get_message().unwrap(); - let payload : String = message.get_payload().unwrap(); - let payload = ObsActivePayload::from_json_str(&payload).unwrap(); - - let twitch_user_id = TwitchUserId::from_str(&payload.twitch_user_id).unwrap(); - - // See if we have a lease already... - - let mut redis = redis_pool.get().unwrap(); - let lease_key = RedisKeys::twitch_pubsub_lease(&payload.twitch_user_id); - let lease_value : Option = redis.get(&lease_key).unwrap(); - - if let Some(value) = lease_value.as_deref() { - let lease = LeasePayload::deserialize(value).unwrap(); - continue; +impl ListenForActiveObsSessionThread { + pub fn new( + server_hostname: String, + mysql_pool: Arc>, + redis_pool: Arc>, + redis_pubsub_pool: Arc>, + runtime: Arc, + ) -> Self { + Self { + server_hostname, + mysql_pool, + redis_pool, + redis_pubsub_pool, + runtime, } + } - info!("No existing lease for {:?}...", &lease_key); + pub async fn start_thread(mut self) { + // TODO: ERROR HANDLING + // TODO: ERROR HANDLING + // TODO: ERROR HANDLING - let thread_id = ThreadId::random_id(); - let lease = LeasePayload::from_thread_id(&hostname, &thread_id); + // Setup PubSub. - let serialized = lease.serialize(); - let _v : Option = redis.set_ex( - &lease_key, - &serialized, - LEASE_TIMEOUT_SECONDS - ).unwrap(); + let mut pubsub_pool = self.redis_pubsub_pool.get().unwrap(); + let mut pubsub = pubsub_pool.as_pubsub(); - // Then launch the thread... + pubsub.subscribe(RedisKeys::obs_active_session_topic()).unwrap(); - let redis_pool2 = redis_pool.clone(); - let mysql_pool2 = mysql_pool.clone(); + loop { + let message = pubsub.get_message().unwrap(); + let payload : String = message.get_payload().unwrap(); + let payload = ObsActivePayload::from_json_str(&payload).unwrap(); - let thread = TwitchPubsubUserSubscriberThread::new( - twitch_user_id, - mysql_pool2, - redis_pool2, - &hostname, - thread_id.clone()); + let twitch_user_id = TwitchUserId::from_str(&payload.twitch_user_id).unwrap(); - runtime.spawn(thread.start_thread()); + // See if we have a lease already... + + let mut redis = self.redis_pool.get().unwrap(); + let lease_key = RedisKeys::twitch_pubsub_lease(&payload.twitch_user_id); + let lease_value : Option = redis.get(&lease_key).unwrap(); + + if let Some(value) = lease_value.as_deref() { + let lease = LeasePayload::deserialize(value).unwrap(); + continue; + } + + info!("No existing lease for {:?}...", &lease_key); + + let thread_id = ThreadId::random_id(); + let lease = LeasePayload::from_thread_id(&self.server_hostname, &thread_id); + + let serialized = lease.serialize(); + let _v : Option = redis.set_ex( + &lease_key, + &serialized, + LEASE_TIMEOUT_SECONDS + ).unwrap(); + + // Then launch the thread... + + let redis_pool2 = self.redis_pool.clone(); + let mysql_pool2 = self.mysql_pool.clone(); + + let thread = TwitchPubsubUserSubscriberThread::new( + twitch_user_id, + mysql_pool2, + redis_pool2, + &self.server_hostname, + thread_id.clone()); + + self.runtime.spawn(thread.start_thread()); + } } } diff --git a/crates/service/obs_gateway/src/threads/twitch_pubsub_user_subscriber_thread.rs b/crates/service/obs_gateway/src/threads/twitch_pubsub_user_subscriber_thread.rs index d1548a987f..63ef01c019 100644 --- a/crates/service/obs_gateway/src/threads/twitch_pubsub_user_subscriber_thread.rs +++ b/crates/service/obs_gateway/src/threads/twitch_pubsub_user_subscriber_thread.rs @@ -36,6 +36,7 @@ pub struct TwitchPubsubUserSubscriberThread { thread_id: ThreadId, server_hostname: String, twitch_user_id: TwitchUserId, + mysql_pool: Arc>, redis_pool: Arc>, twitch_websocket_client: TwitchWebsocketClient, diff --git a/crates/service/obs_gateway/src/twitch_pubsub_job.rs b/crates/service/obs_gateway/src/twitch_pubsub_job.rs index 31a20bdab0..dab36bd0f9 100644 --- a/crates/service/obs_gateway/src/twitch_pubsub_job.rs +++ b/crates/service/obs_gateway/src/twitch_pubsub_job.rs @@ -19,7 +19,7 @@ pub mod util; use config::shared_constants::{DEFAULT_RUST_LOG, DEFAULT_REDIS_DATABASE_1_CONNECTION_STRING, DEFAULT_MYSQL_CONNECTION_STRING}; use container_common::anyhow_result::AnyhowResult; -use crate::threads::listen_for_active_obs_sessions_thread::listen_for_active_obs_session_thread; +use crate::threads::listen_for_active_obs_sessions_thread::ListenForActiveObsSessionThread; use crate::twitch::websocket_client::TwitchWebsocketClient; use futures::executor::{ThreadPool, ThreadPoolBuilder}; use futures::task::SpawnExt; @@ -112,9 +112,15 @@ pub async fn main() -> AnyhowResult<()> { info!("Thread pool created"); - runtime.spawn(listen_for_active_obs_session_thread( - mysql_pool, redis_pool, redis_pubsub_pool, runtime_2, - server_hostname.to_string())); + let thread = ListenForActiveObsSessionThread::new( + server_hostname.to_string(), + mysql_pool, + redis_pool, + redis_pubsub_pool, + runtime_2, + ); + + runtime.spawn(thread.start_thread()); loop { sleep(Duration::from_millis(10_000));