diff --git a/crates/service/twitch_pubsub_subscriber/src/threads/twitch_pubsub_user_subscriber/tts_writer.rs b/crates/service/twitch_pubsub_subscriber/src/threads/twitch_pubsub_user_subscriber/tts_writer.rs index a23a724206..d6409e1fff 100644 --- a/crates/service/twitch_pubsub_subscriber/src/threads/twitch_pubsub_user_subscriber/tts_writer.rs +++ b/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 = redis.rpush(&redis_key, job_token.clone())?; - let _size : Option = 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. diff --git a/crates/service/websocket_gateway/src/endpoints_ws/obs_gateway_websocket_handler.rs b/crates/service/websocket_gateway/src/endpoints_ws/obs_gateway_websocket_handler.rs index 6863649a90..d575ea98f8 100644 --- a/crates/service/websocket_gateway/src/endpoints_ws/obs_gateway_websocket_handler.rs +++ b/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> 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(); } } diff --git a/crates/service/websocket_gateway/src/endpoints_ws/threads/redis_pubsub_event_listener_thread.rs b/crates/service/websocket_gateway/src/endpoints_ws/threads/redis_pubsub_event_listener_thread.rs index 1e3e97dc8e..29fb6ae240 100644 --- a/crates/service/websocket_gateway/src/endpoints_ws/threads/redis_pubsub_event_listener_thread.rs +++ b/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(); }