mecha10-core 4.2.10

//! Shared "subscribe on one topic -> republish unchanged on another" pattern.
//!
//! Several nodes need the same shape of logic: subscribe to a command/input topic, and on every
//! received message, republish that exact value (no transformation) on a separate output topic
//! of the same message type. The motivating case is cheap hobby servos with no real position
//! sensor: a `pan-tilt`/`servo`-style node subscribes `/actuator/joint_cmd` and echoes the
//! commanded joint positions straight back out as `/sensor/joint_states`, because "echo the
//! command" *is* the sensor reading. A simulator node doing the same `joint_cmd -> joint_states`
//! echo is another consumer of the identical shape.
//!
//! Before this module, each node hand-rolled this: its own subscribe call, its own loop,
//! its own re-publish on receipt - and neither of the first two implementations published
//! anything on the output topic until the first command arrived, which breaks downstream
//! consumers that treat "no message ever received" on that topic as a hard failure at startup.
//! [`forward_echo`] and [`echo_topic`] factor out both the forwarding loop and an optional
//! immediate seed/initial value published before the first real message.
//!
//! What stays node-specific, deliberately:
//! - **The subscribe/publish topics and message type.** This module is generic over any `T`
//!   that already satisfies [`Context::subscribe`]/[`Context::publish_to`]'s bounds - it has no
//!   opinion on what's being echoed.
//! - **Anything the node needs to *do* with a received command besides echoing it** (e.g.
//!   actually moving a servo). [`echo_topic`] hands back every forwarded message on the returned
//!   [`Receiver`], so a node that also needs to act on the command can do so from its own
//!   `tokio::select!` loop - the republish has already happened by the time the message is handed
//!   back, so the node can't skip it or get it out of order with its own handling.
//!
//! # Example
//!
//! Requires the `messaging` feature (`echo_topic` is only available when
//! it's enabled), so this example is `ignore`d rather than compiled/run as
//! a doctest when that feature is off.
//!
//! ```rust,ignore
//! use mecha10_core::echo::echo_topic;
//! use mecha10_core::prelude::*;
//! use mecha10_core::topics::Topic;
//!
//! #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
//! struct JointPositions {
//!     positions: Vec<f32>,
//! }
//!
//! impl Message for JointPositions {}
//!
//! const JOINT_CMD: Topic<JointPositions> = Topic::new("/actuator/joint_cmd");
//! const JOINT_STATES: Topic<JointPositions> = Topic::new("/sensor/joint_states");
//!
//! # async fn example(ctx: &Context) -> Result<()> {
//! // Publishes an all-zero seed on `/sensor/joint_states` immediately, then echoes every
//! // `/actuator/joint_cmd` message straight back out on `/sensor/joint_states` from then on.
//! let seed = JointPositions { positions: vec![0.0, 0.0] };
//! let mut joint_cmds = echo_topic(ctx, JOINT_CMD, JOINT_STATES, Some(seed)).await?;
//!
//! while let Some(cmd) = joint_cmds.recv().await {
//!     // The echo onto JOINT_STATES has already happened; this is where a real node would
//!     // additionally act on the command, e.g. drive the servo hardware.
//!     let _ = cmd;
//! }
//! # Ok(())
//! # }
//! ```

use crate::context::Receiver;
use crate::error::Result;
use std::future::Future;

/// Drive a "seed, then forward every received message unchanged" loop.
///
/// This is the broker-free core of [`echo_topic`]: it knows nothing about [`Context`] or
/// pub/sub, only about a [`Receiver<T>`] to read from and a `publish` callback to hand each
/// value to. It exists as its own function so the forwarding behavior - seed goes out before
/// anything else, every subsequent message is forwarded exactly once and in order, nothing is
/// published when there's no seed and no message has arrived yet - can be unit tested without a
/// running message broker.
///
/// If `seed` is `Some`, it is published first, before anything is read from `incoming`. Then
/// every message received from `incoming` is published in turn, unchanged, until `incoming`
/// closes (all senders dropped), at which point this returns `Ok(())`.
///
/// Returns early with an error the first time `publish` does, without draining the rest of
/// `incoming`.
///
/// [`Context`]: crate::context::Context
pub async fn forward_echo<T, P, Fut>(mut incoming: Receiver<T>, seed: Option<T>, mut publish: P) -> Result<()>
where
    P: FnMut(T) -> Fut,
    Fut: Future<Output = Result<()>>,
{
    if let Some(seed) = seed {
        publish(seed).await?;
    }

    while let Some(message) = incoming.recv().await {
        publish(message).await?;
    }

    Ok(())
}

#[cfg(feature = "messaging")]
mod subscribe {
    use super::forward_echo;
    use crate::context::{Context, Receiver};
    use crate::error::Result;
    use crate::messages::Message;
    use crate::topics::Topic;
    use serde::de::DeserializeOwned;
    use serde::Serialize;

    /// Subscribe to `subscribe_topic` and echo every received message, unchanged, as a publish
    /// on `publish_topic`.
    ///
    /// If `seed` is `Some`, it is published on `publish_topic` immediately, before any message
    /// has been received on `subscribe_topic` - so downstream consumers that treat "no message
    /// ever published" on `publish_topic` as a hard failure see a value right away instead of
    /// waiting for the first real command.
    ///
    /// Returns a [`Receiver<T>`] that yields every message *after* it has already been
    /// re-published on `publish_topic` (the seed, if any, is published but not re-yielded here -
    /// it never arrived as a subscribed message). Forward this into your node's own
    /// `tokio::select!` loop if it needs to react to the same commands beyond echoing them (e.g.
    /// to actually drive hardware); drop it immediately if the echo is the entire behavior you
    /// need.
    ///
    /// Publish failures are logged and stop the background forwarding task; they do not panic
    /// and do not propagate to the returned `Receiver` (it simply stops yielding further
    /// messages, the same as if the subscription had ended).
    pub async fn echo_topic<T>(
        ctx: &Context,
        subscribe_topic: Topic<T>,
        publish_topic: Topic<T>,
        seed: Option<T>,
    ) -> Result<Receiver<T>>
    where
        T: Message + DeserializeOwned + Serialize + Send + Clone + 'static,
    {
        let incoming = ctx.subscribe(subscribe_topic).await?;
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<T>();
        let ctx = ctx.clone();

        tokio::spawn(async move {
            let result = forward_echo(incoming, seed, move |message: T| {
                let ctx = ctx.clone();
                let tx = tx.clone();
                async move {
                    ctx.publish_to(publish_topic, &message).await?;
                    // The receiving half may already be gone (caller dropped the handle because
                    // it only wanted the echo side effect); that's not an error for us.
                    let _ = tx.send(message);
                    Ok(())
                }
            })
            .await;

            if let Err(error) = result {
                tracing::warn!(
                    subscribe_topic = subscribe_topic.path(),
                    publish_topic = publish_topic.path(),
                    %error,
                    "echo_topic forwarding task stopped due to a publish error"
                );
            }
        });

        Ok(Receiver::from(rx))
    }
}

#[cfg(feature = "messaging")]
pub use subscribe::echo_topic;

#[cfg(test)]
mod tests {
    use super::*;
    use crate::error::Mecha10Error;
    use std::sync::{Arc, Mutex};

    #[derive(Debug, Clone, PartialEq, Eq)]
    struct Reading(i32);

    fn channel<T>() -> (tokio::sync::mpsc::UnboundedSender<T>, Receiver<T>) {
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
        (tx, Receiver::from(rx))
    }

    #[tokio::test]
    async fn forwards_a_single_received_message_unchanged() {
        let (tx, rx) = channel::<Reading>();
        let published = Arc::new(Mutex::new(Vec::new()));
        let published_clone = Arc::clone(&published);

        tx.send(Reading(42)).unwrap();
        drop(tx);

        forward_echo(rx, None, move |message: Reading| {
            let published = Arc::clone(&published_clone);
            async move {
                published.lock().unwrap().push(message);
                Ok(())
            }
        })
        .await
        .unwrap();

        assert_eq!(*published.lock().unwrap(), vec![Reading(42)]);
    }

    #[tokio::test]
    async fn forwards_multiple_messages_in_order() {
        let (tx, rx) = channel::<Reading>();
        let published = Arc::new(Mutex::new(Vec::new()));
        let published_clone = Arc::clone(&published);

        for value in 0..5 {
            tx.send(Reading(value)).unwrap();
        }
        drop(tx);

        forward_echo(rx, None, move |message: Reading| {
            let published = Arc::clone(&published_clone);
            async move {
                published.lock().unwrap().push(message);
                Ok(())
            }
        })
        .await
        .unwrap();

        let expected: Vec<Reading> = (0..5).map(Reading).collect();
        assert_eq!(*published.lock().unwrap(), expected);
    }

    #[tokio::test]
    async fn publishes_seed_before_any_real_message() {
        let (tx, rx) = channel::<Reading>();
        let published = Arc::new(Mutex::new(Vec::new()));
        let published_clone = Arc::clone(&published);

        tx.send(Reading(1)).unwrap();
        drop(tx);

        forward_echo(rx, Some(Reading(-1)), move |message: Reading| {
            let published = Arc::clone(&published_clone);
            async move {
                published.lock().unwrap().push(message);
                Ok(())
            }
        })
        .await
        .unwrap();

        assert_eq!(*published.lock().unwrap(), vec![Reading(-1), Reading(1)]);
    }

    #[tokio::test]
    async fn publishes_nothing_extra_when_no_seed_and_no_messages() {
        let (tx, rx) = channel::<Reading>();
        let published = Arc::new(Mutex::new(Vec::new()));
        let published_clone = Arc::clone(&published);

        drop(tx);

        forward_echo(rx, None, move |message: Reading| {
            let published = Arc::clone(&published_clone);
            async move {
                published.lock().unwrap().push(message);
                Ok(())
            }
        })
        .await
        .unwrap();

        assert!(published.lock().unwrap().is_empty());
    }

    #[tokio::test]
    async fn works_generically_over_an_arbitrary_message_type() {
        #[derive(Debug, Clone, PartialEq)]
        struct Pose {
            x: f32,
            y: f32,
        }

        let (tx, rx) = channel::<Pose>();
        let published = Arc::new(Mutex::new(Vec::new()));
        let published_clone = Arc::clone(&published);

        tx.send(Pose { x: 1.0, y: 2.0 }).unwrap();
        drop(tx);

        forward_echo(rx, Some(Pose { x: 0.0, y: 0.0 }), move |message: Pose| {
            let published = Arc::clone(&published_clone);
            async move {
                published.lock().unwrap().push(message);
                Ok(())
            }
        })
        .await
        .unwrap();

        assert_eq!(
            *published.lock().unwrap(),
            vec![Pose { x: 0.0, y: 0.0 }, Pose { x: 1.0, y: 2.0 }]
        );
    }

    #[tokio::test]
    async fn stops_and_returns_the_error_on_the_first_publish_failure() {
        let (tx, rx) = channel::<Reading>();
        let published = Arc::new(Mutex::new(Vec::new()));
        let published_clone = Arc::clone(&published);

        for value in 0..3 {
            tx.send(Reading(value)).unwrap();
        }
        drop(tx);

        let result = forward_echo(rx, None, move |message: Reading| {
            let published = Arc::clone(&published_clone);
            async move {
                if message == Reading(1) {
                    return Err(Mecha10Error::MessagingError {
                        message: "boom".to_string(),
                        suggestion: "n/a".to_string(),
                    });
                }
                published.lock().unwrap().push(message);
                Ok(())
            }
        })
        .await;

        assert!(result.is_err());
        // Only the message before the failing one made it through; the failing one and
        // everything after it (message 2, still buffered in the channel) were not published.
        assert_eq!(*published.lock().unwrap(), vec![Reading(0)]);
    }
}