majjo/Pumpkin
1
0
Fork 0
mirror of https://github.com/Pumpkin-MC/Pumpkin.git synced 2026-08-31 20:32:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

allow Custom Plugin APIs (#1080)

Browse Source 
* 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
This commit is contained in:
vyPal
2025-07-27 17:34:41 +02:00
committed by GitHub
parent 27bd9b15c3
commit f463b5bc81
10 changed files with 382 additions and 206 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
pumpkin-macros/src/lib.rs
View File

@@ -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;
}

13
pumpkin/src/command/commands/plugin.rs
View File

@@ -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(()) => {

3
pumpkin/src/command/commands/plugins.rs
View File

@@ -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()

20
pumpkin/src/lib.rs
View File

@@ -47,15 +47,8 @@ pub mod world;
pub static HEAP_PROFILER: LazyLock<Mutex<Option<dhat::Profiler>>> =
LazyLock::new(|| Mutex::new(None));
pub static PLUGIN_MANAGER: LazyLock<Arc<RwLock<PluginManager>>> = 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<Arc<PluginManager>> =
LazyLock::new(|| Arc::new(PluginManager::new()));
pub static PERMISSION_REGISTRY: LazyLock<Arc<RwLock<PermissionRegistry>>> =
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");

11
pumpkin/src/main.rs
View File

@@ -75,15 +75,8 @@ static ALLOC: dhat::Alloc = dhat::Alloc;
#[cfg(feature = "dhat-heap")]
use pumpkin::HEAP_PROFILER;
pub static PLUGIN_MANAGER: LazyLock<Arc<RwLock<PluginManager>>> = 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<Arc<PluginManager>> =
LazyLock::new(|| Arc::new(PluginManager::new()));
pub static PERMISSION_REGISTRY: LazyLock<Arc<RwLock<PermissionRegistry>>> =
LazyLock::new(|| Arc::new(RwLock::new(PermissionRegistry::new())));

6
pumpkin/src/net/query.rs
View File

@@ -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}"))

66
pumpkin/src/plugin/api/context.rs
View File

@@ -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<Server>,
pub handlers: Arc<RwLock<HandlerMap>>,
pub plugin_manager: Arc<RwLock<PluginManager>>,
pub plugin_manager: Arc<PluginManager>,
pub permission_manager: Arc<RwLock<PermissionManager>>,
}
impl Context {
@@ -44,7 +44,7 @@ impl Context {
metadata: PluginMetadata<'static>,
server: Arc<Server>,
handlers: Arc<RwLock<HandlerMap>>,
plugin_manager: Arc<RwLock<PluginManager>>,
plugin_manager: Arc<PluginManager>,
permission_manager: Arc<RwLock<PermissionManager>>,
) -> 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<T: Any + Send + Sync>(&self, name: String, service: Arc<T>) {
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<T>)` if the service exists
/// and the type matches, or `None` otherwise.
///
/// # Arguments
///
/// * `name` - The name of the service to retrieve.
///
/// # Returns
///
/// An `Option<Arc<T>>` containing the service if found and type matches, or `None`.
///
/// # Example
///
/// ```
/// if let Some(service) = context.get_service::<MyService>("my_service").await {
/// // Use the service
/// }
/// ```
pub async fn get_service<T: Any + Send + Sync>(&self, name: &str) -> Option<Arc<T>> {
let services = self.plugin_manager.services.read().await;
services.get(name)?.clone().downcast::<T>().ok()
}
/// Asynchronously registers a command with the server.
///
/// # Arguments
@@ -171,12 +218,14 @@ impl Context {
///
/// # Constraints
/// The handler must implement the `EventHandler<E>` trait.
pub async fn register_event<E: Event + 'static, H: EventHandler<E> + 'static>(
pub async fn register_event<E: Event + 'static, H>(
&self,
handler: Arc<H>,
priority: EventPriority,
blocking: bool,
) {
) where
H: EventHandler<E> + 'static,
{
let mut handlers = self.handlers.write().await;
let handlers_vec = handlers
@@ -216,10 +265,9 @@ impl Context {
&self,
loader: Arc<dyn crate::plugin::loader::PluginLoader>,
) -> 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

6
pumpkin/src/plugin/api/mod.rs
View File

@@ -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<Context>) -> 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<Context>) -> Result<(), String> {
Ok(())
}
}

445
pumpkin/src/plugin/mod.rs
View File

@@ -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<Box<dyn DynEventHandler>>>;
/// Plugin loading state
#[derive(Debug, Clone, PartialEq)]
pub enum PluginState {
Loading,
Loaded,
Failed(String),
}
/// Core plugin management system
pub struct PluginManager {
plugins: Vec<LoadedPlugin>,
loaders: Vec<Arc<dyn PluginLoader>>,
server: Option<Arc<Server>>,
plugins: RwLock<Vec<LoadedPlugin>>,
loaders: RwLock<Vec<Arc<dyn PluginLoader>>>,
server: RwLock<Option<Arc<Server>>>,
handlers: Arc<RwLock<HandlerMap>>,
unloaded_files: HashSet<PathBuf>,
unloaded_files: RwLock<HashSet<PathBuf>>,
// Self-reference for sharing with contexts
self_ref: Option<Arc<RwLock<PluginManager>>>,
self_ref: RwLock<Option<Arc<PluginManager>>>,
services: Arc<RwLock<HashMap<String, Arc<dyn Any + Send + Sync>>>>,
// Plugin state tracking
plugin_states: RwLock<HashMap<String, PluginState>>,
// Notification for plugin state changes
state_notify: Arc<Notify>,
}
/// 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<dyn Plugin>,
instance: Option<Box<dyn Plugin>>,
loader: Arc<dyn PluginLoader>,
loader_data: Box<dyn Any + Send + Sync>,
loader_data: Option<Box<dyn Any + Send + Sync>>,
is_active: bool,
context: Arc<Context>,
}
/// 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<String> = {
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<dyn PluginLoader>) {
self.loaders.push(loader);
pub async fn add_loader(&self, loader: Arc<dyn PluginLoader>) {
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<PathBuf> = self.unloaded_files.iter().cloned().collect();
async fn retry_unloaded_files(&self) {
let files_to_retry: Vec<PathBuf> =
{ 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<Server>) {
self.server = Some(server);
pub async fn set_server(&self, server: Arc<Server>) {
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<RwLock<Self>>) {
self.self_ref = Some(self_ref);
pub async fn set_self_ref(&self, self_ref: Arc<Self>) {
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<Arc<dyn PluginLoader>> {
self.loaders.clone()
pub async fn get_loaders(&self) -> Vec<Arc<dyn PluginLoader>> {
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<tokio::task::JoinHandle<()>, 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<Box<dyn Any + Send + Sync>> = {
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<dyn PluginLoader>,
path: &Path,
) -> Result<LoadedPlugin, ManagerError> {
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<PluginState> {
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<PluginMetadata<'static>> {
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<PluginMetadata<'static>> {
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<String> {
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<E, H>(&self, handler: Arc<H>, priority: EventPriority, blocking: bool)
where
@@ -439,7 +596,7 @@ impl PluginManager {
/// Fire an event to all registered handlers
pub async fn fire<E: Event + Send + Sync + 'static>(&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<_>) =

10
pumpkin/src/world/mod.rs
View File

@@ -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::<BlockBreakEvent>(event)
.await;
let event = PLUGIN_MANAGER.fire::<BlockBreakEvent>(event).await;
if !event.cancelled {
let new_state_id = if broken_block