Method emit_put_event

nodedb/src/data/executor/core_loop/event_emit.rs:41–71

Emit a point write/overwrite/update event derived from the new bytes produced by the handler and the prior bytes returned from storage. Shared by every handler that runs a put-style mutation against a document engine: PointPut, Upsert (both branches), batched PointPut, columnar-row overwrite. Each of these knows its *new* bytes and receives *prior* bytes from the storage API; the Event Plane payl

(
        &mut self,
        task: &super::super::task::ExecutionTask,
        tid: u64,
        collection: &str,
        row_id: &str,
        new_stored: &[u8],
        prior_stored: Option<&[u8]>,

    /// applying the strict→msgpack shim, and the `WriteOp` tag is computed
    /// from their presence.
    pub(in crate::data::executor) fn emit_put_event(
        &mut self,
        task: &super::super::task::ExecutionTask,
        tid: u64,
        collection: &str,
        row_id: &str,
        new_stored: &[u8],
        prior_stored: Option<&[u8]>,
    ) {
        let new_converted = self.resolve_event_payload(tid, collection, new_stored);
        let old_converted =
            prior_stored.and_then(|p| self.resolve_event_payload(tid, collection, p));
        let old_bytes: Option<&[u8]> = match (prior_stored, old_converted.as_deref()) {
            (Some(_), Some(c)) => Some(c),
            (Some(raw), None) => Some(raw),
            (None, _) => None,
        };
        let op = if old_bytes.is_some() {
            crate::event::WriteOp::Update
        } else {
            crate::event::WriteOp::Insert
        };
        self.emit_write_event(
            task,
            collection,
            op,
            row_id,
            Some(new_converted.as_deref().unwrap_or(new_stored)),
            old_bytes,
        );
    }

    /// Set the Event Plane producer (called after open, before event loop).
    pub fn set_event_producer(&mut self, producer: crate::event::bus::EventProducer) {