Migrate/actix ws (#907)

* Began migrating to actix ws.

* Fixed cargo clippy

---------

Co-authored-by: SamTV12345 <noreply+samtv1235@github.com>

Author: SamTV12345

Cargo.lock

@@ -32,14 +32,14 @@ actix-files = "0.6.6"
 actix-web = {version="4.9.0", features=["rustls"]}
 jsonwebtoken = {version="9.3.0"}
 log = "0.4.22"
-futures-util = "0.3.29"
+futures-util = "0.3.30"
 substring = "1.4.5"
 opml = "1.1.6"
 rand = "0.8.5"
 env_logger = "0.11.5"
 chrono = {version  = "0.4.38", default-features=false, features = ["serde"]}
-actix-web-actors = "4.3.1"
-rss = "2.0.8"
+rss = "2.0.9"
+actix-ws = "0.3.0"
 frankenstein = "0.32.4"
 regex = "1.10.6"
 xml-builder = "0.5.2"
@@ -58,7 +58,7 @@ serde_json = "1.0.127"
 dotenv = "0.15.0"
 thiserror = "1.0.63"
 sha1 = "0.10.6"
-sha256 = "1.4.0"
+sha256 = "1.5.0"
 strfmt="0.2.4"
 urlencoding="2.1.3"
 id3 = "1.14.0"

Cargo.toml

@@ -142,3 +142,6 @@ pub static ITUNES: &str = "itunes";
 pub const REVERSE_PROXY: &str = "REVERSE_PROXY";
 pub const REVERSE_PROXY_HEADER: &str = "REVERSE_PROXY_HEADER";
 pub const REVERSE_PROXY_AUTO_SIGN_UP: &str = "REVERSE_PROXY_AUTO_SIGN_UP";
+
+
+pub const MAIN_ROOM: &str = "main";

src/constants/inner_constants.rs

@@ -10,3 +10,4 @@ pub mod user_controller;
 pub mod watch_time_controller;
 pub mod web_socket;
 pub mod websocket_controller;
+pub mod server;

src/controllers/mod.rs

@@ -1,17 +1,13 @@
-use crate::constants::inner_constants::{
-    PodcastType, BASIC_AUTH, COMMON_USER_AGENT, DEFAULT_IMAGE_URL, ENVIRONMENT_SERVICE, OIDC_AUTH,
-};
+use crate::constants::inner_constants::{PodcastType, BASIC_AUTH, COMMON_USER_AGENT, DEFAULT_IMAGE_URL, ENVIRONMENT_SERVICE, MAIN_ROOM, OIDC_AUTH};
 use crate::models::dto_models::PodcastFavorUpdateModel;
 use crate::models::misc_models::{PodcastAddModel, PodcastInsertModel};
 use crate::models::opml_model::OpmlModel;
 use crate::models::search_type::SearchType::{ITunes, Podindex};
-use crate::models::web_socket_message::Lobby;
 use crate::service::environment_service::EnvironmentService;
 use crate::service::mapping_service::MappingService;
 use crate::service::podcast_episode_service::PodcastEpisodeService;
 use crate::service::rust_service::PodcastService;
 use crate::{get_default_image, unwrap_string, DbPool};
-use actix::Addr;
 use actix_web::dev::PeerAddr;
 use actix_web::http::Method;
 use actix_web::web::{Data, Json, Path};
@@ -239,7 +235,7 @@ tag="podcasts"
 #[post("/podcast/itunes")]
 pub async fn add_podcast(
     track_id: web::Json<PodcastAddModel>,
-    lobby: Data<Addr<Lobby>>,
+    lobby: Data<ChatServerHandle>,
     conn: Data<DbPool>,
     requester: Option<web::ReqData<User>>,
 ) -> Result<HttpResponse, CustomError> {
@@ -290,7 +286,7 @@ tag="podcasts"
 #[post("/podcast/feed")]
 pub async fn add_podcast_by_feed(
     rss_feed: web::Json<PodcastRSSAddModel>,
-    lobby: Data<Addr<Lobby>>,
+    lobby: Data<ChatServerHandle>,
     podcast_service: Data<Mutex<PodcastService>>,
     conn: Data<DbPool>,
     requester: Option<web::ReqData<User>>,
@@ -349,7 +345,7 @@ tag="podcasts"
 #[post("/podcast/opml")]
 pub async fn import_podcasts_from_opml(
     opml: web::Json<OpmlModel>,
-    lobby: Data<Addr<Lobby>>,
+    lobby: Data<ChatServerHandle>,
     conn: Data<DbPool>,
     requester: Option<web::ReqData<User>>,
 ) -> Result<HttpResponse, CustomError> {
@@ -388,7 +384,7 @@ tag="podcasts"
 #[post("/podcast/podindex")]
 pub async fn add_podcast_from_podindex(
     id: web::Json<PodcastAddModel>,
-    lobby: Data<Addr<Lobby>>,
+    lobby: Data<ChatServerHandle>,
     conn: Data<DbPool>,
     requester: Option<web::ReqData<User>>,
 ) -> Result<HttpResponse, CustomError> {
@@ -424,7 +420,7 @@ pub async fn add_podcast_from_podindex(
 
 fn start_download_podindex(
     id: i32,
-    lobby: Data<Addr<Lobby>>,
+    lobby: Data<ChatServerHandle>,
     conn: &mut DbConnection,
 ) -> Result<Podcast, CustomError> {
     let rt = tokio::runtime::Runtime::new().unwrap();
@@ -464,7 +460,7 @@ tag="podcasts"
 )]
 #[post("/podcast/all")]
 pub async fn refresh_all_podcasts(
-    lobby: Data<Addr<Lobby>>,
+    lobby: Data<ChatServerHandle>,
     podcast_service: Data<Mutex<PodcastService>>,
     conn: Data<DbPool>,
     requester: Option<web::ReqData<User>>,
@@ -485,13 +481,13 @@ pub async fn refresh_all_podcasts(
                     conn.get().map_err(map_r2d2_error).unwrap().deref_mut(),
                 )
                 .unwrap();
-            lobby.clone().do_send(BroadcastMessage {
+            lobby.send_broadcast_sync(MAIN_ROOM.parse().unwrap(), serde_json::to_string(&BroadcastMessage {
                 podcast_episode: None,
                 type_of: PodcastType::RefreshPodcast,
                 message: format!("Refreshed podcast: {}", podcast.name),
                 podcast: Option::from(podcast.clone()),
                 podcast_episodes: None,
-            });
+            }).unwrap());
         }
     });
     Ok(HttpResponse::Ok().into())
@@ -506,7 +502,7 @@ tag="podcasts"
 #[post("/podcast/{id}/refresh")]
 pub async fn download_podcast(
     id: Path<String>,
-    lobby: Data<Addr<Lobby>>,
+    lobby: Data<ChatServerHandle>,
     podcast_service: Data<Mutex<PodcastService>>,
     conn: Data<DbPool>,
     requester: Option<web::ReqData<User>>,
@@ -621,7 +617,7 @@ pub async fn update_active_podcast(
 async fn insert_outline(
     podcast: Outline,
     client: Client,
-    lobby: Data<Addr<Lobby>>,
+    lobby: Data<ChatServerHandle>,
     mut rng: ThreadRng,
     environment: EnvironmentService,
     conn: Data<DbPool>,
@@ -647,13 +643,13 @@ async fn insert_outline(
 
     let feed_response = client.get(feed_url.unwrap()).send().await;
     if feed_response.is_err() {
-        lobby.do_send(BroadcastMessage {
+        lobby.send_broadcast(MAIN_ROOM.parse().unwrap(),serde_json::to_string(&BroadcastMessage {
             type_of: PodcastType::OpmlErrored,
             message: feed_response.err().unwrap().to_string(),
             podcast: None,
             podcast_episodes: None,
             podcast_episode: None,
-        });
+        }).unwrap()).await;
         return;
     }
     let content = feed_response.unwrap().bytes().await.unwrap();
@@ -689,35 +685,43 @@ async fn insert_outline(
                 )
                 .await;
             match inserted_podcast {
-                Ok(podcast) => lobby.do_send(BroadcastMessage {
-                    type_of: PodcastType::OpmlAdded,
-                    message: "Refreshed podcasts".to_string(),
-                    podcast: Option::from(podcast),
-                    podcast_episodes: None,
-                    podcast_episode: None,
-                }),
-                Err(e) => lobby.do_send(BroadcastMessage {
-                    type_of: PodcastType::OpmlErrored,
-                    message: e.to_string(),
-                    podcast: None,
-                    podcast_episodes: None,
-                    podcast_episode: None,
-                }),
+                Ok(podcast) => {
+
+                    let _ = lobby.send_broadcast(MAIN_ROOM.parse().unwrap(), serde_json::to_string(&BroadcastMessage {
+                        type_of: PodcastType::OpmlAdded,
+                        message: "Refreshed podcasts".to_string(),
+                        podcast: Option::from(podcast),
+                        podcast_episodes: None,
+                        podcast_episode: None,
+                    }).unwrap()).await;
+                },
+                Err(e) => {
+                    let _ = lobby.send_broadcast(MAIN_ROOM.parse().unwrap(), serde_json::to_string(&BroadcastMessage {
+                        type_of: PodcastType::OpmlErrored,
+                        message: e.to_string(),
+                        podcast: None,
+                        podcast_episodes: None,
+                        podcast_episode: None,
+                    }).unwrap()).await;
+                },
             }
         }
-        Err(e) => lobby.do_send(BroadcastMessage {
-            type_of: PodcastType::OpmlErrored,
-            message: e.to_string(),
-            podcast: None,
-            podcast_episodes: None,
-            podcast_episode: None,
-        }),
+        Err(e) => {
+            let _ = lobby.send_broadcast(MAIN_ROOM.parse().unwrap(),serde_json::to_string(&BroadcastMessage {
+                type_of: PodcastType::OpmlErrored,
+                message: e.to_string(),
+                podcast: None,
+                podcast_episodes: None,
+                podcast_episode: None,
+            }).unwrap()).await;
+        },
     }
 }
 use crate::models::episode::Episode;
 use utoipa::ToSchema;
 
 use crate::controllers::podcast_episode_controller::EpisodeFormatDto;
+use crate::controllers::server::ChatServerHandle;
 use crate::controllers::websocket_controller::RSSAPiKey;
 use crate::models::podcast_settings::PodcastSetting;
 use crate::models::settings::Setting;

src/controllers/podcast_controller.rs

@@ -1,18 +1,16 @@
-use crate::constants::inner_constants::PodcastType;
+use crate::constants::inner_constants::{PodcastType, MAIN_ROOM};
 use crate::db::TimelineItem;
 use crate::models::episode::Episode;
 use crate::models::favorites::Favorite;
 use crate::models::messages::BroadcastMessage;
 use crate::models::podcast_episode::PodcastEpisode;
 use crate::models::podcasts::Podcast;
 use crate::models::user::User;
-use crate::models::web_socket_message::Lobby;
 
 use crate::service::mapping_service::MappingService;
 use crate::service::podcast_episode_service::PodcastEpisodeService;
 use crate::utils::error::{map_r2d2_error, CustomError};
 use crate::DbPool;
-use actix::Addr;
 use actix_web::web::{Data, Json, Query};
 use actix_web::{delete, get, post, put};
 use actix_web::{web, HttpResponse};
@@ -22,6 +20,7 @@ use std::ops::DerefMut;
 use crate::models::settings::Setting;
 use crate::service::file_service::perform_episode_variable_replacement;
 use std::thread;
+use crate::controllers::server::ChatServerHandle;
 
 #[derive(Debug, Serialize, Deserialize, Clone)]
 pub struct OptionalId {
@@ -214,7 +213,7 @@ pub async fn delete_podcast_episode_locally(
     id: web::Path<String>,
     requester: Option<web::ReqData<User>>,
     db: Data<DbPool>,
-    lobby: Data<Addr<Lobby>>,
+    lobby: Data<ChatServerHandle>,
 ) -> Result<HttpResponse, CustomError> {
     if !requester.unwrap().is_privileged_user() {
         return Err(CustomError::Forbidden);
@@ -224,13 +223,13 @@ pub async fn delete_podcast_episode_locally(
         &id.into_inner(),
         &mut db.get().unwrap(),
     )?;
-    lobby.do_send(BroadcastMessage {
+    lobby.send_broadcast(MAIN_ROOM.parse().unwrap(),serde_json::to_string(&BroadcastMessage {
         podcast_episode: Some(delted_podcast_episode),
         podcast_episodes: None,
         type_of: PodcastType::DeletePodcastEpisode,
         podcast: None,
         message: "Deleted podcast episode locally".to_string(),
-    });
+    }).unwrap()).await;
 
     Ok(HttpResponse::NoContent().finish())
 }