nu-zenoh 0.4.0

//
// Copyright (c) 2025 ZettaScale Technology
//
// This program and the accompanying materials are made available under the
// terms of the Eclipse Public License 2.0 which is available at
// http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
// which is available at https://www.apache.org/licenses/LICENSE-2.0.
//
// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
//
// Contributors:
//   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
//
use nu_engine::CallExt;
use nu_protocol::{
    engine::{Call, Command, EngineState, Stack},
    record, IntoValue, ListStream, PipelineData, ShellError, Signature, Type,
};
use zenoh::Wait;

use crate::{
    call_ext2::CallExt2, interruptible_channel::InterruptibleChannel, signature_ext::SignatureExt,
    State,
};

#[derive(Clone)]
pub(crate) struct Pub {
    state: State,
}

impl Pub {
    pub(crate) fn new(state: State) -> Self {
        Self { state }
    }
}

impl Command for Pub {
    fn name(&self) -> &str {
        "zenoh pub"
    }

    fn signature(&self) -> Signature {
        Signature::build(self.name())
            .session()
            .zenoh_category()
            .keyexpr()
            .allowed_destination()
            .congestion_control(self.state.options.experimental_options)
            .reliable()
            .express()
            .priority()
            .encoding()
            .input_output_type(Type::Any, Type::Any)
    }

    fn description(&self) -> &str {
        "Declare a publisher"
    }

    fn run(
        &self,
        engine_state: &EngineState,
        stack: &mut Stack,
        call: &Call,
        input: PipelineData,
    ) -> Result<PipelineData, ShellError> {
        let key = call.req::<String>(engine_state, stack, 0)?;

        let pub_ = self
            .state
            .with_session(&call.session(engine_state, stack)?, |sess| {
                let mut pub_ = sess.declare_publisher(key);

                if let Some(encoding) = call.encoding(engine_state, stack)? {
                    pub_ = pub_.encoding(encoding);
                }

                if let Some(priority) = call.priority(engine_state, stack)? {
                    pub_ = pub_.priority(priority);
                }

                if let Some(congestion_control) = call.congestion_control(
                    engine_state,
                    stack,
                    self.state.options.experimental_options,
                )? {
                    pub_ = pub_.congestion_control(congestion_control);
                }

                if let Some(reliability) = call.reliable(engine_state, stack)? {
                    pub_ = pub_.reliability(reliability);
                }

                if let Some(express) = call.express(engine_state, stack)? {
                    pub_ = pub_.express(express);
                }

                if let Some(destination) = call.allowed_destination(engine_state, stack)? {
                    pub_ = pub_.allowed_destination(destination);
                }

                pub_.wait()
            })?
            .map_err(|e| {
                nu_protocol::LabeledError::new("Declare publisher operation failed")
                    .with_label(format!("Declare publisher failed: {e}"), call.head)
            })?;

        for value in input {
            pub_.put(value.as_str()?).wait().unwrap();
        }

        Ok(nu_protocol::PipelineData::empty())
    }
}

#[derive(Clone)]
pub(crate) struct MatchingListener {
    state: State,
}

impl MatchingListener {
    pub(crate) fn new(state: State) -> Self {
        Self { state }
    }
}

impl Command for MatchingListener {
    fn name(&self) -> &str {
        "zenoh pub matching-listener"
    }

    fn signature(&self) -> Signature {
        Signature::build(self.name())
            .session()
            .zenoh_category()
            .keyexpr()
            .allowed_destination()
            .input_output_type(Type::Nothing, Type::list(Type::record()))
    }

    fn description(&self) -> &str {
        "Returns a stream of matching status updates for a publisher"
    }

    fn run(
        &self,
        engine_state: &EngineState,
        stack: &mut Stack,
        call: &Call,
        _input: PipelineData,
    ) -> Result<PipelineData, ShellError> {
        const MATCHING_CHANNEL_SIZE: usize = 256;
        let (tx, rx) = flume::bounded(MATCHING_CHANNEL_SIZE);

        let span = call.head;
        let key = call.req::<String>(engine_state, stack, 0)?;

        let (pub_, listener) = self
            .state
            .with_session(
                &call.session(engine_state, stack)?,
                move |sess| -> zenoh::Result<_> {
                    let mut pub_ = sess.declare_publisher(key);

                    if let Some(destination) = call.allowed_destination(engine_state, stack)? {
                        pub_ = pub_.allowed_destination(destination);
                    }

                    let pub_ = pub_.wait()?;
                    let listener = pub_
                        .matching_listener()
                        .callback(move |status| {
                            let _ = tx.send(status);
                        })
                        .wait()?;

                    Ok((pub_, listener))
                },
            )?
            .map_err(|e| {
                nu_protocol::LabeledError::new("Failed to declare publisher matching listener")
                    .with_label(
                        format!("Failed to declare publisher matching listener: {e}"),
                        call.head,
                    )
            })?;

        let iter =
            InterruptibleChannel::with_data(rx, engine_state.signals().clone(), (pub_, listener))
                .map(move |status| {
                    record!(
                        "matching" => status.matching().into_value(span),
                    )
                    .into_value(span)
                });

        Ok(ListStream::new(iter, call.head, engine_state.signals().clone()).into())
    }
}