wasmeld/components/resident-probe/src/lib.rs

//! Resident event/effect fixture covering every current Host-driven I/O family.
//!
//! The callback echoes stream chunks and datagrams, rearms fired timers,
//! acknowledges messages, and reflects extension commands. It owns no socket,
//! timer, or subscription handles; those resources remain in the Host driver.

use std::sync::atomic::{AtomicU64, Ordering};

mod bindings {
    // One generated module contains both the base service export and resident
    // actor export because this component world composes both WIT packages.
    wit_bindgen::generate!({
        path: "wit",
        world: "resident-probe-component",
        generate_all,
    });
}

use bindings::exports::wasmeld::resident::actor::{
    DatagramSend, Effect, Event, Guest as ResidentGuest, MessageAck, SourceCommand, StreamWrite,
    TimerArm,
};

static EVENTS_HANDLED: AtomicU64 = AtomicU64::new(0);

struct ResidentProbe;

impl bindings::Guest for ResidentProbe {
    fn init(_config: Vec<u8>) -> Result<(), bindings::ServiceError> {
        EVENTS_HANDLED.store(0, Ordering::Relaxed);
        Ok(())
    }

    fn invoke(_input: Vec<u8>) -> Result<Vec<u8>, bindings::ServiceError> {
        Ok(EVENTS_HANDLED
            .load(Ordering::Relaxed)
            .to_le_bytes()
            .to_vec())
    }
}

impl ResidentGuest for ResidentProbe {
    fn handle_event(
        input: Event,
    ) -> Result<Vec<Effect>, bindings::exports::wasmeld::resident::actor::ResidentError> {
        // Event callbacks are serialized with normal invocations by the same
        // Actor mailbox, so this in-memory count is deterministic per Actor.
        EVENTS_HANDLED.fetch_add(1, Ordering::Relaxed);
        Ok(match input {
            Event::StreamData(chunk) => vec![Effect::WriteStream(StreamWrite {
                stream_id: chunk.stream_id,
                bytes: chunk.bytes,
            })],
            Event::Datagram(datagram) => vec![Effect::SendDatagram(DatagramSend {
                endpoint_id: datagram.endpoint_id,
                peer: datagram.peer,
                bytes: datagram.bytes,
            })],
            Event::Timer(timer) => vec![Effect::ArmTimer(TimerArm {
                timer_id: timer.timer_id,
                delay_ms: 10,
                interval_ms: None,
            })],
            Event::Message(message) => vec![Effect::AcknowledgeMessage(MessageAck {
                subscription_id: message.subscription_id,
                message_id: message.message_id,
            })],
            Event::Source(source) => vec![Effect::SourceCommand(SourceCommand {
                source_id: source.source_id,
                command: source.kind,
                payload: source.payload,
            })],
            Event::StreamOpened(_)
            | Event::StreamWritable(_)
            | Event::StreamHalfClosed(_)
            | Event::StreamClosed(_)
            | Event::Shutdown => Vec::new(),
        })
    }
}

bindings::export!(ResidentProbe with_types_in bindings);