cleanup

crates/lib/redis_common/src/redis_keys.rs

@@ -47,8 +47,15 @@ impl RedisKeys {
 
   /// This is a PubSub topic.
   /// We publish when new viewers arrive and as a "keep-alive ping".
-  pub fn obs_active_session_topic() -> &'static str {
-    "obsActiveSessionTopic"
+  pub fn obs_active_sessions_topic() -> &'static str {
+    "obsActiveSessionsTopic"
+  }
+
+  /// This is a key that denotes a user is actively on the OBS gateway.
+  /// These keys have a short expiry and OBS gateway must continually bump them.
+  /// If a key goes away, the thread monitoring that user should exit.
+  pub fn obs_active_session(twitch_user_id: &str) -> String {
+    format!("obsActiveSession:{}", twitch_user_id)
   }
 
   /// This is a key that denotes a PubSub lease subscriber.
@@ -61,7 +68,7 @@ impl RedisKeys {
   /// This is a PubSub topic.
   /// These are "unenriched" Twitch events from the PubSub (and eventually IRC)
   /// subscribers. Downstream listeners will enrich these for user-facing functionality.
-  pub fn twitchEventTopic(twitch_user_id: &str) -> String {
-    format!("twitchEventTopic:{}", twitch_user_id)
+  pub fn twitch_events_topic(twitch_user_id: &str) -> String {
+    format!("twitchEventsTopic:{}", twitch_user_id)
   }
 }

crates/service/obs_gateway/src/endpoints_ws/obs_gateway_websocket_handler.rs

@@ -77,7 +77,7 @@ pub async fn obs_gateway_websocket_handler(
         CommonServerError::ServerError
       })?;
 
-  let channel = RedisKeys::obs_active_session_topic();
+  let channel = RedisKeys::obs_active_sessions_topic();
 
   let payload = ObsActivePayload::new(&token_record.twitch_user_id);
   let json_payload = payload.serialize()
@@ -92,36 +92,12 @@ pub async fn obs_gateway_websocket_handler(
         CommonServerError::ServerError
       })?;
 
-  let mut client = TwitchWebsocketClient::new().unwrap();
-
-  info!("Connecting to Twitch PubSub...");
-  client.connect().await.unwrap();
-
-//  info!("Connected to Twitch PubSub");
-//
-//  info!("Sending Twitch PubSub PING...");
-//  client.send_ping().await.unwrap();
-//
-//  info!("Try read next from Twitch PubSub...");
-//  let r = client.try_next().await.unwrap();
-//  info!("Twitch PubSub Result: {:?}", r);
-//
-//  info!("Begin TwitchPubSub LISTEN on authenticated OAuth topics...");
-//  let topics = build_pubsub_topics_for_user(user_id);
-//  client.listen(&auth_token, &topics).await.unwrap();
-
-  //let (tx, rx) = crossbeam::channel::bounded(1);
-  //let client = Arc::new(&self.twitch_client);
-  //let client2 = client.clone();
-  //let res = rx.recv().unwrap();
-
   info!("Begin Javascript WebSocket...");
 
   let server_state_arc = server_state.get_ref().clone();
 
   let websocket = ObsGatewayWebSocket::new(
     twitch_user_id.clone(),
-    client,
     server_state_arc
   );
 
@@ -135,22 +111,16 @@ pub async fn obs_gateway_websocket_handler(
 /// Websocket behavior
 struct ObsGatewayWebSocket {
   twitch_user_id: TwitchUserId,
-  twitch_thread: Arc<ObsTwitchThread>,
   server_state: Arc<ObsGatewayServerState>,
 }
 
 impl ObsGatewayWebSocket {
   fn new(
     twitch_user_id: TwitchUserId,
-    twitch_client: TwitchWebsocketClient,
     server_state: Arc<ObsGatewayServerState>,
   ) -> Self {
-    let twitch_thread = Arc::new(ObsTwitchThread::new(
-      twitch_user_id.clone(),
-      twitch_client));
     Self {
       twitch_user_id,
-      twitch_thread,
       server_state,
     }
   }
@@ -160,29 +130,6 @@ impl Actor for ObsGatewayWebSocket {
   type Context = ws::WebsocketContext<Self>;
 
   fn started(&mut self, _ctx: &mut Self::Context) {
-    let handle = Handle::current();
-    let twitch_thread = self.twitch_thread.clone();
-
-
-    //let future = self.twitch_thread.run_until_exit();
-    ////let now_future = Delay::new(Duration::from_secs(5));
-    //warn!("Starting thread...");
-    ////actix_rt::spawn
-    ////actix_rt::spawn(twitch_thread.run_until_exit());
-
-    //actix_rt::spawn(future.map(|x| {
-    //  println!("waited for 5 secs");
-    //}));
-
-    //handle.spawn_blocking(result);
-//    handle.spawn_blocking(async {
-//      warn!("inside thread 1");
-//      let twitch_thread2 = twitch_thread.clone();
-//      //async move {
-//      //  warn!("inside thread 2");
-//      //  twitch_thread2.run_until_exit().await;
-//      //}.await;
-//    });
     warn!("Thread started");
   }
 }
@@ -194,29 +141,6 @@ impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for ObsGatewayWebSock
     ctx: &mut Self::Context,
   ) {
     if let Ok(msg) = msg {
-      //info!(">>>>>> obs streamhandler::handle (msg = {:?})", msg);
-
-      //let (tx, rx) = crossbeam::channel::bounded(1);
-      //let handle = Handle::current();
-
-      //let client = Arc::new(&self.twitch_client);
-      //let client2 = client.clone();
-
-      //handle.spawn(async {
-      //  error!("Twitch PubSub Try read next...");
-      //  match client2.try_next().await {
-      //    Ok(r) => {
-      //      error!("Twitch PubSub Result: {:?}", r);
-      //    },
-      //    Err(e) => {
-      //      warn!("pubsub error: {:?}", e);
-      //    }
-      //  }
-      //});
-
-      //let res = rx.recv().unwrap();
-
-      //info!("process message: {:?}", &msg);
 
       // TODO: Only send this every 60 seconds.
       // TODO: Error handling.
@@ -230,7 +154,7 @@ impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for ObsGatewayWebSock
             CommonServerError::ServerError
           }).unwrap(); // TODO: FIXME
 
-      let channel = RedisKeys::obs_active_session_topic();
+      let channel = RedisKeys::obs_active_sessions_topic();
 
       let payload = ObsActivePayload::new(self.twitch_user_id.get_str());
 
@@ -266,38 +190,3 @@ impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for ObsGatewayWebSock
     }
   }
 }
-
-// From the Actix homepage:
-
-//use actix::{Actor, StreamHandler};
-//use actix_web::{web, App, Error, HttpRequest, HttpResponse, HttpServer};
-//use actix_web_actors::ws;
-//
-///// Define HTTP actor
-//struct MyWs;
-//
-//impl Actor for MyWs {
-//  type Context = ws::WebsocketContext<Self>;
-//}
-//
-///// Handler for ws::Message message
-//impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for MyWs {
-//  fn handle(
-//    &mut self,
-//    msg: Result<ws::Message, ws::ProtocolError>,
-//    ctx: &mut Self::Context,
-//  ) {
-//    match msg {
-//      Ok(ws::Message::Ping(msg)) => ctx.pong(&msg),
-//      Ok(ws::Message::Text(text)) => ctx.text(text),
-//      Ok(ws::Message::Binary(bin)) => ctx.binary(bin),
-//      _ => (),
-//    }
-//  }
-//}
-//
-//pub async fn twitch_pubsub_gateway(req: HttpRequest, stream: web::Payload) -> Result<HttpResponse, Error> {
-//  let resp = ws::start(MyWs {}, &req, stream);
-//  println!("{:?}", resp);
-//  resp
-//}

crates/service/obs_gateway/src/redis/constants.rs

@@ -13,3 +13,7 @@ pub const LEASE_RENEW_PERIOD : Duration = Duration::seconds(30);
 
 /// The amount of time to wait before checking or lease is valid.
 pub const LEASE_CHECK_PERIOD : Duration = Duration::seconds(15);
+
+/// The OBS websocket polls a key to inform the backend that the
+/// user is still active.
+pub const OBS_ACTIVE_TTL_SECONDS : usize = 120;

crates/service/obs_gateway/src/redis/mod.rs

@@ -1,3 +1,3 @@
 pub mod lease_payload;
-pub mod lease_timeout;
+pub mod constants;
 pub mod obs_active_payload;

crates/service/obs_gateway/src/threads/listen_for_active_obs_sessions_thread.rs

@@ -1,5 +1,6 @@
+use container_common::thread::thread_id::ThreadId;
+use crate::redis::constants::LEASE_TIMEOUT_SECONDS;
 use crate::redis::lease_payload::LeasePayload;
-use crate::redis::lease_timeout::LEASE_TIMEOUT_SECONDS;
 use crate::redis::obs_active_payload::ObsActivePayload;
 use crate::threads::twitch_pubsub_user_subscriber_thread::TwitchPubsubUserSubscriberThread;
 use crate::twitch::twitch_user_id::TwitchUserId;
@@ -15,7 +16,6 @@ use std::sync::Arc;
 use std::thread::sleep;
 use std::time::Duration;
 use tokio::runtime::Runtime;
-use container_common::thread::thread_id::ThreadId;
 
 // TODO: Apart from error handling, this looks mostly good.
 
@@ -55,7 +55,7 @@ impl ListenForActiveObsSessionThread {
     let mut pubsub_pool = self.redis_pubsub_pool.get().unwrap();
     let mut pubsub = pubsub_pool.as_pubsub();
 
-    pubsub.subscribe(RedisKeys::obs_active_session_topic()).unwrap();
+    pubsub.subscribe(RedisKeys::obs_active_sessions_topic()).unwrap();
 
     loop {
       let message = pubsub.get_message().unwrap();

crates/service/obs_gateway/src/threads/twitch_pubsub_user_subscriber_thread.rs

@@ -1,7 +1,7 @@
 use container_common::anyhow_result::AnyhowResult;
 use container_common::thread::thread_id::ThreadId;
 use crate::redis::lease_payload::LeasePayload;
-use crate::redis::lease_timeout::{LEASE_TIMEOUT_SECONDS, LEASE_RENEW_PERIOD, LEASE_CHECK_PERIOD};
+use crate::redis::constants::{LEASE_TIMEOUT_SECONDS, LEASE_RENEW_PERIOD, LEASE_CHECK_PERIOD};
 use crate::twitch::constants::TWITCH_PING_CADENCE;
 use crate::twitch::pubsub::build_pubsub_topics_for_user::build_pubsub_topics_for_user;
 use crate::twitch::twitch_user_id::TwitchUserId;