1use crate::draft14::message::{ControlMessage, FetchPayload};
2use crate::fields::{FieldMap as Map, FieldValue as Value};
3use crate::types::*;
4
5use crate::kvp::{KeyValuePair, KvpValue};
6use crate::varint::VarInt;
7
8fn vi(v: u64) -> Value {
9 Value::Uint(v)
10}
11
12fn ns_to_json(ns: &TrackNamespace) -> Value {
13 Value::Array(
14 ns.0.iter().map(|e| Value::Text(String::from_utf8_lossy(e).into_owned())).collect(),
15 )
16}
17
18fn loc_to_json(loc: &Location) -> Value {
19 let mut o = Map::new();
20 o.insert("group".into(), vi(loc.group.into_inner()));
21 o.insert("object".into(), vi(loc.object.into_inner()));
22 Value::Map(o)
23}
24
25fn auth_token_to_json_d14(bytes: &[u8]) -> Value {
29 let mut buf = bytes;
30 let alias_type = match VarInt::decode(&mut buf) {
31 Ok(v) => v,
32 Err(_) => return Value::Bytes(bytes.to_vec()),
33 };
34 let at = alias_type.into_inner();
35 let mut o = Map::new();
36 o.insert("alias_type".to_string(), Value::Uint(at));
37 match at {
38 0 | 2 => {
39 if let Ok(ta) = VarInt::decode(&mut buf) {
40 o.insert("token_alias".to_string(), Value::Uint(ta.into_inner()));
41 }
42 }
43 1 => {
44 if let Ok(ta) = VarInt::decode(&mut buf) {
45 o.insert("token_alias".to_string(), Value::Uint(ta.into_inner()));
46 }
47 if let Ok(tt) = VarInt::decode(&mut buf) {
48 o.insert("token_type".to_string(), Value::Uint(tt.into_inner()));
49 }
50 o.insert("token_value".to_string(), Value::Bytes(buf.to_vec()));
51 }
52 _ => {
53 if let Ok(tt) = VarInt::decode(&mut buf) {
54 o.insert("token_type".to_string(), Value::Uint(tt.into_inner()));
55 }
56 o.insert("token_value".to_string(), Value::Bytes(buf.to_vec()));
57 }
58 }
59 Value::Map(o)
60}
61
62fn d14_setup_param_name(key: u64) -> Option<&'static str> {
105 match key {
106 0x01 => Some("path"),
107 0x02 => Some("max_request_id"),
108 0x03 => Some("authorization_token"),
109 0x04 => Some("max_auth_token_cache_size"),
110 0x05 => Some("authority"),
111 _ => None,
112 }
113}
114
115fn d14_msg_param_name(key: u64) -> Option<&'static str> {
117 match key {
118 0x02 => Some("delivery_timeout"),
119 0x03 => Some("authorization_token"),
120 0x04 => Some("max_cache_duration"),
121 _ => None,
122 }
123}
124
125fn kvp_to_json(params: &[KeyValuePair], name_fn: fn(u64) -> Option<&'static str>) -> Value {
127 crate::fields::kvp_entries(params, |key, value| {
128 let Some(name) = name_fn(key) else {
129 return (None, None);
130 };
131 let rendered = match value {
132 KvpValue::Varint(v) => Value::Uint(v.into_inner()),
133 KvpValue::Bytes(b) if name == "authorization_token" => auth_token_to_json_d14(b),
134 KvpValue::Bytes(b) => Value::Text(String::from_utf8_lossy(b).into_owned()),
135 };
136 (Some(name), Some(rendered))
137 })
138}
139
140fn kvp_to_json_d14(params: &[KeyValuePair]) -> Value {
141 kvp_to_json(params, d14_msg_param_name)
142}
143
144pub(crate) fn kvp_to_json_d14_setup(params: &[KeyValuePair]) -> Value {
145 kvp_to_json(params, d14_setup_param_name)
146}
147
148pub fn message_fields(msg: &ControlMessage) -> Map {
154 let obj = match msg {
155 ControlMessage::ClientSetup(m) => {
156 let mut o = Map::new();
157 o.insert(
158 "supported_versions".into(),
159 Value::Array(m.supported_versions.iter().map(|v| vi(v.into_inner())).collect()),
160 );
161 o.insert("parameters".into(), kvp_to_json_d14_setup(&m.parameters));
162 o
163 }
164 ControlMessage::ServerSetup(m) => {
165 let mut o = Map::new();
166 o.insert("selected_version".into(), vi(m.selected_version.into_inner()));
167 o.insert("parameters".into(), kvp_to_json_d14_setup(&m.parameters));
168 o
169 }
170 ControlMessage::GoAway(m) => {
171 let mut o = Map::new();
172 o.insert(
173 "new_session_uri".into(),
174 Value::Text(String::from_utf8_lossy(&m.new_session_uri).into_owned()),
175 );
176 o
177 }
178 ControlMessage::MaxRequestId(m) => {
179 let mut o = Map::new();
180 o.insert("request_id".into(), vi(m.request_id.into_inner()));
181 o
182 }
183 ControlMessage::RequestsBlocked(m) => {
184 let mut o = Map::new();
185 o.insert("request_id".into(), vi(m.maximum_request_id.into_inner()));
186 o
187 }
188 ControlMessage::Subscribe(m) => {
189 let mut o = Map::new();
190 o.insert("request_id".into(), vi(m.request_id.into_inner()));
191 o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
192 o.insert(
193 "track_name".into(),
194 Value::Text(String::from_utf8_lossy(&m.track_name).into_owned()),
195 );
196 o.insert("subscriber_priority".into(), vi(m.subscriber_priority as u64));
197 o.insert("group_order".into(), vi(m.group_order as u64));
198 o.insert("forward".into(), vi(m.forward as u64));
199 o.insert("filter_type".into(), vi(m.filter_type as u64));
200 if let Some(loc) = &m.start_location {
201 o.insert("start_group".into(), vi(loc.group.into_inner()));
202 o.insert("start_object".into(), vi(loc.object.into_inner()));
203 }
204 if let Some(eg) = &m.end_group {
205 o.insert("end_group".into(), vi(eg.into_inner()));
206 }
207 o.insert("parameters".into(), kvp_to_json_d14(&m.parameters));
208 o
209 }
210 ControlMessage::SubscribeOk(m) => {
211 let mut o = Map::new();
212 o.insert("request_id".into(), vi(m.request_id.into_inner()));
213 o.insert("track_alias".into(), vi(m.track_alias.into_inner()));
214 o.insert("expires".into(), vi(m.expires.into_inner()));
215 o.insert("group_order".into(), vi(m.group_order as u64));
216 o.insert("content_exists".into(), vi(m.content_exists as u64));
217 if let Some(loc) = &m.largest_location {
218 o.insert("largest_location".into(), loc_to_json(loc));
219 }
220 o.insert("parameters".into(), kvp_to_json_d14(&m.parameters));
221 o
222 }
223 ControlMessage::SubscribeError(m) => {
224 let mut o = Map::new();
225 o.insert("request_id".into(), vi(m.request_id.into_inner()));
226 o.insert("error_code".into(), vi(m.error_code.into_inner()));
227 o.insert(
228 "reason_phrase".into(),
229 Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
230 );
231 o
232 }
233 ControlMessage::SubscribeUpdate(m) => {
234 let mut o = Map::new();
235 o.insert("request_id".into(), vi(m.request_id.into_inner()));
236 o.insert("subscription_request_id".into(), vi(m.subscription_request_id.into_inner()));
237 o.insert("start_group".into(), vi(m.start_location.group.into_inner()));
238 o.insert("start_object".into(), vi(m.start_location.object.into_inner()));
239 o.insert("end_group".into(), vi(m.end_group.into_inner()));
240 o.insert("subscriber_priority".into(), vi(m.subscriber_priority as u64));
241 o.insert("forward".into(), vi(m.forward as u64));
242 o.insert("parameters".into(), kvp_to_json_d14(&m.parameters));
243 o
244 }
245 ControlMessage::Unsubscribe(m) => {
246 let mut o = Map::new();
247 o.insert("request_id".into(), vi(m.request_id.into_inner()));
248 o
249 }
250 ControlMessage::Publish(m) => {
251 let mut o = Map::new();
252 o.insert("request_id".into(), vi(m.request_id.into_inner()));
253 o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
254 o.insert(
255 "track_name".into(),
256 Value::Text(String::from_utf8_lossy(&m.track_name).into_owned()),
257 );
258 o.insert("track_alias".into(), vi(m.track_alias.into_inner()));
259 o.insert("group_order".into(), vi(m.group_order as u64));
260 o.insert("content_exists".into(), vi(m.content_exists as u64));
261 if let Some(loc) = &m.largest_location {
262 o.insert("largest_location".into(), loc_to_json(loc));
263 }
264 o.insert("forward".into(), vi(m.forward as u64));
265 o.insert("parameters".into(), kvp_to_json_d14(&m.parameters));
266 o
267 }
268 ControlMessage::PublishOk(m) => {
269 let mut o = Map::new();
270 o.insert("request_id".into(), vi(m.request_id.into_inner()));
271 o.insert("forward".into(), vi(m.forward as u64));
272 o.insert("subscriber_priority".into(), vi(m.subscriber_priority as u64));
273 o.insert("group_order".into(), vi(m.group_order as u64));
274 o.insert("filter_type".into(), vi(m.filter_type as u64));
275 if let Some(loc) = &m.start_location {
276 o.insert("start_group".into(), vi(loc.group.into_inner()));
277 o.insert("start_object".into(), vi(loc.object.into_inner()));
278 }
279 if let Some(eg) = &m.end_group {
280 o.insert("end_group".into(), vi(eg.into_inner()));
281 }
282 o.insert("parameters".into(), kvp_to_json_d14(&m.parameters));
283 o
284 }
285 ControlMessage::PublishError(m) => {
286 let mut o = Map::new();
287 o.insert("request_id".into(), vi(m.request_id.into_inner()));
288 o.insert("error_code".into(), vi(m.error_code.into_inner()));
289 o.insert(
290 "reason_phrase".into(),
291 Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
292 );
293 o
294 }
295 ControlMessage::PublishDone(m) => {
296 let mut o = Map::new();
297 o.insert("request_id".into(), vi(m.request_id.into_inner()));
298 o.insert("status_code".into(), vi(m.status_code.into_inner()));
299 o.insert("stream_count".into(), vi(m.stream_count.into_inner()));
300 o.insert(
301 "reason_phrase".into(),
302 Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
303 );
304 o
305 }
306 ControlMessage::PublishNamespace(m) => {
307 let mut o = Map::new();
308 o.insert("request_id".into(), vi(m.request_id.into_inner()));
309 o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
310 o.insert("parameters".into(), kvp_to_json_d14(&m.parameters));
311 o
312 }
313 ControlMessage::PublishNamespaceOk(m) => {
314 let mut o = Map::new();
315 o.insert("request_id".into(), vi(m.request_id.into_inner()));
316 o
317 }
318 ControlMessage::PublishNamespaceError(m) => {
319 let mut o = Map::new();
320 o.insert("request_id".into(), vi(m.request_id.into_inner()));
321 o.insert("error_code".into(), vi(m.error_code.into_inner()));
322 o.insert(
323 "reason_phrase".into(),
324 Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
325 );
326 o
327 }
328 ControlMessage::PublishNamespaceDone(m) => {
329 let mut o = Map::new();
330 o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
331 o
332 }
333 ControlMessage::PublishNamespaceCancel(m) => {
334 let mut o = Map::new();
335 o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
336 o.insert("error_code".into(), vi(m.error_code.into_inner()));
337 o.insert(
338 "reason_phrase".into(),
339 Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
340 );
341 o
342 }
343 ControlMessage::SubscribeNamespace(m) => {
344 let mut o = Map::new();
345 o.insert("request_id".into(), vi(m.request_id.into_inner()));
346 o.insert("namespace_prefix".into(), ns_to_json(&m.track_namespace));
347 o.insert("parameters".into(), kvp_to_json_d14(&m.parameters));
348 o
349 }
350 ControlMessage::SubscribeNamespaceOk(m) => {
351 let mut o = Map::new();
352 o.insert("request_id".into(), vi(m.request_id.into_inner()));
353 o
354 }
355 ControlMessage::SubscribeNamespaceError(m) => {
356 let mut o = Map::new();
357 o.insert("request_id".into(), vi(m.request_id.into_inner()));
358 o.insert("error_code".into(), vi(m.error_code.into_inner()));
359 o.insert(
360 "reason_phrase".into(),
361 Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
362 );
363 o
364 }
365 ControlMessage::UnsubscribeNamespace(m) => {
366 let mut o = Map::new();
367 o.insert("track_namespace_prefix".into(), ns_to_json(&m.track_namespace_prefix));
368 o
369 }
370 ControlMessage::Fetch(m) => {
371 let mut o = Map::new();
372 o.insert("request_id".into(), vi(m.request_id.into_inner()));
373 o.insert("subscriber_priority".into(), vi(m.subscriber_priority as u64));
374 o.insert("group_order".into(), vi(m.group_order as u64));
375 o.insert("fetch_type".into(), vi(m.fetch_type as u64));
376 match &m.fetch_payload {
377 FetchPayload::Standalone {
378 track_namespace,
379 track_name,
380 start_group,
381 start_object,
382 end_group,
383 end_object,
384 } => {
385 o.insert("track_namespace".into(), ns_to_json(track_namespace));
386 o.insert(
387 "track_name".into(),
388 Value::Text(String::from_utf8_lossy(track_name).into_owned()),
389 );
390 o.insert("start_group".into(), vi(start_group.into_inner()));
391 o.insert("start_object".into(), vi(start_object.into_inner()));
392 o.insert("end_group".into(), vi(end_group.into_inner()));
393 o.insert("end_object".into(), vi(end_object.into_inner()));
394 }
395 FetchPayload::Joining { joining_request_id, joining_start } => {
396 o.insert("joining_request_id".into(), vi(joining_request_id.into_inner()));
397 o.insert("joining_start".into(), vi(joining_start.into_inner()));
398 }
399 }
400 o.insert("parameters".into(), kvp_to_json_d14(&m.parameters));
401 o
402 }
403 ControlMessage::FetchOk(m) => {
404 let mut o = Map::new();
405 o.insert("request_id".into(), vi(m.request_id.into_inner()));
406 o.insert("group_order".into(), vi(m.group_order as u64));
407 o.insert("end_of_track".into(), vi(m.end_of_track as u64));
408 o.insert("end_location".into(), loc_to_json(&m.end_location));
409 o.insert("parameters".into(), kvp_to_json_d14(&m.parameters));
410 o
411 }
412 ControlMessage::FetchError(m) => {
413 let mut o = Map::new();
414 o.insert("request_id".into(), vi(m.request_id.into_inner()));
415 o.insert("error_code".into(), vi(m.error_code.into_inner()));
416 o.insert(
417 "reason_phrase".into(),
418 Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
419 );
420 o
421 }
422 ControlMessage::FetchCancel(m) => {
423 let mut o = Map::new();
424 o.insert("request_id".into(), vi(m.request_id.into_inner()));
425 o
426 }
427 ControlMessage::TrackStatus(m) => {
428 let mut o = Map::new();
429 o.insert("request_id".into(), vi(m.request_id.into_inner()));
430 o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
431 o.insert(
432 "track_name".into(),
433 Value::Text(String::from_utf8_lossy(&m.track_name).into_owned()),
434 );
435 o.insert("subscriber_priority".into(), vi(m.subscriber_priority as u64));
436 o.insert("group_order".into(), vi(m.group_order as u64));
437 o.insert("forward".into(), vi(m.forward as u64));
438 o.insert("filter_type".into(), vi(m.filter_type as u64));
439 if let Some(loc) = &m.start_location {
440 o.insert("start_group".into(), vi(loc.group.into_inner()));
441 o.insert("start_object".into(), vi(loc.object.into_inner()));
442 }
443 if let Some(eg) = &m.end_group {
444 o.insert("end_group".into(), vi(eg.into_inner()));
445 }
446 o.insert("parameters".into(), kvp_to_json_d14(&m.parameters));
447 o
448 }
449 ControlMessage::TrackStatusOk(m) => {
450 let mut o = Map::new();
451 o.insert("request_id".into(), vi(m.request_id.into_inner()));
452 o.insert("track_alias".into(), vi(m.track_alias.into_inner()));
453 o.insert("expires".into(), vi(m.expires.into_inner()));
454 o.insert("group_order".into(), vi(m.group_order as u64));
455 o.insert("content_exists".into(), vi(m.content_exists as u64));
456 if let Some(loc) = &m.largest_location {
457 o.insert("largest_location".into(), loc_to_json(loc));
458 }
459 o.insert("parameters".into(), kvp_to_json_d14(&m.parameters));
460 o
461 }
462 ControlMessage::TrackStatusError(m) => {
463 let mut o = Map::new();
464 o.insert("request_id".into(), vi(m.request_id.into_inner()));
465 o.insert("error_code".into(), vi(m.error_code.into_inner()));
466 o.insert(
467 "reason_phrase".into(),
468 Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
469 );
470 o
471 }
472 };
473 obj
474}