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 37aaf0474d..623afec364 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,7 +1,7 @@ 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::{twitch_pubsub_user_subscriber_thread, TwitchPubsubUserSubscriberThread}; +use crate::threads::twitch_pubsub_user_subscriber_thread::TwitchPubsubUserSubscriberThread; use crate::twitch::twitch_user_id::TwitchUserId; use log::error; use log::info; 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 d73cc4a84a..1e35cb50d6 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 @@ -16,6 +16,8 @@ use std::thread::sleep; use std::time::Duration; use twitch_api2::pubsub::Response; use time::Instant; +use container_common::anyhow_result::AnyhowResult; +use crate::twitch::constants::TWITCH_PING_CADENCE; // TODO: ERROR HANDLING // TODO: ERROR HANDLING @@ -35,194 +37,137 @@ pub struct TwitchPubsubUserSubscriberThread { twitch_user_id: TwitchUserId, mysql_pool: Arc>, redis_pool: Arc>, + twitch_websocket_client: TwitchWebsocketClient, - // Thread state - lease_last_renewed_at: Option, + // ========== Thread state ========== + + // The thread must renew the lease, or another worker will pick it up. + // If the lease gets taken by another, we abandon our own workload. + redis_lease_last_renewed_at: Option, + + /// Twitch PubSub requires PINGs at regular intervals, + /// "To keep the server from closing the connection, clients must send a PING + /// command at least once every 5 minutes. If a client does not receive a PONG + /// message within 10 seconds of issuing a PING command, it should reconnect + /// to the server. See details in Handling Connection Failures." + twitch_last_pinged_at: Option, } impl TwitchPubsubUserSubscriberThread { - pub fn new( twitch_user_id: TwitchUserId, mysql_pool: Arc>, redis_pool: Arc>, ) -> Self { + let twitch_websocket_client = TwitchWebsocketClient::new().unwrap(); Self { twitch_user_id, mysql_pool, redis_pool, - lease_last_renewed_at: None, + twitch_websocket_client, + redis_lease_last_renewed_at: None, + twitch_last_pinged_at: None, } } pub async fn start_thread(mut self) { - let mut last_renewed_lease = Instant::now(); + // We lease from outside before starting the thread. + self.redis_lease_last_renewed_at = Some(Instant::now()); - self.foo = Some("asdf".to_string()); + // Lookup the OAuth credentials + let oauth_lookup_result = self.lookup_oauth_record().await; + let oauth_token_record = match oauth_lookup_result { + Err(e) => { + info!("Error looking up oauth creds: {:?}", e); + return; // TODO: Better flow. + } + Ok(None) => { + info!("Twitch user oauth token does not exit. Auth please."); + return; // TODO: Better flow. + } + Ok(Some(record)) => record, + }; + + warn!("Connecting to Twitch PubSub for user {}...", &oauth_token_record.twitch_username); + self.twitch_websocket_client.connect().await.unwrap(); // TODO: Error handling loop { - info!("Twitch subscriber thread"); - - let now = Instant::now(); - let mut should_renew = false; - - match now.checked_sub(LEASE_RENEW_PERIOD) { - None => { should_renew = true } - Some(before) => { - - } - } - - let mut redis = self.redis_pool.get().unwrap(); - - let lease_key = RedisKeys::twitch_pubsub_lease(self.twitch_user_id.get_str()); - let lease = LeasePayload::new("foo", "bar"); - - let serialized = lease.serialize(); - let _v : Option = redis.set_ex( - &lease_key, - &serialized, - LEASE_TIMEOUT_SECONDS - ).unwrap(); - - let result = TwitchOauthTokenFinder::new() - .scope_twitch_user_id(Some(self.twitch_user_id.get_numeric())) - .perform_query(&self.mysql_pool) - .await - .unwrap(); - - let record = match result { - None => { - info!("Twitch user oauth token does not exit. Auth please"); - sleep(Duration::from_secs(5)); - continue; - } - Some(record) => record - }; - - let mut client = TwitchWebsocketClient::new().unwrap(); - //let token_refresher = OauthTokenRefresher::new( - // user_id, - // &token_record.access_token, - // token_record.maybe_refresh_token.as_deref()); - - warn!("Connecting to Twitch PubSub for user {}...", &record.twitch_username); - client.connect().await.unwrap(); - - // TODO: BACKGROUND PINGS UGH! - warn!("Sending ping..."); - client.send_ping().await.unwrap(); + self.maybe_renew_redis_lease().unwrap(); + self.maybe_send_twitch_ping().await.unwrap(); warn!("Listen to user..."); let topics = build_pubsub_topics_for_user(self.twitch_user_id.get_numeric()); - - client.listen(&record.access_token, &topics).await.unwrap(); + self.twitch_websocket_client.listen(&oauth_token_record.access_token, &topics).await.unwrap(); loop { - warn!("client.next()"); - let maybe_event = client.try_next().await.unwrap(); + let maybe_event = self.twitch_websocket_client.try_next() + .await + .unwrap(); - if let Some(event) = maybe_event.as_ref() { - warn!("event: {:?}", event); + match maybe_event { + None => { + sleep(Duration::from_secs(5)); + } + Some(ref event) => { + warn!("event: {:?}", event); - match event { - Response::Response(_) => {} - Response::Message { .. } => {} - Response::Pong => {} - Response::Reconnect => {} + match event { + Response::Response(_) => {} + Response::Message { .. } => {} + Response::Pong => {} + Response::Reconnect => {} + } } } - - sleep(Duration::from_secs(5)); } } } -} + fn maybe_renew_redis_lease(&mut self) -> AnyhowResult<()> { + let mut should_renew_lease = self.redis_lease_last_renewed_at + .map(|last_write| last_write.elapsed().gt(&LEASE_RENEW_PERIOD)) + .unwrap_or(true); - -pub async fn twitch_pubsub_user_subscriber_thread( - twitch_user_id: TwitchUserId, - mysql_pool: Arc>, - redis_pool: Arc>, -) { - - let mut last_renewed_lease = Instant::now(); - - loop { - info!("Twitch subscriber thread"); - - let now = Instant::now(); - let mut should_renew = false; - - match now.checked_sub(LEASE_RENEW_PERIOD) { - None => { should_renew = true } - Some(before) => { - - } + if should_renew_lease { + info!("Renewing Redis Lease for user {}", self.twitch_user_id.get_numeric()); + self.renew_redis_lease()?; + self.redis_lease_last_renewed_at = Some(Instant::now()); } - let mut redis = redis_pool.get().unwrap(); + Ok(()) + } - let lease_key = RedisKeys::twitch_pubsub_lease(twitch_user_id.get_str()); - let lease = LeasePayload::new("foo", "bar"); - - let serialized = lease.serialize(); + fn renew_redis_lease(&mut self) -> AnyhowResult<()> { + // TODO: Error handling + let mut redis = self.redis_pool.get().unwrap(); + let lease_key = RedisKeys::twitch_pubsub_lease(self.twitch_user_id.get_str()); + let lease_value = LeasePayload::new("foo", "bar").serialize(); let _v : Option = redis.set_ex( &lease_key, - &serialized, + &lease_value, LEASE_TIMEOUT_SECONDS ).unwrap(); + Ok(()) + } - let result = TwitchOauthTokenFinder::new() - .scope_twitch_user_id(Some(twitch_user_id.get_numeric())) - .perform_query(&mysql_pool) + async fn lookup_oauth_record(&mut self) -> AnyhowResult> { + TwitchOauthTokenFinder::new() + .scope_twitch_user_id(Some(self.twitch_user_id.get_numeric())) + .perform_query(&self.mysql_pool) .await - .unwrap(); + } - let record = match result { - None => { - info!("Twitch user oauth token does not exit. Auth please"); - sleep(Duration::from_secs(5)); - continue; - } - Some(record) => record - }; + async fn maybe_send_twitch_ping(&mut self) -> AnyhowResult<()> { + let mut should_send_ping = self.twitch_last_pinged_at + .map(|last_ping| last_ping.elapsed().gt(&TWITCH_PING_CADENCE)) + .unwrap_or(true); - let mut client = TwitchWebsocketClient::new().unwrap(); - //let token_refresher = OauthTokenRefresher::new( - // user_id, - // &token_record.access_token, - // token_record.maybe_refresh_token.as_deref()); - - warn!("Connecting to Twitch PubSub for user {}...", &record.twitch_username); - client.connect().await.unwrap(); - - // TODO: BACKGROUND PINGS UGH! - warn!("Sending ping..."); - client.send_ping().await.unwrap(); - - warn!("Listen to user..."); - let topics = build_pubsub_topics_for_user(twitch_user_id.get_numeric()); - - client.listen(&record.access_token, &topics).await.unwrap(); - - loop { - warn!("client.next()"); - let maybe_event = client.try_next().await.unwrap(); - - if let Some(event) = maybe_event.as_ref() { - warn!("event: {:?}", event); - - match event { - Response::Response(_) => {} - Response::Message { .. } => {} - Response::Pong => {} - Response::Reconnect => {} - } - } - - sleep(Duration::from_secs(5)); + if should_send_ping { + info!("Sending Twitch ping for user {}", self.twitch_user_id.get_numeric()); + self.twitch_websocket_client.send_ping().await?; + self.twitch_last_pinged_at = Some(Instant::now()); } + + Ok(()) } } diff --git a/crates/service/obs_gateway/src/twitch/constants.rs b/crates/service/obs_gateway/src/twitch/constants.rs new file mode 100644 index 0000000000..02b790c58b --- /dev/null +++ b/crates/service/obs_gateway/src/twitch/constants.rs @@ -0,0 +1,11 @@ +use time::Duration; + +/// Twitch PubSub requires PINGs at regular intervals, +/// +/// "To keep the server from closing the connection, clients must send a PING +/// command at least once every 5 minutes. If a client does not receive a PONG +/// message within 10 seconds of issuing a PING command, it should reconnect +/// to the server. See details in Handling Connection Failures." +/// +/// We'll keep this intentionally short so we don't elapse. +pub const TWITCH_PING_CADENCE: Duration = Duration::minutes(3); diff --git a/crates/service/obs_gateway/src/twitch/mod.rs b/crates/service/obs_gateway/src/twitch/mod.rs index 76a8eebc7e..879380886c 100644 --- a/crates/service/obs_gateway/src/twitch/mod.rs +++ b/crates/service/obs_gateway/src/twitch/mod.rs @@ -1,3 +1,4 @@ +pub mod constants; pub mod oauth; pub mod polling_websocket_client; pub mod pubsub;