1use crate::draft11::message::*;
2use crate::fields::{FieldMap as Map, FieldValue as Value};
3use crate::kvp::{KeyValuePair, KvpValue};
4use crate::types::*;
5use crate::varint::VarInt;
6
7fn vi(v: u64) -> Value {
8 Value::Uint(v)
9}
10
11fn ns_to_json(ns: &TrackNamespace) -> Value {
12 Value::Array(
13 ns.0.iter().map(|e| Value::Text(String::from_utf8_lossy(e).into_owned())).collect(),
14 )
15}
16
17fn loc_to_json(loc: &Location) -> Value {
18 let mut o = Map::new();
19 o.insert("group".into(), vi(loc.group.into_inner()));
20 o.insert("object".into(), vi(loc.object.into_inner()));
21 Value::Map(o)
22}
23
24fn auth_token_to_json(bytes: &[u8]) -> Value {
67 let mut buf = bytes;
68 let Ok(alias_type) = VarInt::decode(&mut buf) else {
69 return Value::Bytes(bytes.to_vec());
70 };
71 let at = alias_type.into_inner();
72 let mut o = Map::new();
73 o.insert("alias_type".into(), vi(at));
74 match at {
75 0 | 2 => {
77 if let Ok(ta) = VarInt::decode(&mut buf) {
78 o.insert("token_alias".into(), vi(ta.into_inner()));
79 }
80 }
81 1 => {
83 if let Ok(ta) = VarInt::decode(&mut buf) {
84 o.insert("token_alias".into(), vi(ta.into_inner()));
85 }
86 if let Ok(tt) = VarInt::decode(&mut buf) {
87 o.insert("token_type".into(), vi(tt.into_inner()));
88 }
89 o.insert("token_value".into(), Value::Bytes(buf.to_vec()));
90 }
91 _ => {
98 if let Ok(tt) = VarInt::decode(&mut buf) {
99 o.insert("token_type".into(), vi(tt.into_inner()));
100 }
101 o.insert("token_value".into(), Value::Bytes(buf.to_vec()));
102 }
103 }
104 Value::Map(o)
105}
106
107pub(crate) fn kvp_to_json_setup(params: &[KeyValuePair]) -> Value {
117 crate::fields::kvp_entries(params, |key, value| match (key, value) {
118 (0x01, KvpValue::Bytes(b)) => {
119 (Some("path"), Some(Value::Text(String::from_utf8_lossy(b).into_owned())))
120 }
121 (0x02, KvpValue::Varint(v)) => (Some("max_request_id"), Some(vi(v.into_inner()))),
122 (0x04, KvpValue::Varint(v)) => {
125 (Some("max_auth_token_cache_size"), Some(vi(v.into_inner())))
126 }
127 _ => (None, None),
128 })
129}
130
131fn kvp_to_json_msg(params: &[KeyValuePair]) -> Value {
133 crate::fields::kvp_entries(params, |key, value| match (key, value) {
134 (0x01, KvpValue::Bytes(b)) => (Some("authorization_token"), Some(auth_token_to_json(b))),
135 (0x02, KvpValue::Varint(v)) => (Some("delivery_timeout"), Some(vi(v.into_inner()))),
136 (0x04, KvpValue::Varint(v)) => (Some("max_cache_duration"), Some(vi(v.into_inner()))),
137 _ => (None, None),
138 })
139}
140
141pub fn message_fields(msg: &ControlMessage) -> Map {
147 let obj = match msg {
148 ControlMessage::ClientSetup(m) => {
149 let mut o = Map::new();
150 o.insert(
151 "supported_versions".into(),
152 Value::Array(m.supported_versions.iter().map(|v| vi(v.into_inner())).collect()),
153 );
154 o.insert("parameters".into(), kvp_to_json_setup(&m.parameters));
155 o
156 }
157 ControlMessage::ServerSetup(m) => {
158 let mut o = Map::new();
159 o.insert("selected_version".into(), vi(m.selected_version.into_inner()));
160 o.insert("parameters".into(), kvp_to_json_setup(&m.parameters));
161 o
162 }
163 ControlMessage::GoAway(m) => {
164 let mut o = Map::new();
165 o.insert(
166 "new_session_uri".into(),
167 Value::Text(String::from_utf8_lossy(&m.new_session_uri).into_owned()),
168 );
169 o
170 }
171 ControlMessage::MaxRequestId(m) => {
172 let mut o = Map::new();
173 o.insert("request_id".into(), vi(m.request_id.into_inner()));
174 o
175 }
176 ControlMessage::RequestsBlocked(m) => {
177 let mut o = Map::new();
178 o.insert("maximum_request_id".into(), vi(m.maximum_request_id.into_inner()));
179 o
180 }
181 ControlMessage::Subscribe(m) => {
182 let mut o = Map::new();
183 o.insert("request_id".into(), vi(m.request_id.into_inner()));
184 o.insert("track_alias".into(), vi(m.track_alias.into_inner()));
185 o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
186 o.insert(
187 "track_name".into(),
188 Value::Text(String::from_utf8_lossy(&m.track_name).into_owned()),
189 );
190 o.insert("subscriber_priority".into(), vi(m.subscriber_priority as u64));
191 o.insert("group_order".into(), vi(m.group_order as u64));
192 o.insert("forward".into(), vi(m.forward as u64));
193 o.insert("filter_type".into(), vi(m.filter_type.into_inner()));
194 if let Some(sg) = &m.start_group {
195 o.insert("start_group".into(), vi(sg.into_inner()));
196 }
197 if let Some(so) = &m.start_object {
198 o.insert("start_object".into(), vi(so.into_inner()));
199 }
200 if let Some(eg) = &m.end_group {
201 o.insert("end_group".into(), vi(eg.into_inner()));
202 }
203 o.insert("parameters".into(), kvp_to_json_msg(&m.parameters));
204 o
205 }
206 ControlMessage::SubscribeOk(m) => {
207 let mut o = Map::new();
208 o.insert("request_id".into(), vi(m.request_id.into_inner()));
209 o.insert("expires".into(), vi(m.expires.into_inner()));
210 o.insert("group_order".into(), vi(m.group_order as u64));
211 o.insert("content_exists".into(), vi(m.content_exists as u64));
212 if let Some(loc) = &m.largest_location {
213 o.insert("largest_location".into(), loc_to_json(loc));
214 }
215 o.insert("parameters".into(), kvp_to_json_msg(&m.parameters));
216 o
217 }
218 ControlMessage::SubscribeError(m) => {
219 let mut o = Map::new();
220 o.insert("request_id".into(), vi(m.request_id.into_inner()));
221 o.insert("error_code".into(), vi(m.error_code.into_inner()));
222 o.insert(
223 "reason_phrase".into(),
224 Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
225 );
226 o.insert("track_alias".into(), vi(m.track_alias.into_inner()));
227 o
228 }
229 ControlMessage::SubscribeUpdate(m) => {
230 let mut o = Map::new();
231 o.insert("request_id".into(), vi(m.request_id.into_inner()));
232 o.insert("start_group".into(), vi(m.start_group.into_inner()));
233 o.insert("start_object".into(), vi(m.start_object.into_inner()));
234 o.insert("end_group".into(), vi(m.end_group.into_inner()));
235 o.insert("subscriber_priority".into(), vi(m.subscriber_priority as u64));
236 o.insert("forward".into(), vi(m.forward as u64));
237 o.insert("parameters".into(), kvp_to_json_msg(&m.parameters));
238 o
239 }
240 ControlMessage::SubscribeDone(m) => {
241 let mut o = Map::new();
242 o.insert("request_id".into(), vi(m.request_id.into_inner()));
243 o.insert("status_code".into(), vi(m.status_code.into_inner()));
244 o.insert("stream_count".into(), vi(m.stream_count.into_inner()));
245 o.insert(
246 "reason_phrase".into(),
247 Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
248 );
249 o
250 }
251 ControlMessage::Unsubscribe(m) => {
252 let mut o = Map::new();
253 o.insert("request_id".into(), vi(m.request_id.into_inner()));
254 o
255 }
256 ControlMessage::Announce(m) => {
257 let mut o = Map::new();
258 o.insert("request_id".into(), vi(m.request_id.into_inner()));
259 o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
260 o.insert("parameters".into(), kvp_to_json_msg(&m.parameters));
261 o
262 }
263 ControlMessage::AnnounceOk(m) => {
264 let mut o = Map::new();
265 o.insert("request_id".into(), vi(m.request_id.into_inner()));
266 o
267 }
268 ControlMessage::AnnounceError(m) => {
269 let mut o = Map::new();
270 o.insert("request_id".into(), vi(m.request_id.into_inner()));
271 o.insert("error_code".into(), vi(m.error_code.into_inner()));
272 o.insert(
273 "reason_phrase".into(),
274 Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
275 );
276 o
277 }
278 ControlMessage::AnnounceCancel(m) => {
279 let mut o = Map::new();
280 o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
281 o.insert("error_code".into(), vi(m.error_code.into_inner()));
282 o.insert(
283 "reason_phrase".into(),
284 Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
285 );
286 o
287 }
288 ControlMessage::Unannounce(m) => {
289 let mut o = Map::new();
290 o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
291 o
292 }
293 ControlMessage::SubscribeAnnounces(m) => {
294 let mut o = Map::new();
295 o.insert("request_id".into(), vi(m.request_id.into_inner()));
296 o.insert("track_namespace_prefix".into(), ns_to_json(&m.track_namespace_prefix));
297 o.insert("parameters".into(), kvp_to_json_msg(&m.parameters));
298 o
299 }
300 ControlMessage::SubscribeAnnouncesOk(m) => {
301 let mut o = Map::new();
302 o.insert("request_id".into(), vi(m.request_id.into_inner()));
303 o
304 }
305 ControlMessage::SubscribeAnnouncesError(m) => {
306 let mut o = Map::new();
307 o.insert("request_id".into(), vi(m.request_id.into_inner()));
308 o.insert("error_code".into(), vi(m.error_code.into_inner()));
309 o.insert(
310 "reason_phrase".into(),
311 Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
312 );
313 o
314 }
315 ControlMessage::UnsubscribeAnnounces(m) => {
316 let mut o = Map::new();
317 o.insert("track_namespace_prefix".into(), ns_to_json(&m.track_namespace_prefix));
318 o
319 }
320 ControlMessage::TrackStatusRequest(m) => {
321 let mut o = Map::new();
322 o.insert("request_id".into(), vi(m.request_id.into_inner()));
323 o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
324 o.insert(
325 "track_name".into(),
326 Value::Text(String::from_utf8_lossy(&m.track_name).into_owned()),
327 );
328 o.insert("parameters".into(), kvp_to_json_msg(&m.parameters));
329 o
330 }
331 ControlMessage::TrackStatus(m) => {
332 let mut o = Map::new();
333 o.insert("request_id".into(), vi(m.request_id.into_inner()));
334 o.insert("status_code".into(), vi(m.status_code.into_inner()));
335 o.insert("largest_location".into(), loc_to_json(&m.largest_location));
336 o.insert("parameters".into(), kvp_to_json_msg(&m.parameters));
337 o
338 }
339 ControlMessage::Fetch(m) => {
340 let mut o = Map::new();
341 o.insert("request_id".into(), vi(m.request_id.into_inner()));
342 o.insert("subscriber_priority".into(), vi(m.subscriber_priority as u64));
343 o.insert("group_order".into(), vi(m.group_order as u64));
344 o.insert("fetch_type".into(), vi(m.fetch_type as u64));
345 match &m.fetch_payload {
346 FetchPayload::Standalone {
347 track_namespace,
348 track_name,
349 start_group,
350 start_object,
351 end_group,
352 end_object,
353 } => {
354 o.insert("track_namespace".into(), ns_to_json(track_namespace));
355 o.insert(
356 "track_name".into(),
357 Value::Text(String::from_utf8_lossy(track_name).into_owned()),
358 );
359 o.insert("start_group".into(), vi(start_group.into_inner()));
360 o.insert("start_object".into(), vi(start_object.into_inner()));
361 o.insert("end_group".into(), vi(end_group.into_inner()));
362 o.insert("end_object".into(), vi(end_object.into_inner()));
363 }
364 FetchPayload::Joining { joining_subscribe_id, joining_start } => {
365 o.insert("joining_subscribe_id".into(), vi(joining_subscribe_id.into_inner()));
366 o.insert("joining_start".into(), vi(joining_start.into_inner()));
367 }
368 }
369 o.insert("parameters".into(), kvp_to_json_msg(&m.parameters));
370 o
371 }
372 ControlMessage::FetchOk(m) => {
373 let mut o = Map::new();
374 o.insert("request_id".into(), vi(m.request_id.into_inner()));
375 o.insert("group_order".into(), vi(m.group_order as u64));
376 o.insert("end_of_track".into(), vi(m.end_of_track as u64));
377 o.insert("end_location".into(), loc_to_json(&m.end_location));
378 o.insert("parameters".into(), kvp_to_json_msg(&m.parameters));
379 o
380 }
381 ControlMessage::FetchError(m) => {
382 let mut o = Map::new();
383 o.insert("request_id".into(), vi(m.request_id.into_inner()));
384 o.insert("error_code".into(), vi(m.error_code.into_inner()));
385 o.insert(
386 "reason_phrase".into(),
387 Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
388 );
389 o
390 }
391 ControlMessage::FetchCancel(m) => {
392 let mut o = Map::new();
393 o.insert("request_id".into(), vi(m.request_id.into_inner()));
394 o
395 }
396 };
397 obj
398}