From 00fef2a2d2430259f904bbe2fd7cca77f17d9d90 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Sun, 26 Sep 2021 18:16:22 -0400 Subject: [PATCH] more structure, but still too rigid --- Cargo.toml | 2 +- src/main.rs | 18 +++- src/twitch/websocket_client.rs | 165 +++++++++++++++++++++------------ 3 files changed, 121 insertions(+), 64 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index a7847ce405..0a57ac5219 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,7 +13,7 @@ path = "src/main.rs" [dependencies] # Async runtime, Futures, etc. -tokio = { version = "1.10.0", features = ["rt-multi-thread", "macros"] } +tokio = { version = "1.10.0", features = ["rt-multi-thread", "macros", "time"] } futures-util = "0.3.17" # Crypto diff --git a/src/main.rs b/src/main.rs index 9466c213a0..2c8af70a2f 100644 --- a/src/main.rs +++ b/src/main.rs @@ -16,7 +16,8 @@ use twitch_api2::pubsub::Topic; use tokio_tungstenite::{connect_async, tungstenite::{Error as TungsteniteError, Result as TungsteniteResult}, connect_async_with_config}; use reqwest::Url; -use crate::twitch::websocket_client::TwitchWebsocketClient; +use crate::twitch::websocket_client::{TwitchWebsocketClient, PollingTwitchWebsocketClient}; +use std::time::Duration; pub mod twitch; pub mod util; @@ -93,13 +94,24 @@ async fn run() -> AnyhowResult<()> { // To parse the websocket messages, use pubsub::Response::parse - println!("Connected"); + /*println!("Connected"); let mut ws_client = TwitchWebsocketClient::new()?; ws_client.connect().await?; ws_client.send_ping().await?; - println!("success ping"); + println!("success ping");*/ + + let mut client = PollingTwitchWebsocketClient::new()?; + + println!("Connecting..."); + client.connect().await?; + println!("Connected"); + + println!("Starting polling thread..."); + client.polling_thread().await; + + std::thread::sleep(Duration::from_millis(30000)); //let text = msg.into_text()?; //println!("Text: {}", text); diff --git a/src/twitch/websocket_client.rs b/src/twitch/websocket_client.rs index 2c171b1409..a8fa06df82 100644 --- a/src/twitch/websocket_client.rs +++ b/src/twitch/websocket_client.rs @@ -4,7 +4,10 @@ use crate::util::anyhow_result::AnyhowResult; use futures_util::{SinkExt, StreamExt}; use reqwest::Url; use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex, PoisonError, MutexGuard}; +use std::time::Duration; use tokio::net::TcpStream; +use tokio; use tokio_tungstenite::tungstenite::Message; use tokio_tungstenite::{connect_async, WebSocketStream, MaybeTlsStream}; use twitch_api2::pubsub::{Response, TopicData}; @@ -17,17 +20,108 @@ const WEBSOCKET_GATEWAY : &'static str = "wss://pubsub-edge.twitch.tv"; const PING_COMMAND : &'static str = "{ \"type\": \"PING\" }"; pub struct PollingTwitchWebsocketClient { - client: TwitchWebsocketClient, + client: Arc>, is_polling: AtomicBool, } impl PollingTwitchWebsocketClient { pub fn new() -> AnyhowResult { Ok(Self { - client: TwitchWebsocketClient::new()?, + client: Arc::new(Mutex::new(TwitchWebsocketClient::new()?)), is_polling: AtomicBool::new(false), }) } + + async fn connect_keepalive(&self) -> AnyhowResult<()> { + + //tokio::spawn(self.polling_thread); + + Ok(()) + } + + pub async fn connect(&mut self) -> AnyhowResult<()> { + match self.client.lock() { + Err(e) => Err(anyhow!("could not connect: {:?}", e)), + Ok(mut client) => { + client.connect().await + } + } + } + + pub async fn polling_thread(&self) { + let client = self.client.clone(); + tokio::spawn(async move { + loop { + println!("Poll"); + + match client.lock() { + Ok(mut c) => { + println!("Ping..."); + c.send_ping(); + println!("Read response..."); + c.read_response(); + } + Err(_) => {} + } + + tokio::time::sleep(Duration::from_millis(5000)).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 { .. } => {} + } + } + } + } + }*/ + } pub struct TwitchWebsocketClient { @@ -43,56 +137,27 @@ impl TwitchWebsocketClient { pub async fn connect(&mut self) -> AnyhowResult<()> { let url = Url::parse(WEBSOCKET_GATEWAY)?; - let (mut socket, _response) = connect_async(url).await?; - - let message = Message::Text(PING_COMMAND.to_string()); - socket.send(message).await?; - self.socket = Some(socket); - - 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 { .. } => {} - } - } - } - } - } - Ok(()) } + pub fn is_connected(&self) -> bool { + // TODO: This probably isn't accurate + self.socket.is_some() + } + pub async fn send_ping(&mut self) -> AnyhowResult<()> { let mut socket = match self.socket { None => return Err(anyhow!("not connected")), Some(ref mut socket) => socket, }; + // TODO: + // 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. let message = Message::Text(PING_COMMAND.to_string()); socket.send(message).await?; @@ -119,24 +184,4 @@ impl TwitchWebsocketClient { }, } } - - /*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", - )?; - - }); - / - - - }*/ } \ No newline at end of file