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 230c13469c..22de958c0b 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 @@ -10,6 +10,7 @@ use crate::server_state::ObsGatewayServerState; use crate::twitch::oauth::oauth_token_refresher::OauthTokenRefresher; use crate::twitch::polling_websocket_client::PollingTwitchWebsocketClient; 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::{TwitchOauthTokenFinder, TwitchOauthTokenRecord}; use futures_timer::Delay; @@ -64,7 +65,7 @@ pub async fn obs_gateway_websocket_handler( }, }; - let user_id = token_record.twitch_user_id.parse::() + let twitch_user_id = TwitchUserId::from_string(&token_record.twitch_user_id) .map_err(|e| { error!("Error converting twitch user id: {}, id= {}", e, &token_record.twitch_user_id); CommonServerError::ServerError @@ -95,7 +96,7 @@ pub async fn obs_gateway_websocket_handler( let mut client = TwitchWebsocketClient::new().unwrap(); let token_refresher = OauthTokenRefresher::new( - user_id, + twitch_user_id.get_numeric(), &token_record.access_token, token_record.maybe_refresh_token.as_deref()); @@ -125,7 +126,7 @@ pub async fn obs_gateway_websocket_handler( let server_state_arc = server_state.get_ref().clone(); let websocket = ObsGatewayWebSocket::new( - user_id, + twitch_user_id.clone(), client, token_refresher, server_state_arc @@ -140,20 +141,22 @@ pub async fn obs_gateway_websocket_handler( /// Websocket behavior struct ObsGatewayWebSocket { - //twitch_client: PollingTwitchWebsocketClient, - twitch_user_id: u32, + twitch_user_id: TwitchUserId, twitch_thread: Arc, server_state: Arc, } impl ObsGatewayWebSocket { fn new( - twitch_user_id: u32, + twitch_user_id: TwitchUserId, twitch_client: TwitchWebsocketClient, oauth_token_refresher: OauthTokenRefresher, server_state: Arc, ) -> Self { - let twitch_thread = Arc::new(ObsTwitchThread::new(twitch_user_id, oauth_token_refresher, twitch_client)); + let twitch_thread = Arc::new(ObsTwitchThread::new( + twitch_user_id.clone(), + oauth_token_refresher, + twitch_client)); Self { twitch_user_id, twitch_thread, @@ -238,8 +241,8 @@ impl StreamHandler> for ObsGatewayWebSock let channel = RedisKeys::obs_active_session_topic(); - let twitch_user_id = self.twitch_user_id.to_string(); - let payload = ObsActivePayload::new(&twitch_user_id); + let payload = ObsActivePayload::new(self.twitch_user_id.get_str()); + let json_payload = payload.serialize() .map_err(|e| { error!("Could not serialize JSON: {:?}", e); diff --git a/crates/service/obs_gateway/src/endpoints_ws/obs_twitch_thread.rs b/crates/service/obs_gateway/src/endpoints_ws/obs_twitch_thread.rs index 1d23df24f5..73fa9101c5 100644 --- a/crates/service/obs_gateway/src/endpoints_ws/obs_twitch_thread.rs +++ b/crates/service/obs_gateway/src/endpoints_ws/obs_twitch_thread.rs @@ -1,15 +1,16 @@ +use crate::twitch::oauth::oauth_token_refresher::OauthTokenRefresher; use crate::twitch::polling_websocket_client::PollingTwitchWebsocketClient; +use crate::twitch::twitch_user_id::TwitchUserId; use crate::twitch::websocket_client::TwitchWebsocketClient; +use futures::lock::Mutex; use log::error; use log::info; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, RwLock, PoisonError, RwLockWriteGuard}; use std::thread::sleep; -use futures::lock::Mutex; -use std::time::{Duration, SystemTime, UNIX_EPOCH}; use std::time::Instant; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; use tokio::runtime::Handle; -use crate::twitch::oauth::oauth_token_refresher::OauthTokenRefresher; // TODO: Let's make this an Arc>> // so that multiple browser sessions for the same user can @@ -27,7 +28,7 @@ pub struct ObsTwitchThread { } struct InnerData { - twitch_user_id: u32, + twitch_user_id: TwitchUserId, token_refresher: OauthTokenRefresher, is_connected: bool, twitch_client: TwitchWebsocketClient, @@ -36,7 +37,7 @@ struct InnerData { impl ObsTwitchThread { pub fn new( - twitch_user_id: u32, + twitch_user_id: TwitchUserId, token_refresher: OauthTokenRefresher, twitch_client: TwitchWebsocketClient ) -> Self { 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 5529b52d2c..a525187c9e 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 @@ -15,6 +15,7 @@ use std::thread::sleep; use std::time::Duration; use twitch_api2::pubsub::Response; +/// Receive both forms of twitch userid pub async fn twitch_pubsub_user_subscriber_thread( twitch_user_id: String, mysql_pool: Arc>, diff --git a/crates/service/obs_gateway/src/twitch/mod.rs b/crates/service/obs_gateway/src/twitch/mod.rs index dadd275d32..76a8eebc7e 100644 --- a/crates/service/obs_gateway/src/twitch/mod.rs +++ b/crates/service/obs_gateway/src/twitch/mod.rs @@ -3,4 +3,5 @@ pub mod polling_websocket_client; pub mod pubsub; pub mod twitch_client_wrapper; pub mod twitch_secrets; +pub mod twitch_user_id; pub mod websocket_client; \ No newline at end of file diff --git a/crates/service/obs_gateway/src/twitch/twitch_user_id.rs b/crates/service/obs_gateway/src/twitch/twitch_user_id.rs new file mode 100644 index 0000000000..d3d75f23ef --- /dev/null +++ b/crates/service/obs_gateway/src/twitch/twitch_user_id.rs @@ -0,0 +1,35 @@ +use container_common::anyhow_result::AnyhowResult; + +/// Twitch says to encode these as strings, but the Twitch libraries +/// still deal in integers. This centralizes type juggling. +#[derive(Clone)] +pub struct TwitchUserId { + string_user_id: String, + numeric_user_id: u32, +} + +impl TwitchUserId { + + pub fn from_string(twitch_user_id: &str) -> AnyhowResult { + let numeric_id = twitch_user_id.parse::()?; + Ok(Self { + string_user_id: twitch_user_id.to_string(), + numeric_user_id: numeric_id, + }) + } + + pub fn from_numeric(twitch_user_id: u32) -> Self { + Self { + string_user_id: twitch_user_id.to_string(), + numeric_user_id: twitch_user_id, + } + } + + pub fn get_numeric(&self) -> u32 { + self.numeric_user_id + } + + pub fn get_str(&self) -> &str { + &self.string_user_id + } +}