From 814de89cbed37ea9b1d029eb5ea0c8ad903d4faf Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Thu, 7 Apr 2022 23:35:00 -0400 Subject: [PATCH] Async pubsub finally works, and I think I can kill the thread --- .../tts_writer.rs | 3 + .../obs_gateway_websocket_handler.rs | 10 +- .../redis_pubsub_event_listener_thread.rs | 97 +++++++++++++------ 3 files changed, 74 insertions(+), 36 deletions(-) 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 3b5c086ea4..a23a724206 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 @@ -1,6 +1,7 @@ use container_common::anyhow_result::AnyhowResult; use database_queries::queries::tts::tts_inference_jobs::insert_tts_inference_job::TtsInferenceJobInsertBuilder; use database_queries::tokens::Tokens; +use log::info; use r2d2_redis::RedisConnectionManager; use r2d2_redis::r2d2; use r2d2_redis::redis::Commands; @@ -45,6 +46,8 @@ impl TtsWriter { let sanitized_text = remove_cheers(message_text); let job_token = Tokens::new_tts_inference_job()?; + info!("Writing TTS: model={}, text={}", model_token, sanitized_text); + let mut builder = TtsInferenceJobInsertBuilder::new_for_internal_tts() .set_is_for_twitch(true) .set_priority_level(TWITCH_TTS_PRIORITY_LEVEL) 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 52c89b2b42..34457d23a8 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 @@ -180,10 +180,10 @@ impl StreamHandler> for ObsGatewayWebSock msg: Result, ctx: &mut Self::Context, ) { - debug!("Socket Handler::handle()"); + //debug!("Socket Handler::handle()"); if let Ok(msg) = msg { - debug!("Socket Handler::handle(): msg = {:?}", msg); + //debug!("Socket Handler::handle(): msg = {:?}", msg); // TODO: Only send this every 60 seconds. // TODO: Error handling. @@ -229,11 +229,11 @@ impl StreamHandler> for ObsGatewayWebSock match msg { ws::Message::Ping(bytes) => { - debug!("Socket Handler::handle(): got ping"); + //debug!("Socket Handler::handle(): got ping"); ctx.pong(&bytes) }, ws::Message::Text(text) => { - debug!("Socket Handler::handle(): got text = {:?}", text); + //debug!("Socket Handler::handle(): got text = {:?}", text); let payload = FrontendEventPayload { response_type: ResponseType::Pong, @@ -252,7 +252,7 @@ impl StreamHandler> for ObsGatewayWebSock //ctx.text("response") }, ws::Message::Binary(bin) => { - debug!("Socket Handler::handle(): got binary..."); + //debug!("Socket Handler::handle(): got binary..."); //ctx.binary("response".as_bytes()) }, ws::Message::Close(reason) => { 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 ab6bad96bf..916386ad16 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 @@ -1,15 +1,17 @@ use container_common::anyhow_result::AnyhowResult; -use futures_util::StreamExt; use crate::endpoints_ws::threads::tts_inference_job_token_queue::TtsInferenceJobTokenQueue; use futures::Future; use futures::future::Ready; +use futures_util::Stream; +use futures_util::StreamExt; use log::error; use log::info; use r2d2_redis::RedisConnectionManager; use r2d2_redis::r2d2; use r2d2_redis::redis::{Commands, PubSub, Msg}; -use redis_common::redis_keys::RedisKeys; use redis_async::Client as AsyncClient; +use redis_common::redis_keys::RedisKeys; +use std::pin::Pin; use std::sync::Arc; use std::thread::sleep; use time::Duration; @@ -63,43 +65,76 @@ impl RedisPubsubEventListenerThread { 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 { - info!("Twitch user Redis PubSub thread loop iter"); + 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"); +// } +// } - // 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!("."); - } + while let Some(value) = pubsub_stream.next().await { + error!("got {:?}", value); } + 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();