Function process_node

src/executor.rs:258–385

(
    scope: &rayon::Scope<'scope>,
    ctx: &'scope ExecutionContext<'a, 'scope, W>,
    node: &'scope Node<'a>,
)

/// It handles Serialization -> Compression -> Writing -> Notification.
#[cfg(all(feature = "parallel", not(target_arch = "wasm32")))]
fn process_node<'scope, 'a, W>(
    scope: &rayon::Scope<'scope>,
    ctx: &'scope ExecutionContext<'a, 'scope, W>,
    node: &'scope Node<'a>,
) where
    W: Write + Send,
{
    // 0. Fast abort check
    if ctx.should_abort() {
        return;
    }

    // 1. Prepare data

    let completed_children_raw = {
        let lock = node
            .completed_children
            .lock()
            .map_err(|_| ParcodeError::Internal("Node mutex poisoned".into()));
        match lock {
            Ok(mut guard) => std::mem::take(&mut *guard),
            Err(e) => {
                ctx.signal_error(e);
                return;
            }
        }
    };

    let children_refs: Vec<ChildRef> = match completed_children_raw
        .into_iter()
        .map(|opt| opt.ok_or_else(|| ParcodeError::Internal("Missing child result".into())))
        .collect()
    {
        Ok(v) => v,
        Err(e) => {
            ctx.signal_error(e);
            return;
        }
    };

    let is_chunkable = !children_refs.is_empty();

    let raw_payload = match node.job.execute(&children_refs) {
        Ok(bytes) => bytes,
        Err(e) => {
            ctx.signal_error(e);
            return;
        }
    };

    // 2&3. Compression and formatting

    let config = node.job.config();
    let compression_id = if ctx.use_compression && config.compression_id == 0 {
        1
    } else {
        config.compression_id
    };