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 916386ad16..b319a2f070 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 @@ -19,9 +19,8 @@ use twitch_common::twitch_user_id::TwitchUserId; pub struct RedisPubsubEventListenerThread { twitch_user_id: TwitchUserId, - /// Redis PubSub connection exclusive to *this* thread. - redis: Arc>, - async_redis: AsyncClient, + /// We'll be maintaining a thread-specific async connection to Redis. + redis_pubsub_connection_string: String, token_queue: TtsInferenceJobTokenQueue, } @@ -33,122 +32,45 @@ impl RedisPubsubEventListenerThread { token_queue: TtsInferenceJobTokenQueue ) -> AnyhowResult { - let redis_manager = - RedisConnectionManager::new(redis_pubsub_connection_string.clone())?; - let redis = Arc::new(r2d2::Pool::builder() - .build(redis_manager)?); - - - let async_redis = AsyncClient::open(redis_pubsub_connection_string)?; Ok(Self { twitch_user_id: twitch_user_id.clone(), - async_redis, - redis, + redis_pubsub_connection_string: redis_pubsub_connection_string.to_string(), token_queue, }) } pub async fn start_thread(mut self) { info!("Twitch user Redis PubSub thread starting"); + // TODO: ERROR HANDLING // TODO: ERROR HANDLING // TODO: ERROR HANDLING - // Setup Pubsub. - - let mut pubsub_pool = self.redis.get().unwrap(); - let mut pubsub = pubsub_pool.as_pubsub(); - - let mut async_redis_conn = self.async_redis.get_async_connection().await.unwrap(); - let mut async_pubsub = async_redis_conn.into_pubsub(); - let pubsub_key = RedisKeys::twitch_tts_job_topic(self.twitch_user_id.get_str()); - pubsub.subscribe(&pubsub_key).unwrap(); - async_pubsub.subscribe(&pubsub_key).await.unwrap(); - - let mut pubsub_stream = async_pubsub.on_message(); loop { error!("Twitch user Redis PubSub thread loop iter, key: {}", pubsub_key); -// for maybe_message in pubsub_stream { -// if let Some(message) = futures::stream::iter(maybe_message) { -// let tts_job_token : String = message.get_payload().unwrap(); -// -// error!("Job token received from Redis: {}", tts_job_token); -// self.token_queue.enqueue_token(&tts_job_token).unwrap(); -// -// } else { -// error!("NO MESSAGE"); -// } -// } + let async_redis = AsyncClient::open(self.redis_pubsub_connection_string.as_str()).unwrap(); + let mut async_redis_conn = async_redis.get_async_connection().await.unwrap(); + let mut async_pubsub = async_redis_conn.into_pubsub(); - while let Some(value) = pubsub_stream.next().await { - error!("got {:?}", value); + async_pubsub.subscribe(&pubsub_key).await.unwrap(); + + let mut pubsub_stream = async_pubsub.on_message(); + + 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); + self.token_queue.enqueue_token(&tts_job_token).unwrap(); } - error!("NO NEXT"); - -// //let maybe_message = Pin::new(&mut pubsub_stream).poll_next().await; -// let maybe_message = pubsub_stream.next().await; -// -// if let Some(message) = maybe_message { -// let tts_job_token : String = message.get_payload().unwrap(); -// -// error!("Job token received from Redis: {}", tts_job_token); -// self.token_queue.enqueue_token(&tts_job_token).unwrap(); -// -// } else { -// error!("NO MESSAGE"); -// } - - -// // NB: We can't have calls to read the Twitch websocket client block forever, and they -// // would do exactly that if not for this code. This is adapted from the very good example -// // in the `tokio-tungstenite` repo, which also contains good recipes for splitting sockets -// // into two unidirectional streams: -// // https://github.com/snapview/tokio-tungstenite/blob/master/examples/interval-server.rs -// -// let mut interval = tokio::time::interval(std::time::Duration::from_secs(1)); -// tokio::select! { -// //maybe_message = pubsub.get_message().unwrap() => { -// //maybe_message = Self::blocking_get_message(&mut pubsub) => { -// maybe_message = pubsub_stream.next() => { -// //let tts_job_token : String = maybe_message.unwrap().get_payload().unwrap(); -// if let Some(message) = maybe_message { -// let tts_job_token : String = message.get_payload().unwrap(); -// -// info!("Job token received from Redis: {}", tts_job_token); -// self.token_queue.enqueue_token(&tts_job_token).unwrap(); -// -// } else { -// info!("NO MESSAGE"); -// sleep(std::time::Duration::from_secs(1)); -// -// } -// } -// _ = interval.tick() => { -// sleep(std::time::Duration::from_secs(1)); -// error!("."); -// } -// } - - -// // TODO: This is blocking. We need a way to exit the thread. -// let message = pubsub.get_message().unwrap(); -// let tts_job_token : String = message.get_payload().unwrap(); -// -// info!("Job token received from Redis: {}", tts_job_token); -// self.token_queue.enqueue_token(&tts_job_token).unwrap(); + // TODO: Proper error handling and avoid thundering herd + error!("Twitch user Redis PubSub stream ended. Restarting shortly..."); + sleep(std::time::Duration::from_secs(5)); } } - - // NB: Futures. Ugh - fn blocking_get_message(pubsub: &mut PubSub) -> Ready> { - let message = pubsub.get_message().unwrap(); - let fut = futures::future::ok::(message); - fut - } } \ No newline at end of file