mecha10_behavior_runtime/actions/
set_rl_target.rs

1//! Action node that relays live target-class requests to the `rl` node's control topic.
2
3use crate::{BehaviorNode, Context, NodeStatus};
4use async_trait::async_trait;
5use mecha10_core::messages::Message;
6use mecha10_core::topics::Topic;
7use mecha10_core::Receiver;
8use serde::{Deserialize, Serialize};
9use serde_json::Value;
10
11#[derive(Debug, Clone, Serialize, Deserialize)]
12struct RlControlCommand {
13    action: String,
14    #[serde(skip_serializing_if = "Option::is_none")]
15    params: Option<Value>,
16}
17
18impl Message for RlControlCommand {}
19
20#[derive(Debug, Clone, Serialize, Deserialize)]
21struct RlTargetRequest {
22    #[serde(default)]
23    target_class: Option<String>,
24}
25
26impl Message for RlTargetRequest {}
27
28fn default_topic() -> String {
29    "/rl/control".to_string()
30}
31
32fn default_request_topic() -> String {
33    "/rl/target/request".to_string()
34}
35
36pub struct SetRlTargetNode {
37    topic: String,
38    request_topic: String,
39    target: Option<String>,
40    receiver: Option<Receiver<Value>>,
41    published_initial: bool,
42}
43
44impl std::fmt::Debug for SetRlTargetNode {
45    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
46        f.debug_struct("SetRlTargetNode")
47            .field("topic", &self.topic)
48            .field("request_topic", &self.request_topic)
49            .field("target", &self.target)
50            .field("subscribed", &self.receiver.is_some())
51            .field("published_initial", &self.published_initial)
52            .finish()
53    }
54}
55
56#[derive(Debug, Serialize, Deserialize)]
57struct SetRlTargetConfig {
58    #[serde(default = "default_topic")]
59    topic: String,
60    #[serde(default = "default_request_topic")]
61    request_topic: String,
62    #[serde(default)]
63    target: Option<String>,
64}
65
66impl SetRlTargetNode {
67    /// Create a new SetRlTargetNode from JSON configuration.
68    ///
69    /// Normalizes a `null` config to `{}` so per-field defaults can apply.
70    pub fn from_json(config: Value) -> anyhow::Result<Self> {
71        let config = if config.is_null() {
72            serde_json::json!({})
73        } else {
74            config
75        };
76        let cfg: SetRlTargetConfig = serde_json::from_value(config)?;
77        Ok(Self {
78            topic: cfg.topic,
79            request_topic: cfg.request_topic,
80            target: cfg.target,
81            receiver: None,
82            published_initial: false,
83        })
84    }
85
86    async fn publish(&self, ctx: &Context, target: Option<String>) -> anyhow::Result<()> {
87        let command = RlControlCommand {
88            action: "set_target_class".to_string(),
89            params: Some(serde_json::json!({ "target_class": target })),
90        };
91
92        let topic_static: &'static str = Box::leak(self.topic.clone().into_boxed_str());
93        let topic = Topic::<RlControlCommand>::new(topic_static);
94
95        ctx.publish_to(topic, &command).await?;
96        tracing::debug!("SetRlTargetNode: published set_target_class({:?})", target);
97        Ok(())
98    }
99
100    /// Drains every pending request off `receiver`, returning the last one seen (if any) - only
101    /// the most recent request matters, same as `RlTaskNode::poll_complete`'s "arrival is the
102    /// whole signal" approach to draining a channel once per tick.
103    fn poll_latest_request(receiver: &mut Receiver<Value>) -> Option<Option<String>> {
104        let mut latest = None;
105        while let Ok(value) = receiver.try_recv() {
106            match serde_json::from_value::<RlTargetRequest>(value) {
107                Ok(request) => latest = Some(request.target_class),
108                Err(e) => tracing::warn!("SetRlTargetNode: dropping malformed target request: {}", e),
109            }
110        }
111        latest
112    }
113}
114
115#[async_trait]
116impl BehaviorNode for SetRlTargetNode {
117    async fn tick(&mut self, ctx: &Context) -> anyhow::Result<NodeStatus> {
118        if self.receiver.is_none() {
119            self.receiver = Some(ctx.subscribe_raw::<Value>(&self.request_topic).await?);
120        }
121
122        if !self.published_initial {
123            self.published_initial = true;
124            if self.target.is_some() {
125                self.publish(ctx, self.target.clone()).await?;
126            }
127        }
128
129        let latest = Self::poll_latest_request(self.receiver.as_mut().expect("just subscribed above"));
130        if let Some(target) = latest {
131            self.publish(ctx, target).await?;
132        }
133
134        Ok(NodeStatus::Running)
135    }
136
137    fn name(&self) -> &str {
138        "SetRlTargetNode"
139    }
140}
141
142#[cfg(test)]
143mod tests {
144    use super::*;
145    use crate::test_helpers::create_test_context;
146
147    #[test]
148    fn from_json_reads_topic_request_topic_and_target() {
149        let node = SetRlTargetNode::from_json(serde_json::json!({
150            "topic": "/rl/control",
151            "request_topic": "/rl/target/request",
152            "target": "cat"
153        }))
154        .unwrap();
155
156        assert_eq!(node.topic, "/rl/control");
157        assert_eq!(node.request_topic, "/rl/target/request");
158        assert_eq!(node.target, Some("cat".to_string()));
159    }
160
161    #[test]
162    fn from_json_defaults_topics_and_allows_no_target() {
163        let node = SetRlTargetNode::from_json(serde_json::json!({})).unwrap();
164
165        assert_eq!(node.topic, "/rl/control");
166        assert_eq!(node.request_topic, "/rl/target/request");
167        assert_eq!(node.target, None);
168    }
169
170    #[test]
171    fn from_json_accepts_null_the_same_as_an_empty_object() {
172        let node = SetRlTargetNode::from_json(Value::Null).unwrap();
173
174        assert_eq!(node.topic, "/rl/control");
175        assert_eq!(node.request_topic, "/rl/target/request");
176        assert_eq!(node.target, None);
177    }
178
179    #[test]
180    fn poll_latest_request_returns_none_when_empty() {
181        let (_tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
182        let mut receiver = Receiver::from(rx);
183        assert_eq!(SetRlTargetNode::poll_latest_request(&mut receiver), None);
184    }
185
186    #[test]
187    fn poll_latest_request_keeps_only_the_most_recent_value() {
188        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
189        let mut receiver = Receiver::from(rx);
190        tx.send(serde_json::json!({ "target_class": "dog" })).unwrap();
191        tx.send(serde_json::json!({ "target_class": "cat" })).unwrap();
192
193        assert_eq!(
194            SetRlTargetNode::poll_latest_request(&mut receiver),
195            Some(Some("cat".to_string()))
196        );
197    }
198
199    #[test]
200    fn poll_latest_request_accepts_a_null_target_class_as_a_clear_request() {
201        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
202        let mut receiver = Receiver::from(rx);
203        tx.send(serde_json::json!({ "target_class": null })).unwrap();
204
205        assert_eq!(SetRlTargetNode::poll_latest_request(&mut receiver), Some(None));
206    }
207
208    #[test]
209    fn poll_latest_request_drops_malformed_requests_and_keeps_polling() {
210        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
211        let mut receiver = Receiver::from(rx);
212        tx.send(serde_json::json!({ "target_class": 42 })).unwrap();
213        tx.send(serde_json::json!({ "target_class": "cat" })).unwrap();
214
215        assert_eq!(
216            SetRlTargetNode::poll_latest_request(&mut receiver),
217            Some(Some("cat".to_string()))
218        );
219    }
220
221    #[tokio::test]
222    #[ignore]
223    async fn tick_publishes_configured_target_once_then_stays_running() {
224        let ctx = create_test_context("test_set_rl_target").await;
225
226        let mut node = SetRlTargetNode::from_json(serde_json::json!({
227            "topic": "/rl/control",
228            "target": "cat"
229        }))
230        .unwrap();
231
232        let mut receiver = ctx.subscribe_raw::<Value>("/rl/control").await.unwrap();
233
234        let status = node.tick(&ctx).await.unwrap();
235        assert_eq!(status, NodeStatus::Running);
236
237        let received = tokio::time::timeout(std::time::Duration::from_millis(500), receiver.recv())
238            .await
239            .expect("timed out waiting for published command")
240            .expect("channel closed without a message");
241
242        assert_eq!(received["action"], "set_target_class");
243        assert_eq!(received["params"]["target_class"], "cat");
244
245        let status = node.tick(&ctx).await.unwrap();
246        assert_eq!(status, NodeStatus::Running);
247        let timed_out = tokio::time::timeout(std::time::Duration::from_millis(200), receiver.recv()).await;
248        assert!(timed_out.is_err(), "expected no second publish without a new request");
249    }
250
251    #[tokio::test]
252    #[ignore]
253    async fn tick_relays_a_live_request_published_on_request_topic() {
254        let ctx = create_test_context("test_set_rl_target_request").await;
255
256        let mut node = SetRlTargetNode::from_json(serde_json::json!({
257            "topic": "/rl/control",
258            "request_topic": "/rl/target/request"
259        }))
260        .unwrap();
261
262        node.tick(&ctx).await.unwrap();
263
264        let mut control_receiver = ctx.subscribe_raw::<Value>("/rl/control").await.unwrap();
265        ctx.publish_raw("/rl/target/request", &serde_json::json!({ "target_class": "dog" }))
266            .await
267            .unwrap();
268
269        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
270        node.tick(&ctx).await.unwrap();
271
272        let received = tokio::time::timeout(std::time::Duration::from_millis(500), control_receiver.recv())
273            .await
274            .expect("timed out waiting for relayed command")
275            .expect("channel closed without a message");
276
277        assert_eq!(received["action"], "set_target_class");
278        assert_eq!(received["params"]["target_class"], "dog");
279    }
280}
mecha10_behavior_runtime/actions/set_rl_target.rs

mecha10_behavior_runtime/actions/
set_rl_target.rs
