feat(compilation): add queue handlers representing task_queue

crates/node_binding/src/plugins/mod.rs

@@ -9,7 +9,9 @@ use rspack_binding_values::JsExecuteModuleArg;
 use rspack_binding_values::{AfterResolveData, JsChunkAssetArgs, JsModule};
 use rspack_binding_values::{BeforeResolveData, JsAssetEmittedArgs, ToJsModule};
 use rspack_binding_values::{JsResolveForSchemeInput, JsResolveForSchemeResult};
-use rspack_core::{ChunkAssetArgs, ModuleIdentifier, NormalModuleAfterResolveArgs};
+use rspack_core::{
+  ChunkAssetArgs, ModuleIdentifier, NormalModuleAfterResolveArgs, SucceedModuleArgs,
+};
 use rspack_core::{NormalModuleBeforeResolveArgs, PluginNormalModuleFactoryAfterResolveOutput};
 use rspack_core::{PluginNormalModuleFactoryBeforeResolveOutput, ResourceData};
 use rspack_core::{PluginNormalModuleFactoryResolveForSchemeOutput, PluginShouldEmitHookOutput};
@@ -733,11 +735,12 @@ impl rspack_core::Plugin for JsHooksAdapter {
       .map_err(|err| internal_error!("Failed to call after emit: {err}"))?
   }
 
-  async fn succeed_module(&self, args: &dyn rspack_core::Module) -> rspack_error::Result<()> {
+  async fn succeed_module(&self, args: &SucceedModuleArgs<'_>) -> rspack_error::Result<()> {
     if self.is_hook_disabled(&Hook::SucceedModule) {
       return Ok(());
     }
     let js_module = args
+      .module
       .to_js_module()
       .expect("Failed to convert module to JsModule");
     self

crates/rspack_core/src/compiler/compilation.rs

@@ -33,7 +33,7 @@ use super::{
 use crate::{
   build_chunk_graph::build_chunk_graph,
   cache::{use_code_splitting_cache, Cache, CodeSplittingCache},
-  is_source_equal,
+  create_queue_handle, is_source_equal,
   tree_shaking::{optimizer, visitor::SymbolRef, BailoutFlag, OptimizeDependencyResult},
   AddQueue, AddTask, AddTaskResult, AdditionalChunkRuntimeRequirementsArgs,
   AdditionalModuleRequirementsArgs, AsyncDependenciesBlock, BoxDependency, BoxModule, BuildQueue,
@@ -44,9 +44,9 @@ use crate::{
   DependencyParents, DependencyType, Entry, EntryData, EntryOptions, Entrypoint, FactorizeQueue,
   FactorizeTask, FactorizeTaskResult, Filename, Logger, Module, ModuleCreationCallback,
   ModuleGraph, ModuleIdentifier, ModuleProfile, ModuleType, PathData, ProcessAssetsArgs,
-  ProcessDependenciesQueue, ProcessDependenciesResult, ProcessDependenciesTask, RenderManifestArgs,
-  Resolve, ResolverFactory, RuntimeGlobals, RuntimeModule, RuntimeSpec, SharedPluginDriver,
-  SourceType, Stats, TaskResult, WorkerTask,
+  ProcessDependenciesQueue, ProcessDependenciesResult, ProcessDependenciesTask, QueueHandler,
+  RenderManifestArgs, Resolve, ResolverFactory, RuntimeGlobals, RuntimeModule, RuntimeSpec,
+  SharedPluginDriver, SourceType, Stats, TaskResult, WorkerTask,
 };
 use crate::{tree_shaking::visitor::OptimizeAnalyzeResult, Context};
 
@@ -110,6 +110,8 @@ pub struct Compilation {
   pub build_dependencies: IndexSet<PathBuf, BuildHasherDefault<FxHasher>>,
   pub side_effects_free_modules: IdentifierSet,
   pub module_item_map: IdentifierMap<Vec<ModuleItem>>,
+
+  pub queue_handle: Option<QueueHandler>,
 }
 
 impl Compilation {
@@ -170,6 +172,8 @@ impl Compilation {
       side_effects_free_modules: IdentifierSet::default(),
       module_item_map: IdentifierMap::default(),
       include_module_ids: IdentifierSet::default(),
+
+      queue_handle: None,
     }
   }
 
@@ -455,14 +459,19 @@ impl Compilation {
     let mut active_task_count = 0usize;
     let is_expected_shutdown = Arc::new(AtomicBool::new(false));
     let (result_tx, mut result_rx) = tokio::sync::mpsc::unbounded_channel::<Result<TaskResult>>();
+
     let mut factorize_queue = FactorizeQueue::new();
     let mut add_queue = AddQueue::new();
     let mut build_queue = BuildQueue::new();
     let mut process_dependencies_queue = ProcessDependenciesQueue::new();
+
     let mut make_failed_dependencies: HashSet<BuildDependency> = HashSet::default();
     let mut make_failed_module = HashSet::default();
     let mut errored = None;
 
+    let (handler, mut processor) = create_queue_handle();
+    self.queue_handle = Some(handler);
+
     deps_builder
       .revoke_modules(&mut self.module_graph)
       .into_iter()
@@ -515,6 +524,14 @@ impl Compilation {
 
     tokio::task::block_in_place(|| loop {
       let start = factorize_time.start();
+      processor.try_process(
+        self,
+        &mut factorize_queue,
+        &mut add_queue,
+        &mut build_queue,
+        &mut process_dependencies_queue,
+      );
+
       while let Some(task) = factorize_queue.get_task() {
         tokio::spawn({
           let result_tx = result_tx.clone();
@@ -677,6 +694,37 @@ impl Compilation {
 
       match result_rx.try_recv() {
         Ok(item) => {
+          if let Ok(item) = &item {
+            match item {
+              TaskResult::Factorize(result) => processor.flush_callback(
+                crate::TaskCategory::Factorize,
+                &result.dependency.to_string(),
+                result.module_graph_module.module_identifier,
+                self,
+              ),
+              TaskResult::Add(result) => {
+                let module = match result.as_ref() {
+                  AddTaskResult::ModuleReused { module } => module.identifier(),
+                  AddTaskResult::ModuleAdded { module, .. } => module.identifier(),
+                };
+
+                processor.flush_callback(crate::TaskCategory::Add, &module, module, self)
+              }
+              TaskResult::Build(result) => {
+                let id = result.module.identifier();
+                processor.flush_callback(crate::TaskCategory::Build, &id, id, self);
+              }
+              TaskResult::ProcessDependencies(result) => {
+                processor.flush_callback(
+                  crate::TaskCategory::Build,
+                  &result.module_identifier,
+                  result.module_identifier,
+                  self,
+                );
+              }
+            }
+          }
+
           match item {
             Ok(TaskResult::Factorize(box task_result)) => {
               let FactorizeTaskResult {