action_core/

client_server.rs

use std::collections::VecDeque;

use core_types::{ActionGoalId, RtError, Timestamp};

use crate::message::{
    ActionFeedback, ActionResult, ActionSchema, ActionSessionHealth, GoalAck, GoalStatus,
};

mod client;
mod server;

pub use self::client::BasicActionClient;
pub use self::server::BasicActionServer;

// ─── Traits ───────────────────────────────────────────────────────────────────

/// Client-side handle: sends goals, polls feedback, and retrieves the final result.
///
/// # Comparison with Service and Mission
/// | Feature        | Service       | Action                        | Mission             |
/// |----------------|---------------|-------------------------------|---------------------|
/// | Direction      | req → res     | goal → feedback* → result     | long-lived duplex   |
/// | Streaming      | ✗             | ✓ (best-effort)               | ✓ (reliable)        |
/// | Cancel         | ✗             | ✓                             | ✓ (Pause/Cancel)    |
/// | Reconnect      | ✗             | ✗                             | ✓ (Reconcile)       |
/// | State-machine  | none          | simple (6 states)             | complex (8 states)  |
pub trait ActionClient<G, F, R>
where
    G: Send + 'static,
    F: Send + 'static,
    R: Send + 'static,
{
    fn action_name(&self) -> &str;
    fn schema(&self) -> &ActionSchema;

    /// Send a goal and immediately receive the server's accept/reject acknowledgement.
    ///
    /// The returned `ActionGoalId` is used for subsequent `poll_feedback`,
    /// `poll_result`, and `cancel` calls. If the server rejects the goal
    /// (`GoalAck::accepted == false`), the ID remains valid but no feedback
    /// or result will ever be produced for it.
    fn send_goal(&mut self, goal: G) -> Result<(ActionGoalId, GoalAck), RtError>;

    /// Non-blocking poll: dequeue the next feedback item for the given goal, or `None` if empty.
    fn poll_feedback(&mut self, goal_id: ActionGoalId) -> Option<ActionFeedback<F>>;

    /// Non-blocking poll: take the final result for the given goal, or `None` if not ready.
    ///
    /// Once `Some` is returned the goal's lifecycle is over; the ID need not be tracked further.
    fn poll_result(&mut self, goal_id: ActionGoalId) -> Option<ActionResult<R>>;

    /// Request cancellation of a goal.
    ///
    /// Cancellation is a *request*, not a command; the server may refuse (e.g. the goal
    /// has entered a critical phase). On success, `poll_result` will eventually return
    /// `GoalStatus::Canceled`.
    fn cancel(&mut self, goal_id: ActionGoalId) -> Result<(), RtError>;

    /// Query the current status of a goal without consuming any feedback or result.
    fn goal_status(&self, goal_id: ActionGoalId) -> Option<GoalStatus>;

    /// Evaluate heartbeat timeouts for executing goals.
    ///
    /// Returns the number of goals that were transitioned into timeout failure.
    fn tick_timeouts(&mut self, _now: Timestamp) -> usize {
        0
    }

    /// Return health state of a specific goal.
    fn goal_health(&self, _goal_id: ActionGoalId) -> Option<ActionSessionHealth> {
        None
    }

    /// Return health states for all known goals.
    fn active_goal_health(&self) -> Vec<ActionSessionHealth> {
        Vec::new()
    }

    fn close(&mut self) -> Result<(), RtError>;
}

/// Server-side handle: receives goals, publishes feedback, and sends the final result.
pub trait ActionServer<G, F, R>
where
    G: Send + 'static,
    F: Send + 'static,
    R: Send + 'static,
{
    fn action_name(&self) -> &str;
    fn schema(&self) -> &ActionSchema;

    /// Receive the next pending goal (FIFO).
    ///
    /// After returning `Some`, the server **must** call either `accept_goal` or
    /// `reject_goal`; otherwise the client will wait indefinitely for a `GoalAck`.
    fn recv_goal(&mut self) -> Result<Option<(ActionGoalId, G)>, RtError>;

    /// Accept a goal; sends `GoalAck { accepted: true }` to the client.
    fn accept_goal(&mut self, goal_id: ActionGoalId) -> Result<(), RtError>;

    /// Reject a goal; sends `GoalAck { accepted: false, reason }` to the client.
    fn reject_goal(
        &mut self,
        goal_id: ActionGoalId,
        reason: impl Into<String>,
    ) -> Result<(), RtError>;

    /// Push a feedback update to the client (may be called multiple times).
    fn publish_feedback(&mut self, goal_id: ActionGoalId, feedback: F) -> Result<(), RtError>;

    /// Publish a keepalive heartbeat for a running goal.
    fn heartbeat(&mut self, goal_id: ActionGoalId) -> Result<(), RtError>;

    /// Mark the goal as succeeded and deliver the final result.
    fn succeed(&mut self, goal_id: ActionGoalId, result: R) -> Result<(), RtError>;

    /// Mark the goal as failed and deliver the final result with an error reason.
    fn fail(&mut self, goal_id: ActionGoalId, reason: impl Into<String>) -> Result<(), RtError>;

    /// Poll for a pending cancel request from the client.
    fn poll_cancel_request(&mut self) -> Option<ActionGoalId>;

    /// Confirm that cancellation is complete; delivers a `GoalStatus::Canceled` result.
    fn confirm_cancel(&mut self, goal_id: ActionGoalId) -> Result<(), RtError>;

    fn close(&mut self) -> Result<(), RtError>;
}

// ─── Shared in-process channel ────────────────────────────────────────────────

/// In-memory channel shared between a client and a server (same-process semantics).
///
/// Analogous to `middleware_core::ServiceChannel`: both sides hold an `Arc` to
/// the same instance; individual queues are protected by `Mutex`.
pub struct ActionChannel<G, F, R> {
    /// Client → Server: pending goal queue.
    pub goals: std::sync::Mutex<VecDeque<(ActionGoalId, G)>>,
    /// Server → Client: per-goal acknowledgement.
    pub acks: std::sync::Mutex<std::collections::HashMap<u64, GoalAck>>,
    /// Server → Client: feedback queues keyed by goal id.
    pub feedbacks: std::sync::Mutex<std::collections::HashMap<u64, VecDeque<ActionFeedback<F>>>>,
    /// Server → Client: per-goal final result.
    pub results: std::sync::Mutex<std::collections::HashMap<u64, ActionResult<R>>>,
    /// Client → Server: cancel request queue.
    pub cancels: std::sync::Mutex<VecDeque<ActionGoalId>>,
    /// Current status per goal; readable by both client and server.
    pub statuses: std::sync::Mutex<std::collections::HashMap<u64, GoalStatus>>,
    /// Last heartbeat timestamp per active goal.
    pub heartbeats: std::sync::Mutex<std::collections::HashMap<u64, Timestamp>>,
    /// Last feedback timestamp per goal.
    pub feedback_timestamps: std::sync::Mutex<std::collections::HashMap<u64, Timestamp>>,
    /// Last terminal result timestamp per goal.
    pub result_timestamps: std::sync::Mutex<std::collections::HashMap<u64, Timestamp>>,
}

impl<G, F, R> ActionChannel<G, F, R> {
    pub fn new() -> std::sync::Arc<Self> {
        std::sync::Arc::new(Self {
            goals: Default::default(),
            acks: Default::default(),
            feedbacks: Default::default(),
            results: Default::default(),
            cancels: Default::default(),
            statuses: Default::default(),
            heartbeats: Default::default(),
            feedback_timestamps: Default::default(),
            result_timestamps: Default::default(),
        })
    }
}