diff --git a/src/main.rs b/src/main.rs index dc1ec4fbb3..3eaa665a0c 100644 --- a/src/main.rs +++ b/src/main.rs @@ -12,6 +12,7 @@ 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; pub mod twitch; pub mod util; @@ -91,7 +92,7 @@ async fn run() -> AnyhowResult<()> { println!("Connecting..."); - if let Err(e) = connect_async(&url).await { + /*if let Err(e) = connect_async(&url).await { println!("Error connecting: {:?}", e); } @@ -101,10 +102,14 @@ async fn run() -> AnyhowResult<()> { println!("Connected!"); let msg = socket.next().await.expect("Can't fetch case count")?; - socket.close(None).await?; + socket.close(None).await?;*/ - let text = msg.into_text()?; - println!("Text: {}", text); + println!("Connected"); + let mut ws_client = TwitchWebsocketClient::new()?; + ws_client.connect().await?; + + //let text = msg.into_text()?; + //println!("Text: {}", text); Ok(()) } diff --git a/src/twitch/mod.rs b/src/twitch/mod.rs index f79395be0d..180c2ab938 100644 --- a/src/twitch/mod.rs +++ b/src/twitch/mod.rs @@ -1,2 +1,3 @@ pub mod secrets; -pub mod twitch_client; \ No newline at end of file +pub mod twitch_client; +pub mod websocket_client; \ No newline at end of file diff --git a/src/twitch/websocket_client.rs b/src/twitch/websocket_client.rs new file mode 100644 index 0000000000..05578aadbb --- /dev/null +++ b/src/twitch/websocket_client.rs @@ -0,0 +1,103 @@ +use crate::util::anyhow_result::AnyhowResult; +use reqwest::Url; +use std::sync::atomic::{AtomicBool, Ordering}; +use tokio::net::TcpStream; +use tokio_tungstenite::{connect_async, WebSocketStream, MaybeTlsStream}; +use twitch_api2::pubsub; +use futures_util::{SinkExt, StreamExt}; +use tokio_tungstenite::tungstenite::Message; +use twitch_api2::pubsub::{Response, TopicData}; + +// Reference javascript Twitch Websocket client: +// https://github.com/twitchdev/pubsub-javascript-sample/blob/main/main.js + +const WEBSOCKET_GATEWAY : &'static str = "wss://pubsub-edge.twitch.tv"; +const PING_COMMAND : &'static str = "{ \"type\": \"PING\" }"; + + +pub struct TwitchWebsocketClient { + socket: Option>>, + is_polling: AtomicBool, +} + +impl TwitchWebsocketClient { + + pub fn new() -> AnyhowResult { + Ok(Self { + socket: None, + is_polling: AtomicBool::new(false), + }) + } + + 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?; + + if let Some(response) = socket.next().await { + + if let Ok(r) = response { + let text = r.to_text()?; + println!("Text response: {:?}", text); + + let typed_response = Response::parse(text)?; + + 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 { .. } => {} + } + } + } + } + } + + + self.socket = Some(socket); + + Ok(()) + } + + 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