futuresdr 0.6.0

use std::future::Future;

use crate::runtime::BlockId;
use crate::runtime::BlockPortCtx;
use crate::runtime::Error;
use crate::runtime::Pmt;
use crate::runtime::PortId;
use crate::runtime::PortIndex;
use crate::runtime::PortName;
use crate::runtime::Result;
use crate::runtime::buffer::DynBufferReader;
use crate::runtime::buffer::DynBufferWriter;
use crate::runtime::buffer::PortInboxes;
use crate::runtime::dev::BlockMeta;
use crate::runtime::dev::MessageOutputs;
use crate::runtime::dev::WorkIo;

/// Internal block interface generated by `#[derive(Block)]`.
///
/// Keep this as a structural bridge: generated code exposes concrete stream
/// fields, static message-port metadata, and message-handler dispatch. Runtime
/// behavior such as port initialization, validation, description collection,
/// and input finish handling belongs in the shared helper functions below and
/// in `WrappedKernel`.
#[doc(hidden)]
pub trait KernelInterface {
    /// Whether this block should run in a separate thread instead of on
    /// a normal scheduler worker.
    fn is_blocking() -> bool;
    /// Static block type name.
    fn type_name() -> &'static str;
    /// Access one type-erased stream input and its public name by dense index.
    fn stream_input_at(&mut self, index: PortIndex)
    -> Option<(PortName, &mut dyn DynBufferReader)>;
    /// Access one type-erased stream output and its public name by dense index.
    fn stream_output_at(
        &mut self,
        index: PortIndex,
    ) -> Option<(PortName, &mut dyn DynBufferWriter)>;
    /// Notify adjacent stream peers that this block is done.
    ///
    /// This stays generated because concrete buffer `notify_finished` methods
    /// return `impl Future`; keeping the concrete calls here avoids boxing an
    /// object-safe notification path into every buffer implementation.
    fn stream_ports_notify_finished(&mut self) -> impl Future<Output = ()>;

    /// Message input port names.
    fn message_inputs() -> &'static [&'static str];
    /// Message output port names.
    fn message_outputs() -> &'static [&'static str];
    /// Resolve a message input name to its dense per-block index.
    fn message_input_id(name: impl Into<PortName>) -> Option<PortIndex> {
        let name = name.into();
        Self::message_inputs()
            .iter()
            .position(|candidate| *candidate == name.as_str())
            .map(PortIndex::new)
    }
    /// Call a message handler.
    fn call_handler(
        &mut self,
        _io: &mut WorkIo,
        _mo: &mut MessageOutputs,
        _meta: &BlockMeta,
        id: PortIndex,
        _p: Pmt,
    ) -> impl Future<Output = Result<Pmt, Error>>;
}

pub(crate) fn stream_input_names<K: KernelInterface>(kernel: &mut K) -> Vec<String> {
    let mut names = Vec::new();
    let mut index = 0;
    while let Some((name, _)) = kernel.stream_input_at(PortIndex::new(index)) {
        names.push(name.into_string());
        index += 1;
    }
    names
}

pub(crate) fn stream_output_names<K: KernelInterface>(kernel: &mut K) -> Vec<String> {
    let mut names = Vec::new();
    let mut index = 0;
    while let Some((name, _)) = kernel.stream_output_at(PortIndex::new(index)) {
        names.push(name.into_string());
        index += 1;
    }
    names
}

pub(crate) fn stream_ports_init<K: KernelInterface>(
    kernel: &mut K,
    block_id: BlockId,
    inboxes: PortInboxes,
) {
    let mut index = 0;
    while let Some((_, port)) = kernel.stream_input_at(PortIndex::new(index)) {
        port.init_from(block_id, PortIndex::new(index), &inboxes);
        index += 1;
    }
    let mut index = 0;
    while let Some((_, port)) = kernel.stream_output_at(PortIndex::new(index)) {
        port.init_from(block_id, PortIndex::new(index), &inboxes);
        index += 1;
    }
}

pub(crate) fn stream_ports_validate<K: KernelInterface>(kernel: &mut K) -> Result<(), Error> {
    let mut index = 0;
    while let Some((_, port)) = kernel.stream_input_at(PortIndex::new(index)) {
        port.validate()?;
        index += 1;
    }
    let mut index = 0;
    while let Some((_, port)) = kernel.stream_output_at(PortIndex::new(index)) {
        port.validate()?;
        index += 1;
    }
    Ok(())
}

pub(crate) fn stream_input_finish<K: KernelInterface>(
    kernel: &mut K,
    block_id: BlockId,
    index: PortIndex,
) -> Result<(), Error> {
    let (_, port) = kernel
        .stream_input_at(index)
        .ok_or_else(|| Error::InvalidStreamPort(BlockPortCtx::Id(block_id), PortId::from(index)))?;
    port.finish();
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    struct MissingPorts;

    impl KernelInterface for MissingPorts {
        fn is_blocking() -> bool {
            false
        }

        fn type_name() -> &'static str {
            "MissingPorts"
        }

        fn stream_input_at(
            &mut self,
            _index: PortIndex,
        ) -> Option<(PortName, &mut dyn DynBufferReader)> {
            None
        }

        fn stream_output_at(
            &mut self,
            _index: PortIndex,
        ) -> Option<(PortName, &mut dyn DynBufferWriter)> {
            None
        }

        async fn stream_ports_notify_finished(&mut self) {}

        fn message_inputs() -> &'static [&'static str] {
            &[]
        }

        fn message_outputs() -> &'static [&'static str] {
            &[]
        }

        async fn call_handler(
            &mut self,
            _io: &mut WorkIo,
            _mo: &mut MessageOutputs,
            _meta: &BlockMeta,
            id: PortIndex,
            _p: Pmt,
        ) -> Result<Pmt, Error> {
            Err(Error::InvalidMessagePort(
                BlockPortCtx::None,
                PortId::from(id),
            ))
        }
    }

    #[test]
    fn stream_input_finish_invalid_port_reports_block_id() {
        let mut kernel = MissingPorts;
        let result = stream_input_finish(&mut kernel, BlockId(7), PortIndex::new(3));

        assert!(matches!(
            result,
            Err(Error::InvalidStreamPort(ctx, port))
                if ctx == BlockPortCtx::Id(BlockId(7)) && port == PortId::index(3)
        ));
    }
}