logging cleanup

crates/service/twitch_pubsub_subscriber/src/threads/twitch_pubsub_user_subscriber/tts_writer.rs

@@ -57,13 +57,7 @@ impl TtsWriter {
 
     let _r = builder.insert(&self.mysql_pool).await?;
 
-    // TODO: Report job token to frontend
     let mut redis = self.redis_pool.get()?;
-    let redis_key = RedisKeys::twitch_tts_job_queue(&self.twitch_user_id.get_str());
-
-    // TODO: Deprecate the RPUSH to the list in favor of PubSub.
-    let _size : Option<u64> = redis.rpush(&redis_key, job_token.clone())?;
-    let _size : Option<u64> = redis.expire(&redis_key, STREAMER_TTS_JOB_QUEUE_TTL_SECONDS)?;
 
     // NB: Be very careful when migrating PubSub Redis. This is starting to get a little messy
     //  with different aliases kept everywhere.

crates/service/websocket_gateway/src/endpoints_ws/obs_gateway_websocket_handler.rs

@@ -127,7 +127,10 @@ pub async fn obs_gateway_websocket_handler(
       server_state.multithreading.redis_pubsub_runtime.spawn(thread.start_thread());
 
   server_state.multithreading.redis_pubsub_runtime.spawn(
-    eventually_kill(join_handle, async_thread_kill_signal.clone()));
+    eventually_kill(
+      join_handle,
+      async_thread_kill_signal.clone(),
+      twitch_user_id.clone()));
 
   let websocket = ObsGatewayWebSocket::new(
     tts_job_token_queue,
@@ -148,17 +151,15 @@ pub async fn obs_gateway_websocket_handler(
 async fn eventually_kill(
   handle: JoinHandle<()>,
   async_thread_kill_signal: AsyncThreadKillSignal,
+  twitch_user_id: TwitchUserId,
 ) {
   loop {
-    error!("Waiting to kill pubsub thread...");
-
     if !async_thread_kill_signal.is_alive().unwrap() {
-      error!("Killing pubsub thread...");
+      warn!("Killing pubsub thread for user: {} (they may have multiple browser sessions)",
+        twitch_user_id.get_str());
       handle.abort();
-      error!("Killed pubsub thread.");
       return;
     }
-
     sleep(Duration::from_secs(10));
   }
 }
@@ -260,11 +261,12 @@ impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for ObsGatewayWebSock
           warn!("Socket Handler::handle(): got close, reason = {:?}", reason);
           ctx.close(reason);
           ctx.stop();
+          warn!("Marking PubSub thread for death.");
+          self.async_thread_kill_signal.mark_thread_for_kill().unwrap();
         }
         _ => {}
       }
     } else {
-      error!(">>>>>> obs streamhandler::STOP");
       ctx.stop();
     }
   }

crates/service/websocket_gateway/src/endpoints_ws/threads/redis_pubsub_event_listener_thread.rs

@@ -52,7 +52,7 @@ impl RedisPubsubEventListenerThread {
     let pubsub_key = RedisKeys::twitch_tts_job_topic(self.twitch_user_id.get_str());
 
     loop {
-      error!("Twitch user Redis PubSub thread loop iter, key: {}", pubsub_key);
+      info!("Twitch user Redis PubSub thread loop iter, key: {}", pubsub_key);
 
       let async_redis = AsyncClient::open(self.redis_pubsub_connection_string.as_str()).unwrap();
 
@@ -66,7 +66,7 @@ impl RedisPubsubEventListenerThread {
       while let Some(message) = pubsub_stream.next().await {
         let tts_job_token : String = message.get_payload().unwrap();
 
-        error!("Job token received from Redis: {}", tts_job_token);
+        info!("Job token received from Redis: {}", tts_job_token);
         self.token_queue.enqueue_token(&tts_job_token).unwrap();
       }