SSE

Author: neckaros

Cargo.lock

@@ -53,6 +53,7 @@ strum_macros = "0.26.1"
 tokio-util = { version = "0.7.11", features = ["compat", "io", "io-util"] }
 time = "0.3.34"
 tokio-stream = { version = "0.1.14", features = ["io-util"] }
+async-stream = "0.3"
 socketioxide = { version = "0.13.1", features = ["extensions","state", "v4"] }
 async-trait = "0.1.77"
 mime_guess = "2.0.4"

Cargo.toml

@@ -172,6 +172,7 @@ async fn app() -> Result<Router> {
         .nest("/credentials", routes::credentials::routes(mc.clone()))
         .nest("/backups", routes::backups::routes(mc.clone()))
         .nest("/plugins", routes::plugins::routes(mc.clone()))
+        .nest("/sse", routes::sse::routes(mc.clone()))
         .fallback(fallback)
         .layer(middleware::from_fn(mw_range::mw_range))
         //.layer(middleware::map_response(main_response_mapper))

src/main.rs

@@ -19,6 +19,7 @@ use tokio_util::io::{ReaderStream, StreamReader};
 use crate::{domain::{backup::{self, Backup, BackupError, BackupFile, BackupFileProgress, BackupMessage, BackupProcessStatus, BackupStatus, BackupWithStatus}, library::{LibraryRole, ServerLibrary}, media::{self, Media, MediaForUpdate, DEFAULT_MIME}, progress::{RsProgress, RsProgressType}}, error::{RsError, RsResult}, model::libraries::ServerLibraryForAdd, plugins::sources::{async_reader_progress::ProgressReader, error::SourcesError, AsyncReadPinBox, FileStreamResult, SourceRead}, routes::mw_range::RangeDefinition, tools::{clock::now, encryption::{ceil_to_multiple_of_16, derive_key, estimated_encrypted_size, random_iv, AesTokioDecryptStream, AesTokioEncryptStream}, log::{log_error, log_info}}};
 
 use super::{error::{Error, Result}, medias::{MediaFileQuery, MediaQuery, MediaSource}, store::sql::backups::BackupInfos, users::{ConnectedUser, UserRole}, ModelController};
+use crate::routes::sse::SseEvent;
 
 #[derive(Debug, Serialize, Deserialize, Clone)]
 pub struct BackupForAdd {
@@ -54,6 +55,7 @@ pub struct BackupForUpdate {
 impl ModelController {
 
     pub fn send_backup_status(&self, message: BackupMessage) {
+		self.broadcast_sse(SseEvent::Backups(message.clone()));
 		self.for_connected_users(&message, |user, socket, message| {
             if let Some(library) = &message.backup.backup.library {
                 let r = user.check_library_role(library, LibraryRole::Admin);
@@ -69,6 +71,7 @@ impl ModelController {
 		});
 	}
     pub fn send_backup_file_status(&self, message: BackupFileProgress) {
+		self.broadcast_sse(SseEvent::BackupsFiles(message.clone()));
 		self.for_connected_users(&message, |user, socket, message| {
             let r = user.check_role(&UserRole::Admin);
 			if r.is_ok() {

src/model/backups.rs

@@ -17,6 +17,7 @@ use tokio_util::io::StreamReader;
 use crate::{domain::{deleted::RsDeleted, episode::{self, Episode, EpisodeWithAction, EpisodeWithShow, EpisodesMessage}, library::LibraryRole, people::{PeopleMessage, Person}, serie::{self, Serie, SeriesMessage}, ElementAction}, error::RsResult, plugins::{medias::imdb::ImdbContext, sources::{error::SourcesError, AsyncReadPinBox, FileStreamResult, Source}}, tools::{array_tools::Dedup, clock::now, image_tools::{resize_image_reader, ImageSize}, log::log_info}};
 
 use super::{error::{Error, Result}, medias::{RsSort, RsSortOrder}, store::sql::SqlOrder, users::{ConnectedUser, HistoryQuery}, ModelController};
+use crate::routes::sse::SseEvent;
 
 
 #[derive(Debug, Serialize, Deserialize, Clone, Default)]
@@ -203,6 +204,7 @@ impl ModelController {
 
 
 	pub fn send_episode(&self, message: EpisodesMessage) {
+		self.broadcast_sse(SseEvent::Episodes(message.clone()));
 		self.for_connected_users(&message, |user, socket, message| {
             let r = user.check_library_role(&message.library, LibraryRole::Read);
 			if r.is_ok() {

src/model/episodes.rs

@@ -27,6 +27,7 @@ use crate::{domain::{MediaElement, deleted::RsDeleted, library::LibraryType, med
 use crate::{domain::{library::LibraryRole, media::{FileType, GroupMediaDownload, Media, MediaDownloadUrl, MediaForAdd, MediaForInsert, MediaForUpdate, MediaItemReference, MediaWithAction, MediasMessage, ProgressMessage}, progress::{RsProgress, RsProgressType}, ElementAction}, error::RsResult, plugins::{get_plugin_fodler, sources::{async_reader_progress::ProgressReader, error::SourcesError, AsyncReadPinBox, FileStreamResult, SourceRead}}, routes::mw_range::RangeDefinition, server::get_server_port, tools::{auth::{sign_local, ClaimsLocal}, file_tools::{file_type_from_mime, get_extension_from_mime}, image_tools::{self, resize_image_reader, ImageSize}, log::{log_error, log_warn, log_info, LogServiceType}, prediction::{predict_net, preload_model, PredictionTagResult}, video_tools::{self, probe_video, VideoTime}}};
 
 use super::{error::{Error, Result}, plugins::PluginQuery, store::{self, sql::library::medias::MediaBackup}, users::ConnectedUser, ModelController, VideoConvertQueueElement};
+use crate::routes::sse::SseEvent;
 
 pub const CRYPTO_HEADER_SIZE: u64 = 16 + 4 + 4 + 32 + 256;
 
@@ -258,6 +259,7 @@ impl ModelController {
 
 
 	pub fn send_media(&self, message: MediasMessage) {
+		self.broadcast_sse(SseEvent::Medias(message.clone()));
 		self.for_connected_users(&message, |user, socket, message| {
             let r = user.check_library_role(&message.library, LibraryRole::Read);
 			if r.is_ok() {
@@ -267,6 +269,7 @@ impl ModelController {
 	}
 
 	pub fn send_progress(&self, message: ProgressMessage) {
+		self.broadcast_sse(SseEvent::MediasProgress(message.clone()));
 		self.for_connected_users(&message, |user, socket, message| {
             let r = user.check_library_role(&message.library, LibraryRole::Read);
 			if r.is_ok() {
@@ -276,6 +279,7 @@ impl ModelController {
 	}
 
     pub fn send_convert_progress(&self, message: ConvertMessage) {
+		self.broadcast_sse(SseEvent::ConvertProgress(message.clone()));
 		self.for_connected_users(&message, |user, socket, message| {
             let r = user.check_library_role(&message.library, LibraryRole::Read);
 			if r.is_ok() {

src/model/medias.rs

@@ -73,9 +73,11 @@ use socketioxide::{extract::SocketRef, SocketIo};
 use tokio::{
     fs::{self, remove_file, File},
     io::{copy, AsyncRead, BufReader},
-    sync::RwLock,
+    sync::{broadcast, RwLock},
 };
 
+use crate::routes::sse::SseEvent;
+
 #[derive(Debug, Serialize, Deserialize, Clone)]
 pub struct VideoConvertQueueElement {
     request: VideoConvertRequest,
@@ -137,6 +139,9 @@ pub struct ModelController {
     pub backup_processes: Arc<RwLock<Vec<BackupProcessStatus>>>,
 
     pub chache_libraries: Arc<RwLock<HashMap<String, ServerLibrary>>>,
+
+    /// Broadcast channel for SSE events
+    pub sse_tx: broadcast::Sender<SseEvent>,
 }
 
 // Constructor
@@ -145,6 +150,7 @@ impl ModelController {
         let tmdb = TmdbContext::new("4a01db3a73eed5cf17e9c7c27fd9d008".to_string()).await?;
         let fanart = FanArtContext::new("a6eb2f1acb7b54550e498a9b37a574fa".to_string());
         let scheduler = RsScheduler::new();
+        let (sse_tx, _) = broadcast::channel::<SseEvent>(1024);
 
         let mc = Self {
             store: Arc::new(store),
@@ -164,6 +170,7 @@ impl ModelController {
             convert_current_process: Arc::new(RwLock::new(None)),
 
             backup_processes: Arc::new(RwLock::new(vec![])),
+            sse_tx,
         };
 
         let pm_forload = mc.plugin_manager.clone();
@@ -550,7 +557,13 @@ impl ModelController {
         }
     }
 
+    /// Broadcasts an event to all SSE subscribers
+    pub fn broadcast_sse(&self, event: SseEvent) {
+        let _ = self.sse_tx.send(event);
+    }
+
     pub fn send_library(&self, message: LibraryMessage) {
+        self.broadcast_sse(SseEvent::Library(message.clone()));
         self.for_connected_users(&message, |user, socket, message| {
             if let Some(message) = message.for_socket(user) {
                 let _ = socket.emit("library", message);
@@ -559,6 +572,7 @@ impl ModelController {
     }
 
     pub fn send_library_status(&self, message: LibraryStatusMessage) {
+        self.broadcast_sse(SseEvent::LibraryStatus(message.clone()));
         self.for_connected_users(&message, |user, socket, message| {
             if user
                 .check_library_role(&message.library, LibraryRole::Admin)

src/model/mod.rs

@@ -17,6 +17,7 @@ use tokio_util::io::StreamReader;
 use crate::{domain::{deleted::RsDeleted, library::LibraryRole, movie::{Movie, MovieForUpdate, MovieWithAction, MoviesMessage}, people::{PeopleMessage, Person}, ElementAction, MediaElement}, error::RsResult, plugins::{medias::imdb::ImdbContext, sources::{error::SourcesError, path_provider::PathProvider, AsyncReadPinBox, FileStreamResult, Source}}, server::get_server_folder_path_array, tools::{image_tools::{convert_image_reader, resize_image_reader, ImageSize}, log::log_info}};
 
 use super::{error::{Error, Result}, store::sql::SqlOrder, users::{ConnectedUser, HistoryQuery}, ModelController};
+use crate::routes::sse::SseEvent;
 
 
 
@@ -237,6 +238,7 @@ impl ModelController {
 
 
 	pub fn send_movie(&self, message: MoviesMessage) {
+		self.broadcast_sse(SseEvent::Movies(message.clone()));
 		self.for_connected_users(&message, |user, socket, message| {
             let r = user.check_library_role(&message.library, LibraryRole::Read);
 			if r.is_ok() {

src/model/movies.rs

@@ -48,6 +48,7 @@ use super::{
     users::ConnectedUser,
     ModelController,
 };
+use crate::routes::sse::SseEvent;
 
 // Default face recognition threshold when library doesn't specify one
 pub const DEFAULT_FACE_THRESHOLD: f32 = 0.4;
@@ -273,6 +274,7 @@ impl ModelController {
     }
 
     pub fn send_people(&self, message: PeopleMessage) {
+        self.broadcast_sse(SseEvent::People(message.clone()));
         self.for_connected_users(&message, |user, socket, message| {
             let r = user.check_library_role(&message.library, LibraryRole::Read);
             if r.is_ok() {

src/model/people.rs

@@ -17,6 +17,7 @@ use tokio_util::io::StreamReader;
 use crate::{domain::{deleted::RsDeleted, library::LibraryRole, people::{PeopleMessage, Person}, serie::{Serie, SerieStatus, SerieWithAction, SeriesMessage}, ElementAction, MediaElement}, error::RsResult, plugins::{medias::imdb::ImdbContext, sources::{error::SourcesError, path_provider::PathProvider, AsyncReadPinBox, FileStreamResult, Source}}, server::get_server_folder_path_array, tools::{image_tools::{convert_image_reader, resize_image_reader, ImageSize}, log::log_info}};
 
 use super::{episodes::{EpisodeForUpdate, EpisodeQuery}, error::{Error, Result}, medias::{MediaQuery, RsSort}, store::sql::SqlOrder, users::ConnectedUser, ModelController};
+use crate::routes::sse::SseEvent;
 
 
 impl FromSql for SerieStatus {
@@ -206,6 +207,7 @@ impl ModelController {
 
 
 	pub fn send_serie(&self, message: SeriesMessage) {
+		self.broadcast_sse(SseEvent::Series(message.clone()));
 		self.for_connected_users(&message, |user, socket, message| {
             let r = user.check_library_role(&message.library, LibraryRole::Read);
 			if r.is_ok() {