mecha10_core/

rpc.rs

//! Request/Response (RPC) Pattern
//!
//! Provides request/response helpers on top of the pub/sub messaging system.
//! Enables command-response interactions where a node sends a request and waits
//! for a response with a timeout.
//!
//! # Architecture
//!
//! - Uses correlation IDs to match requests with responses
//! - Request and response share the same base topic
//! - Request topic: `{base_topic}/request`
//! - Response topic: `{base_topic}/response`
//! - Each request gets a unique UUID correlation ID
//! - Responder includes the correlation ID in the response
//! - Requester filters responses by correlation ID
//! - Errors from handlers are automatically sent as error responses
//! - Requesters receive errors as `Mecha10Error::MessagingError`
//!
//! # Example
//!
//! ```rust
//! use mecha10::prelude::*;
//! use mecha10::rpc::RpcExt;
//! use serde::{Serialize, Deserialize};
//!
//! #[derive(Debug, Serialize, Deserialize, Clone)]
//! struct Command {
//!     action: String,
//!     params: Vec<f32>,
//! }
//!
//! #[derive(Debug, Serialize, Deserialize, Clone)]
//! struct CommandResult {
//!     success: bool,
//!     message: String,
//! }
//!
//! // Request side
//! # async fn request_example(ctx: &Context) -> Result<()> {
//! let cmd = Command {
//!     action: "move".to_string(),
//!     params: vec![1.0, 2.0],
//! };
//!
//! let result: CommandResult = ctx
//!     .request("/motor/control", &cmd)
//!     .timeout(Duration::from_secs(5))
//!     .await?;
//!
//! println!("Result: {} - {}", result.success, result.message);
//! # Ok(())
//! # }
//!
//! // Response side
//! # async fn response_example(ctx: &Context) -> Result<()> {
//! ctx.respond("/motor/control", |cmd: Command| async move {
//!     // Process command
//!     execute_command(&cmd).await?;
//!
//!     Ok(CommandResult {
//!         success: true,
//!         message: "Command executed".to_string(),
//!     })
//! }).await?;
//! # Ok(())
//! # }
//! ```

use crate::context::{Context, Receiver};
use crate::error::{Mecha10Error, Result};
use crate::messages::Message;
use serde::{de::DeserializeOwned, Deserialize, Serialize};
use std::future::Future;
use std::time::Duration;
use uuid::Uuid;

/// Request envelope with correlation ID
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Request<T> {
    /// Unique correlation ID for matching request/response
    pub correlation_id: String,

    /// Request payload
    pub payload: T,
}

impl<T: Message> Message for Request<T> {}

/// Response envelope with correlation ID
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Response<T> {
    /// Correlation ID from the original request
    pub correlation_id: String,

    /// Response payload (success or error)
    pub payload: RpcResult<T>,
}

impl<T: Message> Message for Response<T> {}

/// RPC result that can represent success or error
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum RpcResult<T> {
    /// Successful response
    Ok(T),
    /// Error response
    Err(RpcError),
}

/// RPC error details
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RpcError {
    /// Error message
    pub message: String,
    /// Error code (optional)
    pub code: Option<String>,
}

/// Request builder for configuring timeouts
pub struct RequestBuilder<'a, Req, Resp> {
    ctx: &'a Context,
    topic: String,
    request: Request<Req>,
    timeout: Duration,
    _phantom: std::marker::PhantomData<Resp>,
}

impl<'a, Req, Resp> RequestBuilder<'a, Req, Resp>
where
    Req: Message + Serialize + Clone,
    Resp: Message + DeserializeOwned + Send + 'static,
{
    /// Set the timeout for the request
    pub fn timeout(mut self, duration: Duration) -> Self {
        self.timeout = duration;
        self
    }

    /// Execute the request and wait for response
    pub async fn execute(self) -> Result<Resp> {
        let request_topic = format!("{}/request", self.topic);
        let response_topic = format!("{}/response", self.topic);
        let correlation_id = self.request.correlation_id.clone();

        // Subscribe to response topic BEFORE publishing request (to avoid race condition)
        let mut responses = self.ctx.subscribe_raw::<Response<Resp>>(&response_topic).await?;

        // Publish the request
        self.ctx.publish_raw(&request_topic, &self.request).await?;

        // Wait for response with timeout
        let deadline = tokio::time::Instant::now() + self.timeout;

        loop {
            match tokio::time::timeout_at(deadline, responses.recv()).await {
                Ok(Some(response)) => {
                    // Check if correlation ID matches
                    if response.correlation_id == correlation_id {
                        // Unwrap RpcResult
                        match response.payload {
                            RpcResult::Ok(payload) => return Ok(payload),
                            RpcResult::Err(err) => {
                                return Err(Mecha10Error::MessagingError {
                                    message: format!("RPC error: {}", err.message),
                                    suggestion: "Check the handler implementation for the error cause".to_string(),
                                });
                            }
                        }
                    }
                    // Wrong correlation ID, keep waiting
                    continue;
                }
                Ok(None) => {
                    return Err(Mecha10Error::MessagingError {
                        message: "Response channel closed".to_string(),
                        suggestion: "Check if responder node is running".to_string(),
                    });
                }
                Err(_) => {
                    return Err(Mecha10Error::MessagingError {
                        message: format!("Request to {} timed out after {:?}", self.topic, self.timeout),
                        suggestion: "Check if responder node is running and responsive".to_string(),
                    });
                }
            }
        }
    }
}

/// RPC extension trait for Context
pub trait RpcExt {
    /// Send a request and wait for response
    ///
    /// # Arguments
    ///
    /// * `topic` - Base topic for the RPC (will use `{topic}/request` and `{topic}/response`)
    /// * `request` - Request payload
    ///
    /// # Returns
    ///
    /// A `RequestBuilder` that can be configured with `.timeout()` and executed with `.await`
    ///
    /// # Example
    ///
    /// ```rust
    /// use mecha10::prelude::*;
    /// use mecha10::rpc::RpcExt;
    ///
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// let result: Response = ctx
    ///     .request("/motor/control", &command)
    ///     .timeout(Duration::from_secs(5))
    ///     .await?;
    /// # Ok(())
    /// # }
    /// ```
    fn request<Req, Resp>(&self, topic: &str, request: &Req) -> RequestBuilder<'_, Req, Resp>
    where
        Req: Message + Serialize + Clone,
        Resp: Message + DeserializeOwned + Send + 'static;

    /// Respond to requests on a topic
    ///
    /// # Arguments
    ///
    /// * `topic` - Base topic for the RPC
    /// * `handler` - Async function that processes requests and returns responses
    ///
    /// # Example
    ///
    /// ```rust
    /// use mecha10::prelude::*;
    /// use mecha10::rpc::RpcExt;
    ///
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// ctx.respond("/motor/control", |cmd: Command| async move {
    ///     execute_command(&cmd).await?;
    ///     Ok(CommandResult { success: true })
    /// }).await?;
    /// # Ok(())
    /// # }
    /// ```
    fn respond<Req, Resp, F, Fut>(&self, topic: &str, handler: F) -> impl Future<Output = Result<()>>
    where
        Req: Message + DeserializeOwned + Send + 'static,
        Resp: Message + Serialize + Send + 'static,
        F: Fn(Req) -> Fut + Send + 'static,
        Fut: Future<Output = Result<Resp>> + Send;

    /// Low-level: Subscribe to a topic without type-safe wrapper
    fn subscribe_raw<T>(&self, topic: &str) -> impl Future<Output = Result<Receiver<T>>>
    where
        T: Message + DeserializeOwned + Send + 'static;

    /// Low-level: Publish to a topic without type-safe wrapper
    fn publish_raw<T>(&self, topic: &str, message: &T) -> impl Future<Output = Result<()>>
    where
        T: Message + Serialize;
}

impl RpcExt for Context {
    fn request<Req, Resp>(&self, topic: &str, request: &Req) -> RequestBuilder<'_, Req, Resp>
    where
        Req: Message + Serialize + Clone,
        Resp: Message + DeserializeOwned + Send + 'static,
    {
        RequestBuilder {
            ctx: self,
            topic: topic.to_string(),
            request: Request {
                correlation_id: Uuid::new_v4().to_string(),
                payload: request.clone(),
            },
            timeout: Duration::from_secs(10), // Default 10 second timeout
            _phantom: std::marker::PhantomData,
        }
    }

    async fn respond<Req, Resp, F, Fut>(&self, topic: &str, handler: F) -> Result<()>
    where
        Req: Message + DeserializeOwned + Send + 'static,
        Resp: Message + Serialize + Send + 'static,
        F: Fn(Req) -> Fut + Send + 'static,
        Fut: Future<Output = Result<Resp>> + Send,
    {
        let request_topic = format!("{}/request", topic);
        let response_topic = format!("{}/response", topic);

        // Subscribe to requests
        let mut requests = self.subscribe_raw::<Request<Req>>(&request_topic).await?;

        // Spawn task to handle requests
        let ctx = self.clone();
        tokio::spawn(async move {
            while let Some(request) = requests.recv().await {
                let correlation_id = request.correlation_id.clone();

                // Call handler and wrap result
                let result = match handler(request.payload).await {
                    Ok(response_payload) => RpcResult::Ok(response_payload),
                    Err(e) => {
                        tracing::error!(
                            correlation_id = %correlation_id,
                            error = %e,
                            "RPC handler failed"
                        );
                        RpcResult::Err(RpcError {
                            message: e.to_string(),
                            code: None,
                        })
                    }
                };

                let response = Response {
                    correlation_id: correlation_id.clone(),
                    payload: result,
                };

                // Send response (either success or error)
                if let Err(e) = ctx.publish_raw(&response_topic, &response).await {
                    tracing::error!(
                        correlation_id = %correlation_id,
                        topic = %response_topic,
                        error = %e,
                        "Failed to publish RPC response"
                    );
                }
            }
        });

        Ok(())
    }

    async fn subscribe_raw<T>(&self, topic: &str) -> Result<Receiver<T>>
    where
        T: Message + DeserializeOwned + Send + 'static,
    {
        Context::subscribe_raw(self, topic).await
    }

    async fn publish_raw<T>(&self, topic: &str, message: &T) -> Result<()>
    where
        T: Message + Serialize,
    {
        Context::publish_raw(self, topic, message).await
    }
}