From 795f1a7b45b8caabc04f2b9a864995f026a809a4 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Fri, 21 Jan 2022 00:19:29 -0500 Subject: [PATCH] cleanup --- crates/service/obs_gateway/Cargo.toml | 5 +- .../obs_gateway_websocket_handler.rs | 9 -- .../src/endpoints_ws/obs_twitch_thread.rs | 5 - crates/service/obs_gateway/src/main.rs | 120 -------------- .../twitch_pubsub_user_subscriber_thread.rs | 20 +-- crates/service/obs_gateway/src/twitch/mod.rs | 2 - .../obs_gateway/src/twitch/oauth/mod.rs | 3 +- .../src/twitch/oauth/oauth_token_refresher.rs | 24 --- .../src/twitch/polling_websocket_client.rs | 146 ------------------ .../src/twitch/twitch_client_wrapper.rs | 65 -------- .../obs_gateway/src/twitch_pubsub_job.rs | 20 ++- 11 files changed, 13 insertions(+), 406 deletions(-) delete mode 100644 crates/service/obs_gateway/src/twitch/oauth/oauth_token_refresher.rs delete mode 100644 crates/service/obs_gateway/src/twitch/polling_websocket_client.rs delete mode 100644 crates/service/obs_gateway/src/twitch/twitch_client_wrapper.rs diff --git a/crates/service/obs_gateway/Cargo.toml b/crates/service/obs_gateway/Cargo.toml index b1b7e94f6d..4c5ecb3fa0 100644 --- a/crates/service/obs_gateway/Cargo.toml +++ b/crates/service/obs_gateway/Cargo.toml @@ -9,7 +9,7 @@ authors = [ publish = false [[bin]] -name = "obs-gateway-server" +name = "obs-gateway" path = "src/main.rs" [[bin]] @@ -26,19 +26,18 @@ http_server_common = { path = "../../lib/http_server_common" } redis_common = { path = "../../lib/redis_common" } # External -futures-timer = "3.0.2" anyhow = "1.0.40" base64 = "0.13.0" chrono = { version = "0.4.19", features = ["serde"] } derive_more = "0.99.14" dotenv = "0.15.0" easyenv = "0.0.6" +futures-timer = "3.0.2" hex = "0.4.3" hostname = "0.3.1" lazy_static = "1.4.0" lfu = "0.2.5" log = "0.4.14" -#newrelic-telemetry = { path = "../../../vendor/newrelic-telemetry-sdk-rust" } rand = "0.8.3" regex = "1.5.4" time = "0.2.26" 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 d8ddd361c8..e706062ea7 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 @@ -7,8 +7,6 @@ use container_common::token::random_crockford_token::random_crockford_token; use crate::endpoints_ws::obs_twitch_thread::ObsTwitchThread; use crate::redis::obs_active_payload::ObsActivePayload; 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; @@ -95,10 +93,6 @@ pub async fn obs_gateway_websocket_handler( })?; let mut client = TwitchWebsocketClient::new().unwrap(); - let token_refresher = OauthTokenRefresher::new( - twitch_user_id.get_numeric(), - &token_record.access_token, - token_record.maybe_refresh_token.as_deref()); info!("Connecting to Twitch PubSub..."); client.connect().await.unwrap(); @@ -128,7 +122,6 @@ pub async fn obs_gateway_websocket_handler( let websocket = ObsGatewayWebSocket::new( twitch_user_id.clone(), client, - token_refresher, server_state_arc ); @@ -150,12 +143,10 @@ impl ObsGatewayWebSocket { fn new( 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.clone(), - oauth_token_refresher, twitch_client)); Self { twitch_user_id, 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 73fa9101c5..140b2ca7ae 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,5 +1,3 @@ -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; @@ -29,7 +27,6 @@ pub struct ObsTwitchThread { struct InnerData { twitch_user_id: TwitchUserId, - token_refresher: OauthTokenRefresher, is_connected: bool, twitch_client: TwitchWebsocketClient, last_ping: SystemTime, @@ -38,14 +35,12 @@ struct InnerData { impl ObsTwitchThread { pub fn new( twitch_user_id: TwitchUserId, - token_refresher: OauthTokenRefresher, twitch_client: TwitchWebsocketClient ) -> Self { let now = Instant::now(); Self { inner_data: Arc::new(Mutex::new(InnerData { twitch_user_id, - token_refresher, is_connected: false, twitch_client, last_ping: UNIX_EPOCH, diff --git a/crates/service/obs_gateway/src/main.rs b/crates/service/obs_gateway/src/main.rs index 0641a5239e..e53c8e74c8 100644 --- a/crates/service/obs_gateway/src/main.rs +++ b/crates/service/obs_gateway/src/main.rs @@ -32,8 +32,6 @@ use crate::endpoints::oauth_begin_redirect::oauth_begin_enroll_redirect; use crate::endpoints::oauth_end::oauth_end_enroll_from_redirect; use crate::endpoints_ws::obs_gateway_websocket_handler::obs_gateway_websocket_handler; use crate::server_state::{ObsGatewayServerState, EnvConfig, TwitchOauthSecrets, TwitchOauthTemp, BackendsConfig}; -use crate::twitch::polling_websocket_client::PollingTwitchWebsocketClient; -use crate::twitch::twitch_client_wrapper::TwitchClientWrapper; use crate::twitch::twitch_secrets::TwitchSecrets; use futures::Future; use futures::executor::ThreadPool; @@ -234,121 +232,3 @@ pub async fn serve(server_state: ObsGatewayServerState) -> AnyhowResult<()> Ok(()) } - - - -// ======================================================== - -// TODO: Temp comment out to debug actix-web vs. runtime -// -//info!("Getting app access token..."); - -////let scopes = Scope::all(); -//let scopes = vec![ -// Scope::BitsRead, -// Scope::UserReadEmail, -//]; - -//let mut twitch_client = TwitchClientWrapper::new(client_id.clone(), client_secret.clone()); -//twitch_client.request_access_token(scopes).await?; - -//info!("Getting user id ..."); - -////let user_id = twitch_client.get_user_id_from_username("testytest512").await?; -//let user_id = twitch_client.get_user_id_from_username("vocodes").await?; - -//info!("User ID: {}", user_id); - -//std::thread::sleep(Duration::from_secs(5000)); - - -// // ==================== OAUTH FLOW ==================== - - - -// -// let auth_token = user_token.access_token.as_str(); -// -// // ==================== PUBSUB SUBSCRIPTION AND MAIN LOOP ==================== -// -// let mut client = PollingTwitchWebsocketClient::new()?; -// -// println!("Connecting PubSub..."); -// client.connect().await?; -// -// println!("Connected"); -// -// //println!("Starting polling thread..."); -// //client.start_ping_thread().await; -// -// println!("Sending PING..."); -// -// client.send_ping().await?; -// -// println!("Try read next..."); -// let r = client.try_next().await?; -// println!("Result: {:?}", r); -// -// -// println!("Begin LISTEN..."); -// -// let bit_topic = pubsub::channel_bits::ChannelBitsEventsV2 { -// channel_id: user_id, -// }.into_topic(); -// -// let topics = [bit_topic]; -// -// client.listen(&auth_token, &topics).await?; -// -// println!("Try read next..."); -// let r = client.try_next().await?; -// println!("Result: {:?}", r); -// -// /* -// Result: Some(Message { data: ChannelBitsEventsV2 { topic: ChannelBitsEventsV2 { -// channel_id: 652567283 }, reply: BitsEvent { data: BitsEventData { badge_entitlement: -// Some(BadgeEntitlement { new_version: 100, previous_version: 1 }), bits_used: 100, -// channel_id: "652567283", channel_name: "vocodes", chat_message: "Cheer100 testing the cheer", -// context: Cheer, is_anonymous: false, time: "2021-09-27T04:30:53.627717085Z", -// total_bits_used: 101, user_id: "650154491", user_name: "testytest512" }, -// message_id: "793f745e-8f3e-5f71-bc41-4808c4b49a53", version: "1.0", is_anonymous: false } } }) -// -// Result: Some(Message { data: ChannelBitsEventsV2 { topic: ChannelBitsEventsV2 { -// channel_id: 652567283 }, reply: BitsEvent { data: BitsEventData { badge_entitlement: None, -// bits_used: 1, channel_id: "652567283", channel_name: "vocodes", -// chat_message: "mario: hello everybody Cheer1", context: Cheer, is_anonymous: false, -// time: "2021-09-27T04:47:32.943872952Z", total_bits_used: 102, user_id: "650154491", -// user_name: "testytest512" }, message_id: "5844328d-495b-5585-9d6c-3c99949f9a0a", version: "1.0", -// is_anonymous: false } } }) -// */ -// -// -// -// println!("Try read next..."); -// let r = client.try_next().await?; -// println!("Result: {:?}", r); -// -// println!("Sleep..."); -// std::thread::sleep(Duration::from_millis(30000)); -// -// //let text = msg.into_text()?; -// //println!("Text: {}", text); -// -// // ======================================================== - -// info!("Connecting to database..."); -// -// let db_connection_string = -// easyenv::get_env_string_or_default( -// "MYSQL_URL", -// DEFAULT_MYSQL_CONNECTION_STRING); -// -// let pool = MySqlPoolOptions::new() -// .max_connections(5) -// .connect(&db_connection_string) -// .await?; -// -// let redis_manager = RedisConnectionManager::new(redis_connection_string.clone())?; -// -// let redis_pool = r2d2::Pool::builder() -// .build(redis_manager)?; 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 6bd9029094..742600bf62 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 @@ -24,9 +24,7 @@ use crate::twitch::constants::TWITCH_PING_CADENCE; // TODO: ERROR HANDLING // TODO: ERROR HANDLING -// TODO: Timer to ping(). -// TODO: Timer to renew lease. -// TODO: Publish events back to OBS thread. +// TODO: Publish events back to OBS thread (keepalive redis key from websocket). // TODO: Disconnect when OBS is done. // TODO: Refresh oauth token. // TODO: Handle disconnects. @@ -137,24 +135,8 @@ impl TwitchPubsubUserSubscriberThread { sleep(Duration::from_secs(1)); -// match maybe_event { -// None => { -// } -// Some(ref event) => { -// warn!("event: {:?}", event); -// -// match event { -// Response::Response(_) => {} -// Response::Message { .. } => {} -// Response::Pong => {} -// Response::Reconnect => {} -// } -// } -// } - self.maybe_renew_redis_lease().unwrap(); self.maybe_send_twitch_ping().await.unwrap(); - } } } diff --git a/crates/service/obs_gateway/src/twitch/mod.rs b/crates/service/obs_gateway/src/twitch/mod.rs index 879380886c..98cfdf49c5 100644 --- a/crates/service/obs_gateway/src/twitch/mod.rs +++ b/crates/service/obs_gateway/src/twitch/mod.rs @@ -1,8 +1,6 @@ pub mod constants; pub mod oauth; -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/oauth/mod.rs b/crates/service/obs_gateway/src/twitch/oauth/mod.rs index 46b2d4c4f9..c406e2678c 100644 --- a/crates/service/obs_gateway/src/twitch/oauth/mod.rs +++ b/crates/service/obs_gateway/src/twitch/oauth/mod.rs @@ -1,2 +1 @@ -pub mod oauth_token_builder; -pub mod oauth_token_refresher; \ No newline at end of file +pub mod oauth_token_builder; \ No newline at end of file diff --git a/crates/service/obs_gateway/src/twitch/oauth/oauth_token_refresher.rs b/crates/service/obs_gateway/src/twitch/oauth/oauth_token_refresher.rs deleted file mode 100644 index 35d5c86149..0000000000 --- a/crates/service/obs_gateway/src/twitch/oauth/oauth_token_refresher.rs +++ /dev/null @@ -1,24 +0,0 @@ -// TODO: I know this is something we'll need, but I haven't built it yet. -// We'll probably want to write back to the database. - -#[derive(Clone)] -pub struct OauthTokenRefresher { - user_id: u32, - auth_token: String, - refresh_token: Option, -} - -impl OauthTokenRefresher { - - pub fn new(user_id: u32, auth_token: &str, refresh_token: Option<&str>) -> Self { - Self { - user_id, - auth_token: auth_token.to_string(), - refresh_token: refresh_token.map(|str| str.to_string()), - } - } - - pub fn get_auth_token(&self) -> String { - self.auth_token.clone() - } -} diff --git a/crates/service/obs_gateway/src/twitch/polling_websocket_client.rs b/crates/service/obs_gateway/src/twitch/polling_websocket_client.rs deleted file mode 100644 index 4a4cfaf2d7..0000000000 --- a/crates/service/obs_gateway/src/twitch/polling_websocket_client.rs +++ /dev/null @@ -1,146 +0,0 @@ -use anyhow::anyhow; -use container_common::anyhow_result::AnyhowResult; -use crate::twitch::websocket_client::TwitchWebsocketClient; -use std::sync::atomic::AtomicBool; -use std::sync::{Arc, Mutex}; -use std::time::Duration; -use twitch_api2::pubsub::{Response, TopicData, Topic, Topics}; - -const PING_SECONDS : u64 = 1 * 60; // Twitch calls for every 5 minutes, we'll do every 1 minute. - -pub struct PollingTwitchWebsocketClient { - client: Arc>, - is_polling: AtomicBool, -} - -impl PollingTwitchWebsocketClient { - pub fn new() -> AnyhowResult { - Ok(Self { - client: Arc::new(Mutex::new(TwitchWebsocketClient::new()?)), - is_polling: AtomicBool::new(false), - }) - } - - pub async fn connect(&mut self) -> AnyhowResult<()> { - match self.client.lock() { - Err(e) => Err(anyhow!("unlock client err: {:?}", e)), - Ok(mut client) => { - client.connect().await - } - } - } - - pub async fn listen<'t>(&mut self, auth_token: &str, topics: &'t [Topics]) -> AnyhowResult<()> { - match self.client.lock() { - Err(e) => Err(anyhow!("unlock client err: {:?}", e)), - Ok(mut client) => { - client.listen(auth_token, topics).await - } - } - } - - pub async fn send_ping(&mut self) -> AnyhowResult<()> { - match self.client.lock() { - Err(e) => Err(anyhow!("unlock client err: {:?}", e)), - Ok(mut client) => { - client.send_ping().await - } - } - } - - /// Start a pinging background thread. - /// Per Twitch PubSub docs, - /// 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. - pub async fn start_ping_thread(&self) { - let client = self.client.clone(); - tokio::spawn(async move { - loop { - match client.lock() { - Ok(mut c) => { - // TODO: Future isn't Send - //let _r = c.send_ping().await; - // TODO: Need to read from all call sites due to ordering. - //c.read_response(); - } - Err(_) => { - // TODO - } - } - tokio::time::sleep(Duration::from_secs(PING_SECONDS)).await; - } - }); - } - - pub async fn next(&mut self) -> AnyhowResult { - match self.client.lock() { - Err(e) => Err(anyhow!("unlock client err: {:?}", e)), - Ok(mut client) => { - client.next().await - } - } - } - - pub async fn try_next(&mut self) -> AnyhowResult> { - match self.client.lock() { - Err(e) => Err(anyhow!("unlock client err: {:?}", e)), - Ok(mut client) => { - client.try_next().await - } - } - } - - /*async fn start_polling(&self) { - if self.is_polling.load(Ordering::Relaxed) { - return; - } - - tokio::spawn(async move { - self.is_polling.store(true, Ordering::Relaxed); - - let command = pubsub::channel_sub_gifts( - &topics, - "authtoken", - "random nonce string", - )?; - - }); - / - }*/ - - - /*let message = Message::Text(PING_COMMAND.to_string()); - socket.send(message).await?; - match self.read_response().await { - Err(e) => println!("Error getting PONG: {:?}", e), - Ok(typed_response) => { - match typed_response { - Response::Pong => { println!("Pong") } - Response::Reconnect => { println!("Reconnect") } - Response::Response(_0) => { println!("Sub/Unsub (pubsub?)") } - Response::Message { data } => { - match data { - TopicData::ChannelCheerEventsPublicV1 { topic, reply } => {} - TopicData::ChannelSubGiftsV1 { topic, reply } => {} - TopicData::ChannelSubscribeEventsV1 { topic, reply } => {} - TopicData::CommunityPointsChannelV1 { topic, reply } => {} - TopicData::Following { topic, reply} => {} - // Not interested in these events - TopicData::AutoModQueue { .. } => {} - TopicData::ChannelBitsBadgeUnlocks { .. } => {} - TopicData::ChannelBitsEventsV2 { .. } => {} - TopicData::ChannelPointsChannelV1 { .. } => {} - TopicData::ChatModeratorActions { .. } => {} - TopicData::HypeTrainEventsV1 { .. } => {} - TopicData::HypeTrainEventsV1Rewards { .. } => {} - TopicData::Raid { .. } => {} - TopicData::UserModerationNotifications { .. } => {} - TopicData::VideoPlayback { .. } => {} - TopicData::VideoPlaybackById { .. } => {} - } - } - } - } - }*/ -} diff --git a/crates/service/obs_gateway/src/twitch/twitch_client_wrapper.rs b/crates/service/obs_gateway/src/twitch/twitch_client_wrapper.rs deleted file mode 100644 index 881584e87b..0000000000 --- a/crates/service/obs_gateway/src/twitch/twitch_client_wrapper.rs +++ /dev/null @@ -1,65 +0,0 @@ -//! The Helix client is independent of PubSub -//! This just wraps `twitch_api2` for more convenience. - -use anyhow::anyhow; -use container_common::anyhow_result::AnyhowResult; -use twitch_api2::TwitchClient; -use twitch_api2::helix::users::GetUsersRequest; -use twitch_api2::types::{Nickname, UserId}; -use twitch_oauth2::{ClientSecret, ClientId, AccessToken, AppAccessToken, Scope}; - -pub struct TwitchClientWrapper<'a> { - client_id: ClientId, - client_secret: ClientSecret, - twitch_client: TwitchClient<'a, reqwest::Client>, - access_token: Option, -} - -impl <'a> TwitchClientWrapper <'a> { - pub fn new(client_id: ClientId, client_secret: ClientSecret) -> Self { - Self { - client_id, - client_secret, - twitch_client: TwitchClient::default(), - access_token: None, - } - } - - pub async fn request_access_token(&mut self, scopes: Vec) -> AnyhowResult<()> { - self.get_access_token(scopes).await?; - Ok(()) - } - - async fn get_access_token(&mut self, scopes: Vec) -> AnyhowResult<()> { - let mut http_client = reqwest::Client::new(); - let access_token = twitch_oauth2::AppAccessToken::get_app_access_token( - &http_client, - self.client_id.clone(), - self.client_secret.clone(), - scopes, - ).await?; - - self.access_token = Some(access_token); - Ok(()) - } - - /// We frequently need to look up IDs from usernames. - pub async fn get_user_id_from_username(&self, username: &str) -> AnyhowResult { - let access_token = match self.access_token { - None => return Err(anyhow!("no access token has been granted")), - Some(ref token) => token, - }; - - let req = GetUsersRequest::builder() - .login(vec![Nickname::new(username)]) - .build(); - - let response = &self.twitch_client.helix.req_get(req, access_token) - .await?; - - match response.data.get(0) { - None => Err(anyhow!("No such user: `{}`.", username)), - Some(user) => Ok(user.id.as_str().parse::()?), - } - } -} diff --git a/crates/service/obs_gateway/src/twitch_pubsub_job.rs b/crates/service/obs_gateway/src/twitch_pubsub_job.rs index 58920e2a59..d280de4c47 100644 --- a/crates/service/obs_gateway/src/twitch_pubsub_job.rs +++ b/crates/service/obs_gateway/src/twitch_pubsub_job.rs @@ -14,22 +14,20 @@ use config::shared_constants::{DEFAULT_RUST_LOG, DEFAULT_REDIS_DATABASE_1_CONNECTION_STRING, DEFAULT_MYSQL_CONNECTION_STRING}; use container_common::anyhow_result::AnyhowResult; +use crate::threads::listen_for_active_obs_sessions_thread::listen_for_active_obs_session_thread; +use crate::twitch::websocket_client::TwitchWebsocketClient; use futures::executor::{ThreadPool, ThreadPoolBuilder}; -use r2d2_redis::r2d2; +use futures::task::SpawnExt; use log::info; +use r2d2_redis::RedisConnectionManager; +use r2d2_redis::r2d2::Pool; +use r2d2_redis::r2d2; +use redis_common::redis_keys::RedisKeys; +use sqlx::mysql::MySqlPoolOptions; +use std::sync::Arc; use std::thread::sleep; use std::time::Duration; -use crate::twitch::twitch_client_wrapper::TwitchClientWrapper; -use crate::twitch::websocket_client::TwitchWebsocketClient; -use futures::task::SpawnExt; use tokio::runtime::{Builder, Runtime}; -use r2d2_redis::RedisConnectionManager; -use std::sync::Arc; -use r2d2_redis::r2d2::Pool; -use redis_common::redis_keys::RedisKeys; -use crate::threads::listen_for_active_obs_sessions_thread::listen_for_active_obs_session_thread; -use sqlx::mysql::MySqlPoolOptions; - pub mod redis; pub mod threads; pub mod twitch;