daniel543x/artcraft
1
0 Fork 0
mirror of https://github.com/storytold/artcraft.git synced 2026-11-05 17:58:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

cleanup redis pubsub, need to work on killing the thread

Browse Source
This commit is contained in:
Brandon Thomas committed 2022-04-07 23:59:06 -04:00
1 parent 814de89cbe
commit 2a4236f7aa
1 file changed
+19 -97
Download Patch File Download Diff File Expand all files Collapse all files
crates/service/websocket_gateway/src/endpoints_ws/threads/redis_pubsub_event_listener_thread.rs
+19 -97
View File
@@ -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<r2d2::Pool<RedisConnectionManager>>,
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<Self> {
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<Result<Msg,()>> {
let message = pubsub.get_message().unwrap();
let fut = futures::future::ok::<Msg,()>(message);
fut
}
}