diff --git a/crates/service/obs_gateway/src/redis/lease_timeout.rs b/crates/service/obs_gateway/src/redis/lease_timeout.rs index c4e1e2d4f8..ed89179035 100644 --- a/crates/service/obs_gateway/src/redis/lease_timeout.rs +++ b/crates/service/obs_gateway/src/redis/lease_timeout.rs @@ -1,7 +1,12 @@ use time::Duration; +// TODO: Think about these timings more. + /// Our Twitch PubSub subscriber threads must renew leases in /// order to be granted an exclusive "lock" on a single user. /// If the underlying Redis key goes away, another thread can /// begin leasing. pub const LEASE_TIMEOUT_SECONDS : usize = 60; + +/// The amount of time to wait before renewing our lease. +pub const LEASE_RENEW_PERIOD: Duration = Duration::seconds(30); 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 031af0d6e4..13c4477bba 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,5 +1,5 @@ use crate::redis::lease_payload::LeasePayload; -use crate::redis::lease_timeout::LEASE_TIMEOUT_SECONDS; +use crate::redis::lease_timeout::{LEASE_TIMEOUT_SECONDS, LEASE_RENEW_PERIOD}; 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; @@ -15,19 +15,42 @@ use std::sync::Arc; use std::thread::sleep; use std::time::Duration; use twitch_api2::pubsub::Response; +use time::Instant; + +// TODO: ERROR HANDLING +// TODO: ERROR HANDLING +// TODO: ERROR HANDLING +// TODO: ERROR HANDLING + +// TODO: Timer to ping(). +// TODO: Timer to renew lease. +// TODO: Publish events back to OBS thread. +// TODO: Disconnect when OBS is done. +// TODO: Refresh oauth token. +// TODO: Handle disconnects. +// TODO: Server+thread IDs -/// Receive both forms of twitch userid pub async fn twitch_pubsub_user_subscriber_thread( twitch_user_id: TwitchUserId, mysql_pool: Arc>, redis_pool: Arc>, ) { - // TODO: Error handling + 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) => { + + } + } + let mut redis = redis_pool.get().unwrap(); let lease_key = RedisKeys::twitch_pubsub_lease(twitch_user_id.get_str());