1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
//! Request/reply plumbing and the socket reader loop.
use super::*;
impl PluginIpcClient {
/// Send a request under the client's default timeout. See
/// [`send_request_within`](Self::send_request_within).
pub(super) async fn send_request(
&self,
method: &str,
capability: &str,
args: Value,
) -> Result<Envelope, ClientError> {
self.send_request_within(method, capability, args, self.request_timeout)
.await
}
/// Mint a request id, park a waiter, write the frame, then await the routed
/// response for at most `timeout`. Mirrors `_send_request`, including the
/// error-string mapping to typed `capability_denied` /
/// `allowlist_violation` errors.
pub(super) async fn send_request_within(
&self,
method: &str,
capability: &str,
args: Value,
timeout: Duration,
) -> Result<Envelope, ClientError> {
let n = self.next_id.fetch_add(1, Ordering::Relaxed) + 1;
let rid = format!("r{n}");
let env = Envelope {
version: PROTOCOL_VERSION,
kind: "request".to_string(),
method: method.to_string(),
capability: capability.to_string(),
args,
request_id: rid.clone(),
// Read per request, not captured at construction: a rotation mid
// session must be what the next frame carries.
token: self.token.lock().expect("token lock").clone(),
error: None,
};
let frame = env.encode_frame()?;
let (tx, rx) = oneshot::channel();
{
let mut pending = self.pending.lock().expect("pending lock");
pending.insert(rid.clone(), tx);
}
// Write under the shared async lock so concurrent senders serialize.
{
let mut guard = self.writer.lock().await;
let writer = guard.as_mut().ok_or(ClientError::NotConnected);
match writer {
Ok(w) => {
if let Err(e) = w.write_all(&frame).await.and(w.flush().await) {
self.pending.lock().expect("pending lock").remove(&rid);
return Err(ClientError::Io(e));
}
}
Err(_) => {
self.pending.lock().expect("pending lock").remove(&rid);
return Err(ClientError::NotConnected);
}
}
}
let response = match tokio::time::timeout(timeout, rx).await {
Ok(Ok(env)) => env,
// The waiter was dropped: the reader loop ended before the response.
Ok(Err(_)) => {
self.pending.lock().expect("pending lock").remove(&rid);
return Err(ClientError::ConnectionClosed);
}
Err(_) => {
self.pending.lock().expect("pending lock").remove(&rid);
return Err(ClientError::Timeout(timeout));
}
};
if let Some(err) = &response.error {
return Err(map_error_string(err));
}
Ok(response)
}
/// Spawn the reader loop. Owns the read half; routes responses by
/// `request_id`, adopts a rotated capability token, and dispatches
/// `event` / `mavlink.deliver` pushes to the registered callbacks.
/// Mirrors `_reader_loop`.
pub(super) fn spawn_reader_loop(&self, read_half: OwnedReadHalf) -> JoinHandle<()> {
let pending = self.pending.clone();
let event_callbacks = self.event_callbacks.clone();
let mavlink_callbacks = self.mavlink_callbacks.clone();
let vision_callbacks = self.vision_callbacks.clone();
let detection_callbacks = self.detection_callbacks.clone();
let button_callbacks = self.button_callbacks.clone();
let msp_callbacks = self.msp_callbacks.clone();
let aux_callbacks = self.aux_callbacks.clone();
let tool_handler = self.tool_handler.clone();
let writer = self.writer.clone();
let token = self.token.clone();
let plugin_id = self.plugin_id.clone();
let session_lost = self.session_lost.clone();
tokio::spawn(async move {
let mut reader = read_half;
loop {
match read_envelope(&mut reader).await {
Ok(Some(env)) => {
if env.kind == "event" {
if env.method == "mavlink.deliver" {
dispatch_mavlink(&mavlink_callbacks, &env);
} else if env.method == ados_protocol::framebus::methods::DELIVER_FRAME
{
dispatch_vision(&vision_callbacks, &env);
} else if env.method
== ados_protocol::framebus::methods::DELIVER_DETECTION
{
dispatch_detection(&detection_callbacks, &env);
} else if env.method == ados_protocol::buttons::DELIVER {
dispatch_button(&button_callbacks, &env);
} else if env.method == "msp.deliver" {
dispatch_button(&msp_callbacks, &env);
} else if env.method == TOKEN_REFRESH_METHOD {
adopt_refreshed_token(&token, &plugin_id, &env);
} else if env.method == "radio.aux_stream.deliver" {
// Wildcard-only fan-out, exactly like buttons:
// every aux subscriber receives every datagram
// and its own codec routes by channel.
dispatch_button(&aux_callbacks, &env);
} else {
dispatch_event(&event_callbacks, &env);
}
} else if env.kind == "request" && env.method == TOOL_INVOKE_METHOD {
let handler = tool_handler.lock().expect("tool handler lock").clone();
tokio::spawn(answer_tool_invoke(handler, writer.clone(), env));
} else {
// request/response: route to the parked waiter.
if let Some(tx) = pending
.lock()
.expect("pending lock")
.remove(&env.request_id)
{
let _ = tx.send(env);
}
}
}
// Clean EOF (peer closed) or a frame error: end the loop.
Ok(None) => break,
Err(e) => {
tracing::warn!(plugin_id = %plugin_id, error = %e, "plugin ipc reader loop ended");
break;
}
}
}
// Drop every parked waiter so in-flight requests fail fast with
// ConnectionClosed rather than waiting out their timeout.
pending.lock().expect("pending lock").clear();
session_lost.send_replace(true);
})
}
}
/// Run one host-to-plugin `tool.invoke` and write its `response` frame. The
/// request args are `{tool, arguments}`; the response carries the tool result
/// as args, or the error string in `err`. Every invocation is answered, so the
/// host never waits out its timeout for a reply that cannot come: a handler
/// that panics answers `tool_handler_panicked`, and a result that cannot be
/// framed (over the frame cap, or not encodable) answers
/// `tool_reply_unencodable`.
pub(super) async fn answer_tool_invoke(
handler: Option<ToolHandler>,
writer: Arc<AsyncMutex<Option<OwnedWriteHalf>>>,
req: Envelope,
) {
let outcome = match handler {
None => Err("tool_handler_missing".to_string()),
Some(h) => match map_get_str(&req.args, "tool") {
None => Err("tool name missing".to_string()),
Some(tool) => {
let arguments = map_get(&req.args, "arguments").unwrap_or(Value::Nil);
// On its own task so a panic is caught as a JoinError rather
// than unwinding through this one.
match tokio::spawn(h(tool.to_string(), arguments)).await {
Ok(result) => result,
Err(e) => {
tracing::warn!(tool, error = %e, "tool handler did not complete");
Err("tool_handler_panicked".to_string())
}
}
}
},
};
let (args, error) = match outcome {
Ok(v) => (v, None),
Err(e) => (Value::Map(Vec::new()), Some(e)),
};
let mut reply = Envelope {
version: PROTOCOL_VERSION,
kind: "response".to_string(),
method: TOOL_INVOKE_METHOD.to_string(),
capability: req.capability,
args,
request_id: req.request_id,
token: String::new(),
error,
};
let frame = match reply.encode_frame() {
Ok(frame) => frame,
Err(e) => {
reply.args = Value::Map(Vec::new());
reply.error = Some(format!("tool_reply_unencodable: {e}"));
match reply.encode_frame() {
Ok(frame) => frame,
Err(e) => {
tracing::warn!(error = %e, "tool.invoke error reply could not be framed");
return;
}
}
}
};
let mut guard = writer.lock().await;
if let Some(w) = guard.as_mut() {
if w.write_all(&frame).await.and(w.flush().await).is_err() {
tracing::warn!("tool.invoke reply could not be written");
}
}
}