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    /// Every field of `SetRlTargetConfig` is `#[serde(default)]`, so a `config: {}` entry (or no
70    /// `config` field at all, like `rl/config.json`'s own usage) should be valid - but the
71    /// composition loader passes an absent config through as `Value::Null`, not `{}`
72    /// (`loader.rs`'s `load_composition`), and `serde_json::Value`'s `Deserializer` rejects
73    /// `null` for a struct type regardless of per-field defaults ("invalid type: null, expected
74    /// struct SetRlTargetConfig"). Normalize `null` to `{}` here so the per-field defaults can
75    /// actually apply, same as every other builtin action whose config is all-optional.
76    pub fn from_json(config: Value) -> anyhow::Result<Self> {
77        let config = if config.is_null() {
78            serde_json::json!({})
79        } else {
80            config
81        };
82        let cfg: SetRlTargetConfig = serde_json::from_value(config)?;
83        Ok(Self {
84            topic: cfg.topic,
85            request_topic: cfg.request_topic,
86            target: cfg.target,
87            receiver: None,
88            published_initial: false,
89        })
90    }
91
92    async fn publish(&self, ctx: &Context, target: Option<String>) -> anyhow::Result<()> {
93        let command = RlControlCommand {
94            action: "set_target_class".to_string(),
95            params: Some(serde_json::json!({ "target_class": target })),
96        };
97
98        let topic_static: &'static str = Box::leak(self.topic.clone().into_boxed_str());
99        let topic = Topic::<RlControlCommand>::new(topic_static);
100
101        ctx.publish_to(topic, &command).await?;
102        tracing::debug!("SetRlTargetNode: published set_target_class({:?})", target);
103        Ok(())
104    }
105
106    /// Drains every pending request off `receiver`, returning the last one seen (if any) - only
107    /// the most recent request matters, same as `RlTaskNode::poll_complete`'s "arrival is the
108    /// whole signal" approach to draining a channel once per tick.
109    fn poll_latest_request(receiver: &mut Receiver<Value>) -> Option<Option<String>> {
110        let mut latest = None;
111        while let Ok(value) = receiver.try_recv() {
112            match serde_json::from_value::<RlTargetRequest>(value) {
113                Ok(request) => latest = Some(request.target_class),
114                Err(e) => tracing::warn!("SetRlTargetNode: dropping malformed target request: {}", e),
115            }
116        }
117        latest
118    }
119}
120
121#[async_trait]
122impl BehaviorNode for SetRlTargetNode {
123    async fn tick(&mut self, ctx: &Context) -> anyhow::Result<NodeStatus> {
124        if self.receiver.is_none() {
125            self.receiver = Some(ctx.subscribe_raw::<Value>(&self.request_topic).await?);
126        }
127
128        if !self.published_initial {
129            self.published_initial = true;
130            if self.target.is_some() {
131                self.publish(ctx, self.target.clone()).await?;
132            }
133        }
134
135        let latest = Self::poll_latest_request(self.receiver.as_mut().expect("just subscribed above"));
136        if let Some(target) = latest {
137            self.publish(ctx, target).await?;
138        }
139
140        Ok(NodeStatus::Running)
141    }
142
143    fn name(&self) -> &str {
144        "SetRlTargetNode"
145    }
146}
147
148#[cfg(test)]
149mod tests {
150    use super::*;
151    use crate::test_helpers::create_test_context;
152
153    #[test]
154    fn from_json_reads_topic_request_topic_and_target() {
155        let node = SetRlTargetNode::from_json(serde_json::json!({
156            "topic": "/rl/control",
157            "request_topic": "/rl/target/request",
158            "target": "cat"
159        }))
160        .unwrap();
161
162        assert_eq!(node.topic, "/rl/control");
163        assert_eq!(node.request_topic, "/rl/target/request");
164        assert_eq!(node.target, Some("cat".to_string()));
165    }
166
167    #[test]
168    fn from_json_defaults_topics_and_allows_no_target() {
169        let node = SetRlTargetNode::from_json(serde_json::json!({})).unwrap();
170
171        assert_eq!(node.topic, "/rl/control");
172        assert_eq!(node.request_topic, "/rl/target/request");
173        assert_eq!(node.target, None);
174    }
175
176    #[test]
177    fn from_json_accepts_null_the_same_as_an_empty_object() {
178        let node = SetRlTargetNode::from_json(Value::Null).unwrap();
179
180        assert_eq!(node.topic, "/rl/control");
181        assert_eq!(node.request_topic, "/rl/target/request");
182        assert_eq!(node.target, None);
183    }
184
185    #[test]
186    fn poll_latest_request_returns_none_when_empty() {
187        let (_tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
188        let mut receiver = Receiver::from(rx);
189        assert_eq!(SetRlTargetNode::poll_latest_request(&mut receiver), None);
190    }
191
192    #[test]
193    fn poll_latest_request_keeps_only_the_most_recent_value() {
194        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
195        let mut receiver = Receiver::from(rx);
196        tx.send(serde_json::json!({ "target_class": "dog" })).unwrap();
197        tx.send(serde_json::json!({ "target_class": "cat" })).unwrap();
198
199        assert_eq!(
200            SetRlTargetNode::poll_latest_request(&mut receiver),
201            Some(Some("cat".to_string()))
202        );
203    }
204
205    #[test]
206    fn poll_latest_request_accepts_a_null_target_class_as_a_clear_request() {
207        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
208        let mut receiver = Receiver::from(rx);
209        tx.send(serde_json::json!({ "target_class": null })).unwrap();
210
211        assert_eq!(SetRlTargetNode::poll_latest_request(&mut receiver), Some(None));
212    }
213
214    #[test]
215    fn poll_latest_request_drops_malformed_requests_and_keeps_polling() {
216        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
217        let mut receiver = Receiver::from(rx);
218        tx.send(serde_json::json!({ "target_class": 42 })).unwrap();
219        tx.send(serde_json::json!({ "target_class": "cat" })).unwrap();
220
221        assert_eq!(
222            SetRlTargetNode::poll_latest_request(&mut receiver),
223            Some(Some("cat".to_string()))
224        );
225    }
226
227    #[tokio::test]
228    #[ignore]
229    async fn tick_publishes_configured_target_once_then_stays_running() {
230        let ctx = create_test_context("test_set_rl_target").await;
231
232        let mut node = SetRlTargetNode::from_json(serde_json::json!({
233            "topic": "/rl/control",
234            "target": "cat"
235        }))
236        .unwrap();
237
238        let mut receiver = ctx.subscribe_raw::<Value>("/rl/control").await.unwrap();
239
240        let status = node.tick(&ctx).await.unwrap();
241        assert_eq!(status, NodeStatus::Running);
242
243        let received = tokio::time::timeout(std::time::Duration::from_millis(500), receiver.recv())
244            .await
245            .expect("timed out waiting for published command")
246            .expect("channel closed without a message");
247
248        assert_eq!(received["action"], "set_target_class");
249        assert_eq!(received["params"]["target_class"], "cat");
250
251        let status = node.tick(&ctx).await.unwrap();
252        assert_eq!(status, NodeStatus::Running);
253        let timed_out = tokio::time::timeout(std::time::Duration::from_millis(200), receiver.recv()).await;
254        assert!(timed_out.is_err(), "expected no second publish without a new request");
255    }
256
257    #[tokio::test]
258    #[ignore]
259    async fn tick_relays_a_live_request_published_on_request_topic() {
260        let ctx = create_test_context("test_set_rl_target_request").await;
261
262        let mut node = SetRlTargetNode::from_json(serde_json::json!({
263            "topic": "/rl/control",
264            "request_topic": "/rl/target/request"
265        }))
266        .unwrap();
267
268        node.tick(&ctx).await.unwrap();
269
270        let mut control_receiver = ctx.subscribe_raw::<Value>("/rl/control").await.unwrap();
271        ctx.publish_raw("/rl/target/request", &serde_json::json!({ "target_class": "dog" }))
272            .await
273            .unwrap();
274
275        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
276        node.tick(&ctx).await.unwrap();
277
278        let received = tokio::time::timeout(std::time::Duration::from_millis(500), control_receiver.recv())
279            .await
280            .expect("timed out waiting for relayed command")
281            .expect("channel closed without a message");
282
283        assert_eq!(received["action"], "set_target_class");
284        assert_eq!(received["params"]["target_class"], "dog");
285    }
286}
mecha10_behavior_runtime/actions/set_rl_target.rs

mecha10_behavior_runtime/actions/
set_rl_target.rs
