Merge remote-tracking branch 'old_k/master' into merge-more-old-code

.gitignore

@@ -0,0 +1,8 @@
+*.swo
+*.swp
+*~
+.env
+.idea/
+__pycache__/
+secrets.toml
+target/

Cargo.toml

@@ -0,0 +1,55 @@
+[package]
+name = "twitch-tts-desktop-client"
+version = "0.0.0"
+edition = "2018"
+authors = [
+  "Brandon Thomas <bt@brand.io>",
+  "Brandon Thomas <echelon@gmail.com>"
+]
+
+[[bin]]
+name = "twitch-tts-desktop-client"
+path = "src/main.rs"
+
+[dependencies]
+# Async runtime, Futures, etc.
+tokio = { version = "1.10.0", features = ["rt-multi-thread", "macros", "time"] }
+futures-util = "0.3.17"
+
+# Crypto
+rustls = "0.20.0-beta2"
+
+# Websockets
+tokio-tungstenite = { version = "0.15.0", features = ["rustls", "rustls-tls"] }
+
+# HTTP
+reqwest = { version = "0.11.4" }
+
+# Serialization
+serde = "1.0.130"
+serde_derive = "1.0.130"
+toml = "0.5.8"
+
+# Twitch
+# (This is a fairly mature and robust Twitch chat library)
+#twitchchat = { version = "0.14.8", features = ["tokio", "tokio-util", "async", "webpki-roots", "tokio-rustls"] }
+# Twitch Helix etc. API, but no PubSub just yet
+#twitch_api2 = "0.4.1"
+#twitch_oauth2 = "0.4.1"
+#twitch_api2 = "0.5.0"
+#twitch_oauth2 = "0.5.0"
+# NB "unsupported" pubsub features may disappear from the Twitch API at any time. It's an unofficial API.
+twitch_api2 = { version = "0.6.0-rc.2", features = ["helix", "reqwest_client", "pubsub", "chrono", "unsupported"] }
+twitch_oauth2 = "0.6.0-rc.2"
+
+# Utils
+anyhow = "1.0.44"
+easyenv = "0.0.6"
+log = "0.4.14"
+rand = "0.8.4"
+
+# Temporary
+rpassword = "5.0.1"
+
+#[build-dependencies]
+

README.md

@@ -0,0 +1,5 @@
+twitch-tts-desktop-client
+=========================
+Name TBD
+
+

src/main.rs

@@ -0,0 +1,282 @@
+// TODO: These are temporary -
+#![allow(unused_imports)]
+#![allow(unused_mut)]
+#![allow(dead_code)]
+
+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_oauth2::tokens::UserTokenBuilder;
+use twitch_api2::helix::users::GetUsersRequest;
+use twitch_api2::types::Nickname;
+use crate::twitch::client::TwitchClientWrapper;
+
+pub mod twitch;
+pub mod util;
+
+const DEFAULT_LOG : &'static str = "debug,rustls=warn,tungstenite=warn";
+
+const TEST_CHANNEL_ID : u32 = 650154491; // Test channel
+
+fn main() -> AnyhowResult<()> {
+  easyenv::init_all_with_default_logging(Some(DEFAULT_LOG));
+
+  info!("Starting");
+
+  use std::error::Error;
+  if let Err(err) = run() {
+    println!("Error: {:?}", err);
+    //let mut e: &'_ dyn Error = err.as_ref();
+    //while let Some(cause) = e.source() {
+    //  println!("Caused by: {}", cause);
+    //  e = cause;
+    //}
+  }
+
+  Ok(())
+}
+
+#[tokio::main]
+async fn run() -> AnyhowResult<()> {
+  // ==================== APPLICATION ACCESS ====================
+
+  let secrets = TwitchSecrets::from_file("secrets.toml")?;
+  let client_id = ClientId::new(&secrets.app_client_id);
+  let client_secret = ClientSecret::new(&secrets.app_client_secret);
+
+  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 ====================
+
+  println!("Oauth flow...");
+
+  let redirect_url = twitch_oauth2::url::Url::parse("http://localhost/test")?;
+  let mut builder = UserTokenBuilder::new(client_id, client_secret, redirect_url)
+      .set_scopes(Scope::all())
+      .force_verify(true);
+
+  //builder.add_scope(Scope::BitsRead);
+
+  let (url, _csrf_token) = builder.generate_url();
+
+  println!("Go to this page: {}", url);
+
+  //std::thread::sleep(Duration::from_secs(3000));
+
+
+  // ==================== 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);
+
+  let input = rpassword::prompt_password_stdout(
+    "Paste in the resulting adress after authenticating (input hidden): ",
+  )?;
+
+
+  let u = twitch_oauth2::url::Url::parse(&input)?;
+
+  let map: std::collections::HashMap<_, _> = u.query_pairs().collect();
+
+  let user_token = match (map.get("state"), map.get("code")) {
+    (Some(state), Some(code)) => {
+      let token = builder
+          .get_user_token(
+            &reqwest::Client::builder()
+                .redirect(reqwest::redirect::Policy::none())
+                .build()?,
+            state,
+            code,
+          )
+          .await?;
+      println!("Got token: {:?}", token);
+
+      token
+    }
+    _ => match (map.get("error"), map.get("error_description")) {
+      (std::option::Option::Some(error), std::option::Option::Some(error_description)) => {
+        anyhow::bail!(
+                    "twitch errored with error: {} - {}",
+                    error,
+                    error_description
+                );
+      }
+      _ => anyhow::bail!("invalid url passed"),
+    },
+  };
+
+  let auth_token = user_token.access_token.as_str();
+
+  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 } } })
+   */
+
+
+
+  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);
+
+  Ok(())
+}
+
+
+async fn get_case_count() -> TungsteniteResult<u32> {
+  let (mut socket, _) = connect_async(
+    Url::parse("ws://localhost:9001/getCaseCount").expect("Can't connect to case count URL"),
+  )
+      .await?;
+  let msg = socket.next().await.expect("Can't fetch case count")?;
+  socket.close(None).await?;
+  Ok(msg.into_text()?.parse::<u32>().expect("Can't parse case count"))
+}
+
+
+/*let client_id = ClientId::new(&secrets.app_client_id);
+let client_secret = ClientSecret::new(&secrets.app_client_secret);
+
+let client: TwitchClient<reqwest::Client> = TwitchClient::default();
+let token = AppAccessToken::get_app_access_token(
+  &client,
+  client_id,
+  client_secret,
+  Scope::all(),
+).await?;
+
+let req = GetChannelInformationRequest::builder()
+    //.broadcaster_id("27620241")
+    .broadcaster_id("650154491")
+    .build();
+
+println!(
+  "{:?}",
+  &client.helix.req_get(req, &token).await?.data.unwrap().title
+);*/
+
+
+// Want this:
+// twitch_api2::pubsub::channel_bits::BitsEventData
+
+
+
+/*// We want to subscribe to moderator actions on channel with id 1234
+// as if we were a user with id 4321 that is moderator on the channel.
+let chat_mod_actions = pubsub::moderation::ChatModeratorActions {
+  user_id: 4321,
+  channel_id: 1234,
+}
+    .into_topic();*/
+
+// Listen to follows as well
+/*let follows = pubsub::following::Following { channel_id: 1234 }.into_topic();
+let topics = [follows];
+// Create the topic command to send to twitch
+let command = pubsub::listen_command(
+  &topics,
+  "authtoken",
+  "random nonce string",
+)?;
+
+println!("Pubsub command: {:?}", command);*/
+
+// Send the message with your favorite websocket client
+//send_command(command).unwrap();
+// To parse the websocket messages, use pubsub::Response::parse
+
+
+/*println!("Connected");
+let mut ws_client = TwitchWebsocketClient::new()?;
+ws_client.connect().await?;
+
+ws_client.send_ping().await?;
+
+println!("success ping");*/
+
+/*
+
+  // We want to subscribe to moderator actions on channel with id 1234
+  // as if we were a user with id 4321 that is moderator on the channel.
+  let chat_mod_actions = pubsub::moderation::ChatModeratorActions {
+    user_id: 4321,
+    channel_id: 1234,
+  }.into_topic();
+
+  // Listen to follows as well
+  let follows = pubsub::following::Following { channel_id: 1234 }.into_topic();
+ */
+
+//match UserToken::from_existing(reqwest_http_client, token, None, None).await {
+//  Ok(t) => println!("user_token: {}", t.token().secret()),
+//  Err(e) => panic!("got error: {}", e),
+//}
+

src/twitch/client.rs

@@ -0,0 +1,65 @@
+//! The Helix client is independent of PubSub
+//! This just wraps `twitch_api2` for more convenience.
+
+use crate::util::anyhow_result::AnyhowResult;
+use twitch_api2::helix::users::GetUsersRequest;
+use anyhow::anyhow;
+use twitch_api2::TwitchClient;
+use twitch_oauth2::{ClientSecret, ClientId, AccessToken, AppAccessToken, Scope};
+use twitch_api2::types::{Nickname, UserId};
+
+pub struct TwitchClientWrapper<'a> {
+  client_id: ClientId,
+  client_secret: ClientSecret,
+  twitch_client: TwitchClient<'a, reqwest::Client>,
+  access_token: Option<AppAccessToken>,
+}
+
+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<Scope>) -> AnyhowResult<()> {
+    self.get_access_token(scopes).await?;
+    Ok(())
+  }
+
+  async fn get_access_token(&mut self, scopes: Vec<Scope>) -> 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<u32> {
+    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::<u32>()?),
+    }
+  }
+}

src/twitch/mod.rs

@@ -0,0 +1,4 @@
+pub mod client;
+pub mod secrets;
+pub mod twitch_client_old;
+pub mod websocket_client;

src/twitch/secrets.rs

@@ -0,0 +1,43 @@
+use crate::util::anyhow_result::AnyhowResult;
+
+// TODO: Not sure I need all of these. The less I need to keep, the better.
+//  Oauth delegation in particular seems not necessary.
+#[derive(Clone, serde::Deserialize)]
+pub struct TwitchSecrets {
+  pub username: String,
+  pub stream_key: String,
+  pub app_client_id: String,
+  pub app_client_secret: String,
+}
+
+impl TwitchSecrets {
+  pub fn new(
+    username: &str,
+    stream_key: &str,
+    app_client_id: &str,
+    app_client_secret: &str,
+  ) -> Self {
+    Self {
+      username: username.to_string(),
+      stream_key: stream_key.to_string(),
+      app_client_id: app_client_id.to_string(),
+      app_client_secret: app_client_secret.to_string(),
+    }
+  }
+
+  pub fn from_env() -> AnyhowResult<Self> {
+    Ok(Self::new(
+      &easyenv::get_env_string_required("TWITCH_USERNAME")?,
+      &easyenv::get_env_string_required("TWITCH_STREAM_KEY")?,
+      &easyenv::get_env_string_required("TWITCH_APP_CLIENT_ID")?,
+      &easyenv::get_env_string_required("TWITCH_APP_CLIENT_SECRET")?,
+    ))
+  }
+
+  pub fn from_file(filename: &str) -> AnyhowResult<Self> {
+    let contents = std::fs::read_to_string(filename)?;
+    let secrets = toml::from_str(&contents)?;
+    Ok(secrets)
+  }
+}
+

src/twitch/twitch_client_old.rs

@@ -0,0 +1,172 @@
+/*use crate::twitch::secrets::TwitchSecrets;
+use crate::util::anyhow_result::AnyhowResult;
+use log::debug;
+use log::info;
+use log::warn;
+use std::time::Duration;
+use twitchchat::messages::Commands::*;
+use twitchchat::messages::Join;
+use twitchchat::messages::Privmsg;
+use twitchchat::{AsyncRunner, Status, UserConfig};
+
+pub struct TwitchClient2 {
+  secrets: TwitchSecrets,
+  channels_to_join: Vec<String>,
+
+  /// This manages our connection.
+  /// It's meant to be absent when not connected.
+  /// TODO: Probably need to reset this when we disconnect.
+  runner: Option<AsyncRunner>,
+}
+
+impl TwitchClient2 {
+
+  pub fn new(secrets: &TwitchSecrets) -> Self
+  {
+    Self {
+      secrets: secrets.clone(),
+      //channels_to_join: secrets.watch_channels.clone(), // TODO: Configure these elsewhere.
+      channels_to_join: vec![],
+      runner: None,
+    }
+  }
+
+  pub async fn connect(&mut self) -> AnyhowResult<()> {
+    let user_config = get_user_config(&self.secrets)?;
+    let connector = twitchchat::connector::tokio::Connector::twitch()?;
+
+    info!("Connecting to Twitch...");
+    let mut runner = AsyncRunner::connect(connector, &user_config).await?;
+
+    info!("Connected. Our Twitch identity: {:#?}", runner.identity);
+
+    for channel in self.channels_to_join.iter() {
+      info!("Attempting to join channel '{}'", channel);
+      let _ = runner.join(&channel).await?;
+
+      info!("Joined channel '{}'!", channel);
+    }
+
+    self.runner = Some(runner);
+
+    Ok(())
+  }
+
+  pub async fn main_loop(&mut self) {
+    loop {
+      let mut runner = match &mut self.runner {
+        Some(runner) => runner,
+        None => {
+          info!("Connecting to Twitch...");
+          match self.connect().await {
+            Ok(_) => {},
+            Err(e) => {
+              warn!("Failure to connect to Twitch: {:?}", e);
+              std::thread::sleep(Duration::from_secs(5));
+            },
+          }
+          continue;
+        },
+      };
+
+      let next_message = match runner.next_message().await {
+        Ok(msg) => msg,
+        Err(e) => {
+          warn!("Error getting message from Twitch: {:?}", e);
+          std::thread::sleep(Duration::from_secs(5));
+          continue;
+        },
+      };
+
+      match next_message {
+        Status::Message(message) => {
+          // NB: Handles across all channels (and notifications from Twitch)
+          match self.handle_message(message).await {
+            Ok(_) => {},
+            Err(e) => {
+              warn!("Error handling most recent message: {:?}", e);
+            }
+          }
+        },
+        Status::Quit => {
+          // we signaled a quit
+          warn!("Status: Quit.");
+        },
+        Status::Eof => {
+          // the connection closed normally
+          warn!("Status: EOF.");
+        },
+      }
+    }
+  }
+
+  async fn handle_message<'a>(
+    &mut self,
+    message: twitchchat::messages::Commands<'a>
+  ) -> AnyhowResult<()> {
+    match message {
+      // These happen when you initially connect
+      IrcReady(_) => {}
+      Ready(_) => {}
+      Cap(_) => {}
+
+      // This is the one users send to channels
+      Privmsg(msg) => self.handle_privmsg(&msg).await?,
+
+      // This is when a user joins the channel
+      // NB: These messages are delayed about 15-30 seconds.
+      // Doesn't appear to work...?
+      Join(msg) => info!("Join: {:?}", msg),
+
+      // This should happen on subscription, etc.
+      // Doesn't appear to work...?
+      UserNotice(msg) => info!("UserNotice: {:?}", msg),
+
+      // This should happen when a user joins, etc.
+      // Doesn't appear to work...?
+      // https://dev.twitch.tv/docs/irc/tags#userstate-twitch-tags
+      UserState(msg) => info!("UserState: {:?}", msg),
+
+      // and a bunch of other messages you may be interested in
+      ClearChat(msg) => debug!("ClearChat: {:?}", msg),
+      ClearMsg(msg) => debug!("ClearMsg: {:?}", msg),
+      GlobalUserState(msg) => debug!("GlobalUserState: {:?}", msg),
+      HostTarget(msg) => debug!("HostTarget: {:?}", msg),
+      Notice(msg) => debug!("Notice: {:?}", msg),
+      Part(msg) => debug!("Part: {:?}", msg),
+      Ping(_) => {}
+      Pong(_) => {}
+      Reconnect(_) => {}
+      RoomState(msg) => debug!("RoomState: {:?}", msg),
+      Whisper(msg) => debug!("Whisper: {:?}", msg),
+
+      // This one is special, if twitch adds any new message
+      // types, this will catch it until future releases of
+      // this crate add them.
+      Raw(msg) => debug!("Raw: {:?}", msg),
+
+      _ => {}
+    }
+
+    Ok(())
+  }
+
+  /// Handle user messages.
+  /// The "privmsg" type is for normal in-channel messages.
+  async fn handle_privmsg<'a>(&mut self, message: &Privmsg<'a>) -> AnyhowResult<()> {
+    info!("[{}] {}: {}", message.channel(), message.name(), message.data());
+    debug!("Full Message: {:?}", message);
+
+    Ok(())
+  }
+}
+
+pub fn get_user_config(twitch_secrets: &TwitchSecrets) -> AnyhowResult<UserConfig> {
+  let config = UserConfig::builder()
+      .name(twitch_secrets.username.to_string())
+      .token(twitch_secrets.oauth_access_token.to_string())
+      .enable_all_capabilities()
+      .build()?;
+  Ok(config)
+}
+*/

src/twitch/websocket_client.rs

@@ -0,0 +1,273 @@
+use anyhow::anyhow;
+use anyhow::bail;
+use crate::util::anyhow_result::AnyhowResult;
+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};
+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, Topic, Topics};
+use twitch_api2::pubsub;
+use crate::util::random_nonce::random_nonce;
+
+// 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\" }";
+const PING_SECONDS : u64 = 1 * 60; // Twitch calls for every 5 minutes, we'll do every 1 minute.
+
+pub struct PollingTwitchWebsocketClient {
+  client: Arc<Mutex<TwitchWebsocketClient>>,
+  is_polling: AtomicBool,
+}
+
+impl PollingTwitchWebsocketClient {
+  pub fn new() -> AnyhowResult<Self> {
+    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<Response> {
+    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<Option<Response>> {
+    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 { .. } => {}
+          }
+        }
+      }
+    }
+  }*/
+
+}
+
+pub struct TwitchWebsocketClient {
+  socket: Option<WebSocketStream<MaybeTlsStream<TcpStream>>>,
+}
+
+impl TwitchWebsocketClient {
+  pub fn new() -> AnyhowResult<Self> {
+    Ok(Self {
+      socket: None,
+    })
+  }
+
+  pub async fn connect(&mut self) -> AnyhowResult<()> {
+    let url = Url::parse(WEBSOCKET_GATEWAY)?;
+    let (mut socket, _response) =
+        connect_async(url).await?;
+    self.socket = Some(socket);
+    Ok(())
+  }
+
+  pub async fn close(&mut self) -> AnyhowResult<()> {
+    let mut socket = match self.socket {
+      None => return Ok(()), // Already closed.
+      Some(ref mut socket) => socket,
+    };
+
+    socket.close(None).await?;
+    self.socket = None;
+
+    Ok(())
+  }
+
+  pub async fn listen<'t>(&mut self, auth_token: &str, topics: &'t [Topics]) -> AnyhowResult<()> {
+    let mut socket = match self.socket {
+      None => return Err(anyhow!("not connected")),
+      Some(ref mut socket) => socket,
+    };
+
+    let nonce = random_nonce(128);
+
+    let command = pubsub::listen_command(
+      topics,
+      auth_token,
+      nonce.as_ref(),
+    )?;
+
+    debug!("Sending LISTEN");
+    let message = Message::Text(command);
+    socket.send(message).await?;
+
+    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,
+    };
+
+    debug!("Sending PING");
+    let message = Message::Text(PING_COMMAND.to_string());
+    socket.send(message).await?;
+
+    // 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(())
+  }
+
+  pub async fn next(&mut self) -> AnyhowResult<Response> {
+    let socket = match self.socket {
+      None => return Err(anyhow!("not connected")),
+      Some(ref mut s) => s,
+    };
+
+    match socket.next().await {
+      None => Err(anyhow!("nothing to read")),
+      Some(Err(e)) => Err(anyhow!("websocket error: {:?}", e)),
+      Some(Ok(response)) => {
+        let text = response.to_text()?;
+        let typed_response = Response::parse(text)?;
+        Ok(typed_response)
+      },
+    }
+  }
+
+  pub async fn try_next(&mut self) -> AnyhowResult<Option<Response>> {
+    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))
+      },
+    }
+  }
+}

src/util/anyhow_result.rs

@@ -0,0 +1,2 @@
+
+pub type AnyhowResult<T> = anyhow::Result<T>;

src/util/mod.rs

@@ -0,0 +1,3 @@
+pub mod anyhow_result;
+pub mod random_nonce;
+

src/util/random_nonce.rs

@@ -0,0 +1,24 @@
+use rand::distributions::Alphanumeric;
+use rand::{thread_rng, Rng};
+
+/// Returns a random nonce for Twitch
+pub fn random_nonce(length: usize) -> String {
+  thread_rng()
+      .sample_iter(&Alphanumeric)
+      .take(length)
+      .map(char::from)
+      .collect()
+}
+
+#[cfg(test)]
+mod tests {
+  use crate::util::random_nonce::random_nonce;
+
+  #[test]
+  fn random_nonce_length() {
+    assert_eq(1, random_nonce(1).len());
+    assert_eq(10, random_nonce(10).len());
+    assert_eq(15, random_nonce(15).len());
+    assert_eq(64, random_nonce(64).len());
+  }
+}