more structure, but still too rigid

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

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);

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<Mutex<TwitchWebsocketClient>>,
   is_polling: AtomicBool,
 }
 
 impl PollingTwitchWebsocketClient {
   pub fn new() -> AnyhowResult<Self> {
     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",
-      )?;
-
-    });
-     /
-
-
-  }*/
 }