diff --git a/src/main.rs b/src/main.rs index 2c8af70a2f..82e7f304fb 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,29 +1,31 @@ +// TODO: These are temporary - #![allow(unused_imports)] +#![allow(unused_mut)] #![allow(dead_code)] - -extern crate twitch_api2; -use futures_util::{SinkExt, StreamExt}; -extern crate reqwest; - use crate::twitch::secrets::TwitchSecrets; +use crate::twitch::websocket_client::{TwitchWebsocketClient, PollingTwitchWebsocketClient}; use crate::util::anyhow_result::AnyhowResult; +use futures_util::{SinkExt, StreamExt}; +use log::info; +use reqwest::Url; +use std::time::Duration; +use tokio_tungstenite::{connect_async, tungstenite::{Error as TungsteniteError, Result as TungsteniteResult}, connect_async_with_config}; use twitch_api2::TwitchClient; use twitch_api2::helix::channels::GetChannelInformationRequest; +use twitch_api2::pubsub::Topic; use twitch_api2::pubsub; use twitch_oauth2::{AppAccessToken, Scope, TwitchToken, tokens::errors::AppAccessTokenError, ClientId, ClientSecret}; -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, PollingTwitchWebsocketClient}; -use std::time::Duration; pub mod twitch; pub mod util; +const DEFAULT_LOG : &'static str = "debug,rustls=warn,tungstenite=warn"; + fn main() -> AnyhowResult<()> { - println!("Hello World"); + easyenv::init_all_with_default_logging(Some(DEFAULT_LOG)); + + info!("Starting"); use std::error::Error; if let Err(err) = run() { @@ -109,8 +111,16 @@ async fn run() -> AnyhowResult<()> { println!("Connected"); println!("Starting polling thread..."); - client.polling_thread().await; + client.start_ping_thread().await; + println!("Try read next..."); + client.try_next().await?; + println!("Try read next..."); + client.try_next().await?; + println!("Try read next..."); + client.try_next().await?; + + println!("Sleep..."); std::thread::sleep(Duration::from_millis(30000)); //let text = msg.into_text()?; diff --git a/src/twitch/websocket_client.rs b/src/twitch/websocket_client.rs index a8fa06df82..d4e0f7a4f2 100644 --- a/src/twitch/websocket_client.rs +++ b/src/twitch/websocket_client.rs @@ -1,7 +1,8 @@ use anyhow::anyhow; use anyhow::bail; use crate::util::anyhow_result::AnyhowResult; -use futures_util::{SinkExt, StreamExt}; +use futures_util::{SinkExt, StreamExt, TryStreamExt}; +use log::debug; use reqwest::Url; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex, PoisonError, MutexGuard}; @@ -17,7 +18,9 @@ use twitch_api2::pubsub; // 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\" }"; +const PING_SECONDS : u64 = 1 * 60; // Twitch calls for every 5 minutes, we'll do every 1 minute. pub struct PollingTwitchWebsocketClient { client: Arc>, @@ -41,34 +44,55 @@ impl PollingTwitchWebsocketClient { pub async fn connect(&mut self) -> AnyhowResult<()> { match self.client.lock() { - Err(e) => Err(anyhow!("could not connect: {:?}", e)), + Err(e) => Err(anyhow!("unlock client err: {:?}", e)), Ok(mut client) => { client.connect().await } } } - pub async fn polling_thread(&self) { + /// 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 { - println!("Poll"); - match client.lock() { Ok(mut c) => { - println!("Ping..."); c.send_ping(); - println!("Read response..."); - c.read_response(); + // TODO: Need to read from all call sites due to ordering. + //c.read_response(); + } + Err(_) => { + // TODO } - Err(_) => {} } - - tokio::time::sleep(Duration::from_millis(5000)).await; + 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; @@ -154,21 +178,20 @@ impl TwitchWebsocketClient { 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. + debug!("Sending ping"); let message = Message::Text(PING_COMMAND.to_string()); socket.send(message).await?; - match self.read_response().await { - Err(e) => Err(e), - Ok(Response::Pong) => Ok(()), - _ => Err(bail!("wrong response type")), - } + // TODO: Need to read from all call sites due to ordering. + //match self.read_response().await { + // Err(e) => Err(e), + // Ok(Response::Pong) => Ok(()), + // _ => Err(bail!("wrong response type")), + //} + Ok(()) } - async fn read_response(&mut self) -> AnyhowResult { + pub async fn next(&mut self) -> AnyhowResult { let socket = match self.socket { None => return Err(anyhow!("not connected")), Some(ref mut s) => s, @@ -184,4 +207,21 @@ impl TwitchWebsocketClient { }, } } -} \ No newline at end of file + + pub async fn try_next(&mut self) -> AnyhowResult> { + let socket = match self.socket { + None => return Err(anyhow!("not connected")), + Some(ref mut s) => s, + }; + + match socket.try_next().await { + Err(e) => Err(anyhow!("websocket error: {:?}", e)), + Ok(None) => Ok(None), + Ok(Some(response)) => { + let text = response.to_text()?; + let typed_response = Response::parse(text)?; + Ok(Some(typed_response)) + }, + } + } +}