update main thread to struct

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<sqlx::Pool<MySql>>,
   redis_pool: Arc<r2d2::Pool<RedisConnectionManager>>,
   redis_pubsub_pool: Arc<r2d2::Pool<RedisConnectionManager>>,
   runtime: Arc<Runtime>,
-  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<String> = 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<sqlx::Pool<MySql>>,
+    redis_pool: Arc<r2d2::Pool<RedisConnectionManager>>,
+    redis_pubsub_pool: Arc<r2d2::Pool<RedisConnectionManager>>,
+    runtime: Arc<Runtime>,
+  ) -> 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<String> = 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<String> = 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<String> = 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());
+    }
   }
 }

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<sqlx::Pool<MySql>>,
   redis_pool: Arc<r2d2::Pool<RedisConnectionManager>>,
   twitch_websocket_client: TwitchWebsocketClient,

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