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 22de958c0b..d8ddd361c8 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 @@ -65,7 +65,7 @@ pub async fn obs_gateway_websocket_handler( }, }; - let twitch_user_id = TwitchUserId::from_string(&token_record.twitch_user_id) + let twitch_user_id = TwitchUserId::from_str(&token_record.twitch_user_id) .map_err(|e| { error!("Error converting twitch user id: {}, id= {}", e, &token_record.twitch_user_id); CommonServerError::ServerError 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 8193a7e86d..1b000ab504 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 @@ -14,6 +14,7 @@ use std::sync::Arc; use std::thread::sleep; use std::time::Duration; use tokio::runtime::Runtime; +use crate::twitch::twitch_user_id::TwitchUserId; pub async fn listen_for_active_obs_session_thread( mysql_pool: Arc>, @@ -27,26 +28,13 @@ pub async fn listen_for_active_obs_session_thread( let channel = RedisKeys::obs_active_session_topic(); pubsub.subscribe(channel).unwrap(); - let mut count = 0; - - //loop { - // let payload : String = try!(msg.get_payload()); - // println!("channel '{}': {}", msg.get_channel_name(), payload); - //} - loop { - //info!("[PubSub]"); - // TODO: ERROR HANDLING let message = pubsub.get_message().unwrap(); let payload : String = message.get_payload().unwrap(); - //info!("Message: {}", payload); - let payload = ObsActivePayload::from_json_str(&payload).unwrap(); - //info!("Message decoded: {:?}", payload); - let lease_key = RedisKeys::twitch_pubsub_lease(&payload.twitch_user_id); let mut redis = redis_pool.get().unwrap(); @@ -54,11 +42,7 @@ pub async fn listen_for_active_obs_session_thread( let lease_value : Option = redis.get(&lease_key).unwrap(); if let Some(value) = lease_value.as_deref() { - //info!("Already has lease."); let lease = LeasePayload::deserialize(value).unwrap(); - - //info!("Lease value: {:?}", &lease); - continue; } @@ -73,65 +57,11 @@ pub async fn listen_for_active_obs_session_thread( LEASE_TIMEOUT_SECONDS ).unwrap(); - let twitch_user_id = payload.twitch_user_id.clone(); + let twitch_user_id = TwitchUserId::from_str(&payload.twitch_user_id).unwrap(); + let redis_pool2 = redis_pool.clone(); let mysql_pool2 = mysql_pool.clone(); runtime.spawn(twitch_pubsub_user_subscriber_thread(twitch_user_id, mysql_pool2, redis_pool2)); - - - // Publish: (ActiveSession, user_id) - // - On: First connect - // - Every 30 seconds - // - 1,000 users x 10 servers = 10,000 = 333 QPS - // - 100 users x 5 servers = 500 = 16 QPS - // Redis Benchmarks - // SET: 552028.75 requests per second (high spec machine) - // GET: 707463.75 requests per second (high spec machine) - // SET: 100007 requests completed in 0.88 seconds (no pipeline) - // GET: 100000 requests completed in 1.23 seconds (no pipeline) - - // Lease Key w/ 60s TTL: "ActiveSessionLease:{user_id}" - // Lease Value: "{serverNonce}:{threadNonce}" (to kill same-server dupes) - - - - - // Data structures: - // - HashSet of user ids (maybe not needed?) - // - Random threads (don't need to be tracked) - // - Redis leases that expire every 30 seconds - 1 minute - // --- But polling these leases in a big cluster is high QPS - - // > New UserId: - // If not present in HashSet, try to create a Redis lease - // If lease already taken, ignore. - // If lease created, create a new thread. - // Thread continually renews lease (every 10 seconds) - // Puts in a threadlocal random nonce into lease. If differs at - // any point, we know some other process somehow got a lease. Abort. - - - // frontend --> lease key - // -- no, frontend won't know about disconnects, just drought of events - -/* - THIS WORKS - if count < 3 { - info!("Spawning...."); - - let count2 = count; - - runtime.spawn(async move { - let count3 = count2; - loop { - info!(".....spawned..... {}", count3); - sleep(Duration::from_millis(1_000)); - } - }); - } - - count += 1; - */ } } 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 a525187c9e..031af0d6e4 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,6 +1,7 @@ use crate::redis::lease_payload::LeasePayload; use crate::redis::lease_timeout::LEASE_TIMEOUT_SECONDS; use crate::twitch::pubsub::build_pubsub_topics_for_user::build_pubsub_topics_for_user; +use crate::twitch::twitch_user_id::TwitchUserId; use crate::twitch::websocket_client::TwitchWebsocketClient; use database_queries::twitch_oauth::find::{TwitchOauthTokenRecord, TwitchOauthTokenFinder}; use log::info; @@ -17,20 +18,19 @@ use twitch_api2::pubsub::Response; /// Receive both forms of twitch userid pub async fn twitch_pubsub_user_subscriber_thread( - twitch_user_id: String, + twitch_user_id: TwitchUserId, mysql_pool: Arc>, redis_pool: Arc>, ) { // TODO: Error handling - let numeric_twitch_user_id = twitch_user_id.parse::().unwrap(); loop { info!("Twitch subscriber thread"); let mut redis = redis_pool.get().unwrap(); - let lease_key = RedisKeys::twitch_pubsub_lease(&twitch_user_id); + let lease_key = RedisKeys::twitch_pubsub_lease(twitch_user_id.get_str()); let lease = LeasePayload::new("foo", "bar"); let serialized = lease.serialize(); @@ -41,7 +41,7 @@ pub async fn twitch_pubsub_user_subscriber_thread( ).unwrap(); let result = TwitchOauthTokenFinder::new() - .scope_twitch_user_id(Some(numeric_twitch_user_id)) + .scope_twitch_user_id(Some(twitch_user_id.get_numeric())) .perform_query(&mysql_pool) .await .unwrap(); @@ -69,7 +69,7 @@ pub async fn twitch_pubsub_user_subscriber_thread( client.send_ping().await.unwrap(); warn!("Listen to user..."); - let topics = build_pubsub_topics_for_user(numeric_twitch_user_id); + let topics = build_pubsub_topics_for_user(twitch_user_id.get_numeric()); client.listen(&record.access_token, &topics).await.unwrap(); diff --git a/crates/service/obs_gateway/src/twitch/twitch_user_id.rs b/crates/service/obs_gateway/src/twitch/twitch_user_id.rs index d3d75f23ef..a6a5022899 100644 --- a/crates/service/obs_gateway/src/twitch/twitch_user_id.rs +++ b/crates/service/obs_gateway/src/twitch/twitch_user_id.rs @@ -10,7 +10,7 @@ pub struct TwitchUserId { impl TwitchUserId { - pub fn from_string(twitch_user_id: &str) -> AnyhowResult { + pub fn from_str(twitch_user_id: &str) -> AnyhowResult { let numeric_id = twitch_user_id.parse::()?; Ok(Self { string_user_id: twitch_user_id.to_string(),