From a9dbc7303c4e546705f66a2eecae6168fbe98bf8 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Fri, 21 Jan 2022 23:43:51 -0500 Subject: [PATCH] cleanup --- crates/lib/redis_common/src/redis_keys.rs | 15 ++- .../obs_gateway_websocket_handler.rs | 115 +----------------- .../redis/{lease_timeout.rs => constants.rs} | 4 + crates/service/obs_gateway/src/redis/mod.rs | 2 +- .../listen_for_active_obs_sessions_thread.rs | 6 +- .../twitch_pubsub_user_subscriber_thread.rs | 2 +- 6 files changed, 22 insertions(+), 122 deletions(-) rename crates/service/obs_gateway/src/redis/{lease_timeout.rs => constants.rs} (80%) diff --git a/crates/lib/redis_common/src/redis_keys.rs b/crates/lib/redis_common/src/redis_keys.rs index dbd4bd59af..42332d168f 100644 --- a/crates/lib/redis_common/src/redis_keys.rs +++ b/crates/lib/redis_common/src/redis_keys.rs @@ -47,8 +47,15 @@ impl RedisKeys { /// This is a PubSub topic. /// We publish when new viewers arrive and as a "keep-alive ping". - pub fn obs_active_session_topic() -> &'static str { - "obsActiveSessionTopic" + pub fn obs_active_sessions_topic() -> &'static str { + "obsActiveSessionsTopic" + } + + /// This is a key that denotes a user is actively on the OBS gateway. + /// These keys have a short expiry and OBS gateway must continually bump them. + /// If a key goes away, the thread monitoring that user should exit. + pub fn obs_active_session(twitch_user_id: &str) -> String { + format!("obsActiveSession:{}", twitch_user_id) } /// This is a key that denotes a PubSub lease subscriber. @@ -61,7 +68,7 @@ impl RedisKeys { /// This is a PubSub topic. /// These are "unenriched" Twitch events from the PubSub (and eventually IRC) /// subscribers. Downstream listeners will enrich these for user-facing functionality. - pub fn twitchEventTopic(twitch_user_id: &str) -> String { - format!("twitchEventTopic:{}", twitch_user_id) + pub fn twitch_events_topic(twitch_user_id: &str) -> String { + format!("twitchEventsTopic:{}", twitch_user_id) } } diff --git a/crates/service/obs_gateway/src/endpoints_ws/obs_gateway_websocket_handler.rs b/crates/service/obs_gateway/src/endpoints_ws/obs_gateway_websocket_handler.rs index e706062ea7..1b8d35cd9b 100644 --- a/crates/service/obs_gateway/src/endpoints_ws/obs_gateway_websocket_handler.rs +++ b/crates/service/obs_gateway/src/endpoints_ws/obs_gateway_websocket_handler.rs @@ -77,7 +77,7 @@ pub async fn obs_gateway_websocket_handler( CommonServerError::ServerError })?; - let channel = RedisKeys::obs_active_session_topic(); + let channel = RedisKeys::obs_active_sessions_topic(); let payload = ObsActivePayload::new(&token_record.twitch_user_id); let json_payload = payload.serialize() @@ -92,36 +92,12 @@ pub async fn obs_gateway_websocket_handler( CommonServerError::ServerError })?; - let mut client = TwitchWebsocketClient::new().unwrap(); - - info!("Connecting to Twitch PubSub..."); - client.connect().await.unwrap(); - -// info!("Connected to Twitch PubSub"); -// -// info!("Sending Twitch PubSub PING..."); -// client.send_ping().await.unwrap(); -// -// info!("Try read next from Twitch PubSub..."); -// let r = client.try_next().await.unwrap(); -// info!("Twitch PubSub Result: {:?}", r); -// -// info!("Begin TwitchPubSub LISTEN on authenticated OAuth topics..."); -// let topics = build_pubsub_topics_for_user(user_id); -// client.listen(&auth_token, &topics).await.unwrap(); - - //let (tx, rx) = crossbeam::channel::bounded(1); - //let client = Arc::new(&self.twitch_client); - //let client2 = client.clone(); - //let res = rx.recv().unwrap(); - info!("Begin Javascript WebSocket..."); let server_state_arc = server_state.get_ref().clone(); let websocket = ObsGatewayWebSocket::new( twitch_user_id.clone(), - client, server_state_arc ); @@ -135,22 +111,16 @@ pub async fn obs_gateway_websocket_handler( /// Websocket behavior struct ObsGatewayWebSocket { twitch_user_id: TwitchUserId, - twitch_thread: Arc, server_state: Arc, } impl ObsGatewayWebSocket { fn new( twitch_user_id: TwitchUserId, - twitch_client: TwitchWebsocketClient, server_state: Arc, ) -> Self { - let twitch_thread = Arc::new(ObsTwitchThread::new( - twitch_user_id.clone(), - twitch_client)); Self { twitch_user_id, - twitch_thread, server_state, } } @@ -160,29 +130,6 @@ impl Actor for ObsGatewayWebSocket { type Context = ws::WebsocketContext; fn started(&mut self, _ctx: &mut Self::Context) { - let handle = Handle::current(); - let twitch_thread = self.twitch_thread.clone(); - - - //let future = self.twitch_thread.run_until_exit(); - ////let now_future = Delay::new(Duration::from_secs(5)); - //warn!("Starting thread..."); - ////actix_rt::spawn - ////actix_rt::spawn(twitch_thread.run_until_exit()); - - //actix_rt::spawn(future.map(|x| { - // println!("waited for 5 secs"); - //})); - - //handle.spawn_blocking(result); -// handle.spawn_blocking(async { -// warn!("inside thread 1"); -// let twitch_thread2 = twitch_thread.clone(); -// //async move { -// // warn!("inside thread 2"); -// // twitch_thread2.run_until_exit().await; -// //}.await; -// }); warn!("Thread started"); } } @@ -194,29 +141,6 @@ impl StreamHandler> for ObsGatewayWebSock ctx: &mut Self::Context, ) { if let Ok(msg) = msg { - //info!(">>>>>> obs streamhandler::handle (msg = {:?})", msg); - - //let (tx, rx) = crossbeam::channel::bounded(1); - //let handle = Handle::current(); - - //let client = Arc::new(&self.twitch_client); - //let client2 = client.clone(); - - //handle.spawn(async { - // error!("Twitch PubSub Try read next..."); - // match client2.try_next().await { - // Ok(r) => { - // error!("Twitch PubSub Result: {:?}", r); - // }, - // Err(e) => { - // warn!("pubsub error: {:?}", e); - // } - // } - //}); - - //let res = rx.recv().unwrap(); - - //info!("process message: {:?}", &msg); // TODO: Only send this every 60 seconds. // TODO: Error handling. @@ -230,7 +154,7 @@ impl StreamHandler> for ObsGatewayWebSock CommonServerError::ServerError }).unwrap(); // TODO: FIXME - let channel = RedisKeys::obs_active_session_topic(); + let channel = RedisKeys::obs_active_sessions_topic(); let payload = ObsActivePayload::new(self.twitch_user_id.get_str()); @@ -266,38 +190,3 @@ impl StreamHandler> for ObsGatewayWebSock } } } - -// From the Actix homepage: - -//use actix::{Actor, StreamHandler}; -//use actix_web::{web, App, Error, HttpRequest, HttpResponse, HttpServer}; -//use actix_web_actors::ws; -// -///// Define HTTP actor -//struct MyWs; -// -//impl Actor for MyWs { -// type Context = ws::WebsocketContext; -//} -// -///// Handler for ws::Message message -//impl StreamHandler> for MyWs { -// fn handle( -// &mut self, -// msg: Result, -// ctx: &mut Self::Context, -// ) { -// match msg { -// Ok(ws::Message::Ping(msg)) => ctx.pong(&msg), -// Ok(ws::Message::Text(text)) => ctx.text(text), -// Ok(ws::Message::Binary(bin)) => ctx.binary(bin), -// _ => (), -// } -// } -//} -// -//pub async fn twitch_pubsub_gateway(req: HttpRequest, stream: web::Payload) -> Result { -// let resp = ws::start(MyWs {}, &req, stream); -// println!("{:?}", resp); -// resp -//} diff --git a/crates/service/obs_gateway/src/redis/lease_timeout.rs b/crates/service/obs_gateway/src/redis/constants.rs similarity index 80% rename from crates/service/obs_gateway/src/redis/lease_timeout.rs rename to crates/service/obs_gateway/src/redis/constants.rs index c8bef35d1e..e4410d9ca6 100644 --- a/crates/service/obs_gateway/src/redis/lease_timeout.rs +++ b/crates/service/obs_gateway/src/redis/constants.rs @@ -13,3 +13,7 @@ pub const LEASE_RENEW_PERIOD : Duration = Duration::seconds(30); /// The amount of time to wait before checking or lease is valid. pub const LEASE_CHECK_PERIOD : Duration = Duration::seconds(15); + +/// The OBS websocket polls a key to inform the backend that the +/// user is still active. +pub const OBS_ACTIVE_TTL_SECONDS : usize = 120; diff --git a/crates/service/obs_gateway/src/redis/mod.rs b/crates/service/obs_gateway/src/redis/mod.rs index 9bc863f1de..e8a29762c6 100644 --- a/crates/service/obs_gateway/src/redis/mod.rs +++ b/crates/service/obs_gateway/src/redis/mod.rs @@ -1,3 +1,3 @@ pub mod lease_payload; -pub mod lease_timeout; +pub mod constants; pub mod obs_active_payload; \ No newline at end of file diff --git a/crates/service/obs_gateway/src/threads/listen_for_active_obs_sessions_thread.rs b/crates/service/obs_gateway/src/threads/listen_for_active_obs_sessions_thread.rs index 28b0b9945f..7d0555ce6d 100644 --- a/crates/service/obs_gateway/src/threads/listen_for_active_obs_sessions_thread.rs +++ b/crates/service/obs_gateway/src/threads/listen_for_active_obs_sessions_thread.rs @@ -1,5 +1,6 @@ +use container_common::thread::thread_id::ThreadId; +use crate::redis::constants::LEASE_TIMEOUT_SECONDS; use crate::redis::lease_payload::LeasePayload; -use crate::redis::lease_timeout::LEASE_TIMEOUT_SECONDS; use crate::redis::obs_active_payload::ObsActivePayload; use crate::threads::twitch_pubsub_user_subscriber_thread::TwitchPubsubUserSubscriberThread; use crate::twitch::twitch_user_id::TwitchUserId; @@ -15,7 +16,6 @@ use std::sync::Arc; use std::thread::sleep; use std::time::Duration; use tokio::runtime::Runtime; -use container_common::thread::thread_id::ThreadId; // TODO: Apart from error handling, this looks mostly good. @@ -55,7 +55,7 @@ impl ListenForActiveObsSessionThread { let mut pubsub_pool = self.redis_pubsub_pool.get().unwrap(); let mut pubsub = pubsub_pool.as_pubsub(); - pubsub.subscribe(RedisKeys::obs_active_session_topic()).unwrap(); + pubsub.subscribe(RedisKeys::obs_active_sessions_topic()).unwrap(); loop { let message = pubsub.get_message().unwrap(); diff --git a/crates/service/obs_gateway/src/threads/twitch_pubsub_user_subscriber_thread.rs b/crates/service/obs_gateway/src/threads/twitch_pubsub_user_subscriber_thread.rs index c8a5ea7383..2a798a1a91 100644 --- a/crates/service/obs_gateway/src/threads/twitch_pubsub_user_subscriber_thread.rs +++ b/crates/service/obs_gateway/src/threads/twitch_pubsub_user_subscriber_thread.rs @@ -1,7 +1,7 @@ use container_common::anyhow_result::AnyhowResult; use container_common::thread::thread_id::ThreadId; use crate::redis::lease_payload::LeasePayload; -use crate::redis::lease_timeout::{LEASE_TIMEOUT_SECONDS, LEASE_RENEW_PERIOD, LEASE_CHECK_PERIOD}; +use crate::redis::constants::{LEASE_TIMEOUT_SECONDS, LEASE_RENEW_PERIOD, LEASE_CHECK_PERIOD}; use crate::twitch::constants::TWITCH_PING_CADENCE; use crate::twitch::pubsub::build_pubsub_topics_for_user::build_pubsub_topics_for_user; use crate::twitch::twitch_user_id::TwitchUserId;