From f463b5bc81d7d04fcaf8b889b33552984ecd248c Mon Sep 17 00:00:00 2001 From: vyPal Date: Sun, 27 Jul 2025 17:34:41 +0200 Subject: [PATCH] allow Custom Plugin APIs (#1080) * Wrap Context in Arc * Add support for plugin to expose custom APIs * Add functionality to allow plugins to load in parallel * Remove PluginManager from RwLock --- pumpkin-macros/src/lib.rs | 8 - pumpkin/src/command/commands/plugin.rs | 13 +- pumpkin/src/command/commands/plugins.rs | 3 +- pumpkin/src/lib.rs | 20 +- pumpkin/src/main.rs | 11 +- pumpkin/src/net/query.rs | 6 +- pumpkin/src/plugin/api/context.rs | 66 +++- pumpkin/src/plugin/api/mod.rs | 6 +- pumpkin/src/plugin/mod.rs | 445 ++++++++++++++++-------- pumpkin/src/world/mod.rs | 10 +- 10 files changed, 382 insertions(+), 206 deletions(-) diff --git a/pumpkin-macros/src/lib.rs b/pumpkin-macros/src/lib.rs index 788dd4fce..6cb1d096e 100644 --- a/pumpkin-macros/src/lib.rs +++ b/pumpkin-macros/src/lib.rs @@ -100,8 +100,6 @@ pub fn send_cancellable(input: TokenStream) -> TokenStream { if let Some(cancelled_block) = cancelled_block { quote! { let event = crate::PLUGIN_MANAGER - .read() - .await .fire(#event) .await; @@ -115,8 +113,6 @@ pub fn send_cancellable(input: TokenStream) -> TokenStream { } else { quote! { let event = crate::PLUGIN_MANAGER - .read() - .await .fire(#event) .await; @@ -129,8 +125,6 @@ pub fn send_cancellable(input: TokenStream) -> TokenStream { } else if let Some(cancelled_block) = cancelled_block { quote! { let event = crate::PLUGIN_MANAGER - .read() - .await .fire(#event) .await; @@ -142,8 +136,6 @@ pub fn send_cancellable(input: TokenStream) -> TokenStream { } else { quote! { let event = crate::PLUGIN_MANAGER - .read() - .await .fire(#event) .await; } diff --git a/pumpkin/src/command/commands/plugin.rs b/pumpkin/src/command/commands/plugin.rs index 8262504eb..c741793b9 100644 --- a/pumpkin/src/command/commands/plugin.rs +++ b/pumpkin/src/command/commands/plugin.rs @@ -36,8 +36,7 @@ impl CommandExecutor for ListExecutor { _server: &crate::server::Server, _args: &ConsumedArgs<'a>, ) -> Result<(), CommandError> { - let plugin_manager = PLUGIN_MANAGER.read().await; - let plugins = plugin_manager.active_plugins(); + let plugins = PLUGIN_MANAGER.active_plugins().await; let message_text = if plugins.is_empty() { "There are no loaded plugins.".to_string() @@ -84,9 +83,8 @@ impl CommandExecutor for LoadExecutor { let Some(Arg::Simple(plugin_name)) = args.get(PLUGIN_NAME) else { return Err(InvalidConsumption(Some(PLUGIN_NAME.into()))); }; - let mut plugin_manager = PLUGIN_MANAGER.write().await; - if plugin_manager.is_plugin_active(plugin_name) { + if PLUGIN_MANAGER.is_plugin_active(plugin_name).await { sender .send_message( TextComponent::text(format!("Plugin {plugin_name} is already loaded")) @@ -96,7 +94,7 @@ impl CommandExecutor for LoadExecutor { return Ok(()); } - let result = plugin_manager.try_load_plugin(Path::new(plugin_name)).await; + let result = PLUGIN_MANAGER.try_load_plugin(Path::new(plugin_name)).await; match result { Ok(()) => { @@ -134,9 +132,8 @@ impl CommandExecutor for UnloadExecutor { let Some(Arg::Simple(plugin_name)) = args.get(PLUGIN_NAME) else { return Err(InvalidConsumption(Some(PLUGIN_NAME.into()))); }; - let mut plugin_manager = PLUGIN_MANAGER.write().await; - if !plugin_manager.is_plugin_active(plugin_name) { + if !PLUGIN_MANAGER.is_plugin_active(plugin_name).await { sender .send_message( TextComponent::text(format!("Plugin {plugin_name} is not loaded")) @@ -146,7 +143,7 @@ impl CommandExecutor for UnloadExecutor { return Ok(()); } - let result = plugin_manager.unload_plugin(plugin_name).await; + let result = PLUGIN_MANAGER.unload_plugin(plugin_name).await; match result { Ok(()) => { diff --git a/pumpkin/src/command/commands/plugins.rs b/pumpkin/src/command/commands/plugins.rs index 9ddde8480..a96c5f74a 100644 --- a/pumpkin/src/command/commands/plugins.rs +++ b/pumpkin/src/command/commands/plugins.rs @@ -22,8 +22,7 @@ impl CommandExecutor for Executor { _server: &crate::server::Server, _args: &ConsumedArgs<'a>, ) -> Result<(), CommandError> { - let plugin_manager = PLUGIN_MANAGER.read().await; - let plugins = plugin_manager.active_plugins(); + let plugins = PLUGIN_MANAGER.active_plugins().await; let message_text = if plugins.is_empty() { "There are no loaded plugins.".to_string() diff --git a/pumpkin/src/lib.rs b/pumpkin/src/lib.rs index 907c304b9..ee34d3f29 100644 --- a/pumpkin/src/lib.rs +++ b/pumpkin/src/lib.rs @@ -47,15 +47,8 @@ pub mod world; pub static HEAP_PROFILER: LazyLock>> = LazyLock::new(|| Mutex::new(None)); -pub static PLUGIN_MANAGER: LazyLock>> = LazyLock::new(|| { - let manager = PluginManager::new(); - let arc_manager = Arc::new(RwLock::new(manager)); - let clone = Arc::clone(&arc_manager); - let arc_manager_clone = arc_manager.clone(); - let mut manager = futures::executor::block_on(arc_manager_clone.write()); - manager.set_self_ref(clone); - arc_manager -}); +pub static PLUGIN_MANAGER: LazyLock> = + LazyLock::new(|| Arc::new(PluginManager::new())); pub static PERMISSION_REGISTRY: LazyLock>> = LazyLock::new(|| Arc::new(RwLock::new(PermissionRegistry::new()))); @@ -259,16 +252,15 @@ impl PumpkinServer { } pub async fn init_plugins(&self) { - let mut loader_lock = PLUGIN_MANAGER.write().await; - loader_lock.set_server(self.server.clone()); - if let Err(err) = loader_lock.load_plugins().await { + PLUGIN_MANAGER.set_self_ref(PLUGIN_MANAGER.clone()).await; + PLUGIN_MANAGER.set_server(self.server.clone()).await; + if let Err(err) = PLUGIN_MANAGER.load_plugins().await { log::error!("{err}"); }; } pub async fn unload_plugins(&self) { - let mut loader_lock = PLUGIN_MANAGER.write().await; - if let Err(err) = loader_lock.unload_all_plugins().await { + if let Err(err) = PLUGIN_MANAGER.unload_all_plugins().await { log::error!("Error unloading plugins: {err}"); } else { log::info!("All plugins unloaded successfully"); diff --git a/pumpkin/src/main.rs b/pumpkin/src/main.rs index cc2d011a1..b65bf679e 100644 --- a/pumpkin/src/main.rs +++ b/pumpkin/src/main.rs @@ -75,15 +75,8 @@ static ALLOC: dhat::Alloc = dhat::Alloc; #[cfg(feature = "dhat-heap")] use pumpkin::HEAP_PROFILER; -pub static PLUGIN_MANAGER: LazyLock>> = LazyLock::new(|| { - let manager = PluginManager::new(); - let arc_manager = Arc::new(RwLock::new(manager)); - let clone = Arc::clone(&arc_manager); - let arc_manager_clone = arc_manager.clone(); - let mut manager = futures::executor::block_on(arc_manager_clone.write()); - manager.set_self_ref(clone); - arc_manager -}); +pub static PLUGIN_MANAGER: LazyLock> = + LazyLock::new(|| Arc::new(PluginManager::new())); pub static PERMISSION_REGISTRY: LazyLock>> = LazyLock::new(|| Arc::new(RwLock::new(PermissionRegistry::new()))); diff --git a/pumpkin/src/net/query.rs b/pumpkin/src/net/query.rs index 87b8ac0f3..6e9db7da7 100644 --- a/pumpkin/src/net/query.rs +++ b/pumpkin/src/net/query.rs @@ -14,7 +14,7 @@ use rand::Rng; use tokio::{net::UdpSocket, sync::RwLock, time}; use crate::{ - SHOULD_STOP, STOP_INTERRUPT, + PLUGIN_MANAGER, SHOULD_STOP, STOP_INTERRUPT, server::{CURRENT_MC_VERSION, Server}, }; @@ -140,9 +140,9 @@ async fn handle_packet( } } - let plugin_manager = crate::PLUGIN_MANAGER.read().await; - let plugins = plugin_manager + let plugins = PLUGIN_MANAGER .active_plugins() + .await .into_iter() .map(|meta| meta.name.to_string()) .reduce(|acc, name| format!("{acc}, {name}")) diff --git a/pumpkin/src/plugin/api/context.rs b/pumpkin/src/plugin/api/context.rs index 212bed34c..30a452134 100644 --- a/pumpkin/src/plugin/api/context.rs +++ b/pumpkin/src/plugin/api/context.rs @@ -1,4 +1,4 @@ -use std::{fs, path::Path, path::PathBuf, sync::Arc}; +use std::{any::Any, fs, path::Path, path::PathBuf, sync::Arc}; use crate::command::client_suggestions; use pumpkin_util::{ @@ -26,7 +26,7 @@ pub struct Context { metadata: PluginMetadata<'static>, pub server: Arc, pub handlers: Arc>, - pub plugin_manager: Arc>, + pub plugin_manager: Arc, pub permission_manager: Arc>, } impl Context { @@ -44,7 +44,7 @@ impl Context { metadata: PluginMetadata<'static>, server: Arc, handlers: Arc>, - plugin_manager: Arc>, + plugin_manager: Arc, permission_manager: Arc>, ) -> Self { Self { @@ -80,6 +80,53 @@ impl Context { self.server.get_player_by_name(&player_name).await } + /// Registers a service with the plugin context. + /// + /// This method allows you to associate a service instance with a given name, + /// making it available for retrieval by plugins or other components. + /// The service must be wrapped in an `Arc` and implement `Any`, `Send`, and `Sync`. + /// + /// # Arguments + /// + /// * `name` - The unique name to register the service under. + /// * `service` - The service instance to register, wrapped in an `Arc`. + /// + /// # Example + /// + /// ``` + /// context.register_service("my_service".to_string(), Arc::new(MyService::new())).await; + /// ``` + pub async fn register_service(&self, name: String, service: Arc) { + let mut services = self.plugin_manager.services.write().await; + services.insert(name, service); + } + + /// Retrieves a registered service by name and type. + /// + /// This method attempts to fetch a service previously registered under the given name, + /// and downcasts it to the requested type. Returns `Some(Arc)` if the service exists + /// and the type matches, or `None` otherwise. + /// + /// # Arguments + /// + /// * `name` - The name of the service to retrieve. + /// + /// # Returns + /// + /// An `Option>` containing the service if found and type matches, or `None`. + /// + /// # Example + /// + /// ``` + /// if let Some(service) = context.get_service::("my_service").await { + /// // Use the service + /// } + /// ``` + pub async fn get_service(&self, name: &str) -> Option> { + let services = self.plugin_manager.services.read().await; + services.get(name)?.clone().downcast::().ok() + } + /// Asynchronously registers a command with the server. /// /// # Arguments @@ -171,12 +218,14 @@ impl Context { /// /// # Constraints /// The handler must implement the `EventHandler` trait. - pub async fn register_event + 'static>( + pub async fn register_event( &self, handler: Arc, priority: EventPriority, blocking: bool, - ) { + ) where + H: EventHandler + 'static, + { let mut handlers = self.handlers.write().await; let handlers_vec = handlers @@ -216,10 +265,9 @@ impl Context { &self, loader: Arc, ) -> bool { - let mut manager = self.plugin_manager.write().await; - let before_count = manager.loaded_plugins().len(); - manager.add_loader(loader).await; - let after_count = manager.loaded_plugins().len(); + let before_count = self.plugin_manager.loaded_plugins().await.len(); + self.plugin_manager.add_loader(loader).await; + let after_count = self.plugin_manager.loaded_plugins().await.len(); // Return true if any new plugins were loaded after_count > before_count diff --git a/pumpkin/src/plugin/api/mod.rs b/pumpkin/src/plugin/api/mod.rs index 2350506af..7b0ca7ee4 100644 --- a/pumpkin/src/plugin/api/mod.rs +++ b/pumpkin/src/plugin/api/mod.rs @@ -1,6 +1,8 @@ pub mod context; pub mod events; +use std::sync::Arc; + use async_trait::async_trait; pub use context::*; pub use events::*; @@ -38,7 +40,7 @@ pub trait Plugin: Send + Sync + 'static { /// /// # Returns /// - `Ok(())` on success, or `Err(String)` on failure. - async fn on_load(&mut self, _server: &Context) -> Result<(), String> { + async fn on_load(&mut self, _server: Arc) -> Result<(), String> { Ok(()) } @@ -51,7 +53,7 @@ pub trait Plugin: Send + Sync + 'static { /// /// # Returns /// - `Ok(())` on success, or `Err(String)` on failure. - async fn on_unload(&mut self, _server: &Context) -> Result<(), String> { + async fn on_unload(&mut self, _server: Arc) -> Result<(), String> { Ok(()) } } diff --git a/pumpkin/src/plugin/mod.rs b/pumpkin/src/plugin/mod.rs index 11d497347..fc380c6c7 100644 --- a/pumpkin/src/plugin/mod.rs +++ b/pumpkin/src/plugin/mod.rs @@ -8,7 +8,7 @@ use std::{ sync::Arc, }; use thiserror::Error; -use tokio::sync::RwLock; +use tokio::sync::{Notify, RwLock}; pub mod api; pub mod loader; @@ -128,15 +128,28 @@ where /// and the value is a vector of dynamic event handlers. type HandlerMap = HashMap<&'static str, Vec>>; +/// Plugin loading state +#[derive(Debug, Clone, PartialEq)] +pub enum PluginState { + Loading, + Loaded, + Failed(String), +} + /// Core plugin management system pub struct PluginManager { - plugins: Vec, - loaders: Vec>, - server: Option>, + plugins: RwLock>, + loaders: RwLock>>, + server: RwLock>>, handlers: Arc>, - unloaded_files: HashSet, + unloaded_files: RwLock>, // Self-reference for sharing with contexts - self_ref: Option>>, + self_ref: RwLock>>, + services: Arc>>>, + // Plugin state tracking + plugin_states: RwLock>, + // Notification for plugin state changes + state_notify: Arc, } /// Represents a successfully loaded plugin @@ -145,10 +158,11 @@ pub struct PluginManager { /// - Windows: Plugin cannot be unloaded, it can be only active or not struct LoadedPlugin { metadata: PluginMetadata<'static>, - instance: Box, + instance: Option>, loader: Arc, - loader_data: Box, + loader_data: Option>, is_active: bool, + context: Arc, } /// Error types for plugin management @@ -173,12 +187,15 @@ pub enum ManagerError { impl Default for PluginManager { fn default() -> Self { Self { - plugins: Vec::new(), - loaders: vec![Arc::new(NativePluginLoader)], - server: None, + plugins: RwLock::new(Vec::new()), + loaders: RwLock::new(vec![Arc::new(NativePluginLoader)]), + server: RwLock::new(None), handlers: Arc::new(RwLock::new(HashMap::new())), - unloaded_files: HashSet::new(), - self_ref: None, + unloaded_files: RwLock::new(HashSet::new()), + self_ref: RwLock::new(None), + services: Arc::new(RwLock::new(HashMap::new())), + plugin_states: RwLock::new(HashMap::new()), + state_notify: Arc::new(Notify::new()), } } } @@ -191,16 +208,18 @@ impl PluginManager { } /// Unload all loaded plugins - pub async fn unload_all_plugins(&mut self) -> Result<(), ManagerError> { - let plugin_names: Vec<&str> = self - .plugins - .iter() - .filter(|p| p.is_active) - .map(|p| p.metadata.name) - .collect(); + pub async fn unload_all_plugins(&self) -> Result<(), ManagerError> { + let plugin_names: Vec = { + let plugins = self.plugins.read().await; + plugins + .iter() + .filter(|p| p.is_active) + .map(|p| p.metadata.name.to_string()) + .collect() + }; for name in plugin_names { - if let Err(e) = self.unload_plugin(name).await { + if let Err(e) = self.unload_plugin(&name).await { log::error!("Failed to unload plugin {name}: {e}"); } } @@ -209,41 +228,49 @@ impl PluginManager { } /// Add a new plugin loader implementation - pub async fn add_loader(&mut self, loader: Arc) { - self.loaders.push(loader); + pub async fn add_loader(&self, loader: Arc) { + self.loaders.write().await.push(loader); // Try to load previously unloaded files with the new loader self.retry_unloaded_files().await; } /// Retry loading files that couldn't be loaded previously - async fn retry_unloaded_files(&mut self) { - let files_to_retry: Vec = self.unloaded_files.iter().cloned().collect(); + async fn retry_unloaded_files(&self) { + let files_to_retry: Vec = + { self.unloaded_files.read().await.iter().cloned().collect() }; + let mut retry_tasks = Vec::new(); + for path in files_to_retry { - if matches!(self.try_load_plugin(&path).await, Ok(())) { - self.unloaded_files.remove(&path); + if let Ok(task) = self.start_loading_plugin(&path).await { + retry_tasks.push(task); } } + + // Wait for all retry tasks to complete + join_all(retry_tasks).await; } /// Set server reference for plugin context - pub fn set_server(&mut self, server: Arc) { - self.server = Some(server); + pub async fn set_server(&self, server: Arc) { + let mut srv = self.server.write().await; + srv.replace(server); } /// Set self reference for creating contexts - pub fn set_self_ref(&mut self, self_ref: Arc>) { - self.self_ref = Some(self_ref); + pub async fn set_self_ref(&self, self_ref: Arc) { + let mut sref = self.self_ref.write().await; + sref.replace(self_ref); } /// Get a clone of the loaders for context use #[must_use] - pub fn get_loaders(&self) -> Vec> { - self.loaders.clone() + pub async fn get_loaders(&self) -> Vec> { + self.loaders.read().await.clone() } /// Load all plugins from the plugin directory - pub async fn load_plugins(&mut self) -> Result<(), ManagerError> { + pub async fn load_plugins(&self) -> Result<(), ManagerError> { const PLUGIN_DIR: &str = "./plugins"; let path = Path::new(PLUGIN_DIR); @@ -252,6 +279,8 @@ impl PluginManager { return Ok(()); } + let mut load_tasks = Vec::new(); + for entry in std::fs::read_dir(path)? { let entry = entry?; let path = entry.path(); @@ -260,163 +289,291 @@ impl PluginManager { continue; } - if let Err(err) = self.try_load_plugin(&path).await { - log::error!("{err}"); + // Start loading plugin concurrently + if let Ok(task) = self.start_loading_plugin(&path).await { + load_tasks.push(task); } } + // Wait for all plugins to complete loading (but don't block on individual plugin initialization) + join_all(load_tasks).await; + Ok(()) } - /// Attempt to load a single plugin file - pub async fn try_load_plugin(&mut self, path: &Path) -> Result<(), ManagerError> { - for loader in &self.loaders { + /// Start loading a plugin asynchronously + async fn start_loading_plugin( + &self, + path: &Path, + ) -> Result, ManagerError> { + let loaders = self.loaders.read().await; + for loader in loaders.iter() { if loader.can_load(path) { - match self.load_with_loader(loader, path).await { - Ok(plugin) => { - log::info!( - "Loaded {} ({})", - plugin.metadata.name, - plugin.metadata.version - ); - self.plugins.push(plugin); - // Remove from unloaded files if it was there - self.unloaded_files.remove(path); - return Ok(()); + let server = self.server.read().await; + let self_ref = self.self_ref.read().await; + + let (mut instance, metadata, loader_data) = loader.load(path).await?; + + // Mark plugin as loading + self.plugin_states + .write() + .await + .insert(metadata.name.to_string(), PluginState::Loading); + + let self_ref = self_ref + .as_ref() + .ok_or(ManagerError::ServerNotInitialized)?; + + let context = Arc::new(Context::new( + metadata.clone(), + Arc::clone(server.as_ref().ok_or(ManagerError::ServerNotInitialized)?), + Arc::clone(&self.handlers), + Arc::clone(self_ref), + Arc::clone(&PERMISSION_MANAGER), + )); + + // Create the plugin structure first + let plugin = LoadedPlugin { + metadata: metadata.clone(), + instance: None, // Will be set after successful initialization + loader: loader.clone(), + loader_data: Some(loader_data), + is_active: false, // Will be set to true after successful initialization + context: context.clone(), + }; + + let plugin_index = { + let mut plugins = self.plugins.write().await; + plugins.push(plugin); + plugins.len() - 1 + }; + + // Remove from unloaded files if it was there + self.unloaded_files.write().await.remove(path); + + // Spawn async task for plugin initialization + let self_ref_clone = Arc::clone(self_ref); + let state_notify = Arc::clone(&self.state_notify); + let plugin_name = metadata.name.to_string(); + let loader_clone = loader.clone(); + + let task = tokio::spawn(async move { + // Initialize the plugin + match instance.on_load(context.clone()).await { + Ok(()) => { + // Update plugin state to loaded + { + let mut plugins = self_ref_clone.plugins.write().await; + if let Some(plugin) = plugins.get_mut(plugin_index) { + plugin.instance = Some(instance); + plugin.is_active = true; + } + } + self_ref_clone + .plugin_states + .write() + .await + .insert(plugin_name.clone(), PluginState::Loaded); + state_notify.notify_waiters(); + + log::info!("Loaded {} ({})", metadata.name, metadata.version); + } + Err(e) => { + // Handle initialization failure + let error_msg = format!("Initialization failed: {e}"); + let _ = instance.on_unload(context).await; + + // Get the loader data before removing the plugin + let loader_data: Option> = { + let mut plugins = self_ref_clone.plugins.write().await; + if let Some(plugin) = plugins.get_mut(plugin_index) { + plugin.loader_data.take() + } else { + None + } + }; + + // Try to unload the plugin data + if let Some(data) = loader_data { + tokio::spawn(async move { + loader_clone.unload(data).await.ok(); + }); + } + + { + let mut plugins = self_ref_clone.plugins.write().await; + if plugin_index < plugins.len() { + plugins.remove(plugin_index); + } + }; + self_ref_clone.plugin_states.write().await.insert( + plugin_name.clone(), + PluginState::Failed(error_msg.clone()), + ); + state_notify.notify_waiters(); + + log::error!("Failed to initialize plugin {plugin_name}: {error_msg}",); + } } - Err(e) => { - log::error!("Failed to load plugin {}: {}", path.display(), e); - return Ok(()); - } - } + }); + + return Ok(task); } } // No loader could handle this file, track it for future attempts - self.unloaded_files.insert(path.to_path_buf()); + self.unloaded_files.write().await.insert(path.to_path_buf()); Err(ManagerError::PluginNotFound( path.to_string_lossy().to_string(), )) } - /// Load plugin using a specific loader - async fn load_with_loader( - &self, - loader: &Arc, - path: &Path, - ) -> Result { - let server = self - .server - .as_ref() - .ok_or(ManagerError::ServerNotInitialized)?; - let (mut instance, metadata, loader_data) = loader.load(path).await?; - - // Get a self_ref for the context or fail if not set - let self_ref = self - .self_ref - .as_ref() - .ok_or(ManagerError::ServerNotInitialized)?; - - let context = Context::new( - metadata.clone(), - Arc::clone(server), - Arc::clone(&self.handlers), - Arc::clone(self_ref), - Arc::clone(&PERMISSION_MANAGER), - ); - - if let Err(e) = instance.on_load(&context).await { - let data = loader_data; - let loader = loader.clone(); - let _ = instance.on_unload(&context).await; - tokio::spawn(async move { - loader.unload(data).await.ok(); - }); - return Err(ManagerError::LoaderError( - LoaderError::InitializationFailed(e), - )); - } - - Ok(LoadedPlugin { - metadata, - instance, - loader: loader.clone(), - loader_data, - is_active: true, + /// Attempt to load a single plugin file + pub async fn try_load_plugin(&self, path: &Path) -> Result<(), ManagerError> { + self.start_loading_plugin(path).await?.await.map_err(|e| { + ManagerError::LoaderError(LoaderError::InitializationFailed(format!( + "Task join error: {e}" + ))) }) } + /// Wait for a plugin to finish loading + pub async fn wait_for_plugin(&self, plugin_name: &str) -> Result<(), ManagerError> { + loop { + let state = self.plugin_states.read().await.get(plugin_name).cloned(); + if let Some(state) = state { + match state { + PluginState::Loaded => return Ok(()), + PluginState::Failed(error) => { + return Err(ManagerError::LoaderError( + LoaderError::InitializationFailed(error.clone()), + )); + } + PluginState::Loading => { + // Wait for state change notification + self.state_notify.notified().await; + continue; + } + } + } + return Err(ManagerError::PluginNotFound(plugin_name.to_string())); + } + } + + /// Get the current state of a plugin + pub async fn get_plugin_state(&self, plugin_name: &str) -> Option { + self.plugin_states.read().await.get(plugin_name).cloned() + } + /// Checks if plugin active #[must_use] - pub fn is_plugin_active(&self, name: &str) -> bool { - self.plugins + pub async fn is_plugin_active(&self, name: &str) -> bool { + let plugins = self.plugins.read().await; + plugins .iter() - .any(|p| p.metadata.name == name && p.is_active) + .any(|p| p.metadata.name == name && p.is_active && p.instance.is_some()) } /// Get list of active plugins #[must_use] - pub fn active_plugins(&self) -> Vec<&PluginMetadata> { - self.plugins + pub async fn active_plugins(&self) -> Vec> { + let plugins = self.plugins.read().await; + plugins .iter() - .filter(|p| p.is_active) - .map(|p| &p.metadata) + .filter(|p| p.is_active && p.instance.is_some()) + .map(|p| p.metadata.clone()) .collect() } /// Checks if plugin loaded #[must_use] - pub fn is_plugin_loaded(&self, name: &str) -> bool { - self.plugins.iter().any(|p| p.metadata.name == name) + pub async fn is_plugin_loaded(&self, name: &str) -> bool { + let plugins = self.plugins.read().await; + plugins.iter().any(|p| p.metadata.name == name) } /// Get list of loaded plugins #[must_use] - pub fn loaded_plugins(&self) -> Vec<&PluginMetadata> { - self.plugins.iter().map(|p| &p.metadata).collect() + pub async fn loaded_plugins(&self) -> Vec> { + let plugins = self.plugins.read().await; + plugins.iter().map(|p| p.metadata.clone()).collect() } /// Unload a plugin by name - pub async fn unload_plugin(&mut self, name: &str) -> Result<(), ManagerError> { - let index = self - .plugins - .iter() - .position(|p| p.metadata.name == name) - .ok_or_else(|| ManagerError::PluginNotFound(name.to_string()))?; + pub async fn unload_plugin(&self, name: &str) -> Result<(), ManagerError> { + let index = { + let plugins = self.plugins.read().await; + plugins + .iter() + .position(|p| p.metadata.name == name) + .ok_or_else(|| ManagerError::PluginNotFound(name.to_string()))? + }; - let mut plugin = self.plugins.remove(index); - let server = self - .server - .as_ref() - .ok_or(ManagerError::ServerNotInitialized)?; + let mut plugin = { + let mut plugins = self.plugins.write().await; + plugins.remove(index) + }; - // Get a self_ref for the context or fail if not set - let self_ref = self - .self_ref - .as_ref() - .ok_or(ManagerError::ServerNotInitialized)?; - - let context = Context::new( - plugin.metadata.clone(), - Arc::clone(server), - Arc::clone(&self.handlers), - Arc::clone(self_ref), - Arc::clone(&PERMISSION_MANAGER), - ); - - plugin.instance.on_unload(&context).await.ok(); - - if plugin.loader.can_unload() { - plugin.loader.unload(plugin.loader_data).await?; - } else { - plugin.is_active = false; - self.plugins.push(plugin); + if let Some(mut instance) = plugin.instance.take() { + instance.on_unload(plugin.context.clone()).await.ok(); } + if plugin.loader.can_unload() { + if let Some(data) = plugin.loader_data { + plugin.loader.unload(data).await?; + } + } else { + plugin.is_active = false; + self.plugins.write().await.push(plugin); + } + + // Remove from plugin states + self.plugin_states.write().await.remove(name); + Ok(()) } + /// Get all plugins that are currently loading + pub async fn get_loading_plugins(&self) -> Vec { + let plugin_states = self.plugin_states.read().await; + plugin_states + .iter() + .filter(|(_, state)| matches!(state, PluginState::Loading)) + .map(|(name, _)| name.clone()) + .collect() + } + + /// Get all plugins that failed to load + pub async fn get_failed_plugins(&self) -> Vec<(String, String)> { + let plugin_states = self.plugin_states.read().await; + plugin_states + .iter() + .filter_map(|(name, state)| { + if let PluginState::Failed(error) = state { + Some((name.clone(), error.clone())) + } else { + None + } + }) + .collect() + } + + /// Check if all plugins have finished loading (either succeeded or failed) + pub async fn all_plugins_loaded(&self) -> bool { + let plugin_states = self.plugin_states.read().await; + !plugin_states + .values() + .any(|state| matches!(state, PluginState::Loading)) + } + + /// Wait for all plugins to finish loading + pub async fn wait_for_all_plugins(&self) { + while !self.all_plugins_loaded().await { + self.state_notify.notified().await; + } + } + /// Register an event handler pub async fn register(&self, handler: Arc, priority: EventPriority, blocking: bool) where @@ -439,7 +596,7 @@ impl PluginManager { /// Fire an event to all registered handlers pub async fn fire(&self, mut event: E) -> E { - if let Some(server) = &self.server { + if let Some(server) = self.server.read().await.as_ref() { let handlers = self.handlers.read().await; if let Some(handlers) = handlers.get(&E::get_name_static()) { let (blocking, non_blocking): (Vec<_>, Vec<_>) = diff --git a/pumpkin/src/world/mod.rs b/pumpkin/src/world/mod.rs index 046faeb49..676c3d3fa 100644 --- a/pumpkin/src/world/mod.rs +++ b/pumpkin/src/world/mod.rs @@ -1732,7 +1732,7 @@ impl World { .color_named(NamedColor::Yellow); let event = PlayerJoinEvent::new(player.clone(), msg_comp); - let event = PLUGIN_MANAGER.read().await.fire(event).await; + let event = PLUGIN_MANAGER.fire(event).await; if !event.cancelled { let current_players = current_players.clone(); @@ -1785,7 +1785,7 @@ impl World { .color_named(NamedColor::Yellow); let event = PlayerLeaveEvent::new(player.clone(), msg_comp); - let event = PLUGIN_MANAGER.read().await.fire(event).await; + let event = PLUGIN_MANAGER.fire(event).await; if !event.cancelled { let players = self.players.read().await; @@ -2007,11 +2007,7 @@ impl World { let (broken_block, broken_block_state) = self.get_block_and_state_id(position).await; let event = BlockBreakEvent::new(cause.clone(), broken_block, *position, 0, false); - let event = PLUGIN_MANAGER - .read() - .await - .fire::(event) - .await; + let event = PLUGIN_MANAGER.fire::(event).await; if !event.cancelled { let new_state_id = if broken_block