1use crate::draft12::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 {
25 let mut buf = bytes;
26 let alias_type = VarInt::decode(&mut buf).unwrap();
27 let token_type = VarInt::decode(&mut buf).unwrap();
28 let token_value = buf;
29 let mut o = Map::new();
30 o.insert("alias_type".into(), vi(alias_type.into_inner()));
31 o.insert("token_type".into(), vi(token_type.into_inner()));
32 o.insert("token_value".into(), Value::Bytes(token_value.to_vec()));
33 Value::Map(o)
34}
35
36fn kvp_to_json_setup(params: &[KeyValuePair]) -> Value {
37 let mut obj = Map::new();
38 for p in params {
39 let key = p.key.into_inner();
40 match (key, &p.value) {
41 (0x01, KvpValue::Bytes(b)) => {
42 obj.insert("path".into(), Value::Text(String::from_utf8_lossy(b).into_owned()));
43 }
44 (0x02, KvpValue::Varint(v)) => {
45 obj.insert("max_request_id".into(), vi(v.into_inner()));
46 }
47 _ => {}
48 }
49 }
50 Value::Map(obj)
51}
52
53fn kvp_to_json_msg(params: &[KeyValuePair]) -> Value {
54 let mut obj = Map::new();
55 for p in params {
56 let key = p.key.into_inner();
57 match (key, &p.value) {
58 (0x03, KvpValue::Bytes(b)) => {
59 obj.insert("authorization_token".into(), auth_token_to_json(b));
60 }
61 (0x02, KvpValue::Varint(v)) => {
62 obj.insert("delivery_timeout".into(), vi(v.into_inner()));
63 }
64 (0x04, KvpValue::Varint(v)) => {
65 obj.insert("max_cache_duration".into(), vi(v.into_inner()));
66 }
67 _ => {}
68 }
69 }
70 Value::Map(obj)
71}
72
73pub fn message_fields(msg: &ControlMessage) -> Map {
79 let obj = match msg {
80 ControlMessage::ClientSetup(m) => {
81 let mut o = Map::new();
82 o.insert(
83 "supported_versions".into(),
84 Value::Array(m.supported_versions.iter().map(|v| vi(v.into_inner())).collect()),
85 );
86 o.insert("parameters".into(), kvp_to_json_setup(&m.parameters));
87 o
88 }
89 ControlMessage::ServerSetup(m) => {
90 let mut o = Map::new();
91 o.insert("selected_version".into(), vi(m.selected_version.into_inner()));
92 o.insert("parameters".into(), kvp_to_json_setup(&m.parameters));
93 o
94 }
95 ControlMessage::GoAway(m) => {
96 let mut o = Map::new();
97 o.insert(
98 "new_session_uri".into(),
99 Value::Text(String::from_utf8_lossy(&m.new_session_uri).into_owned()),
100 );
101 o
102 }
103 ControlMessage::MaxRequestId(m) => {
104 let mut o = Map::new();
105 o.insert("request_id".into(), vi(m.request_id.into_inner()));
106 o
107 }
108 ControlMessage::RequestsBlocked(m) => {
109 let mut o = Map::new();
110 o.insert("maximum_request_id".into(), vi(m.maximum_request_id.into_inner()));
111 o
112 }
113 ControlMessage::Subscribe(m) => {
114 let mut o = Map::new();
115 o.insert("request_id".into(), vi(m.request_id.into_inner()));
116 o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
117 o.insert(
118 "track_name".into(),
119 Value::Text(String::from_utf8_lossy(&m.track_name).into_owned()),
120 );
121 o.insert("subscriber_priority".into(), vi(m.subscriber_priority as u64));
122 o.insert("group_order".into(), vi(m.group_order as u64));
123 o.insert("forward".into(), vi(m.forward as u64));
124 o.insert("filter_type".into(), vi(m.filter_type.into_inner()));
125 if let Some(sg) = &m.start_group {
126 o.insert("start_group".into(), vi(sg.into_inner()));
127 }
128 if let Some(so) = &m.start_object {
129 o.insert("start_object".into(), vi(so.into_inner()));
130 }
131 if let Some(eg) = &m.end_group {
132 o.insert("end_group".into(), vi(eg.into_inner()));
133 }
134 o.insert("parameters".into(), kvp_to_json_msg(&m.parameters));
135 o
136 }
137 ControlMessage::SubscribeOk(m) => {
138 let mut o = Map::new();
139 o.insert("request_id".into(), vi(m.request_id.into_inner()));
140 o.insert("track_alias".into(), vi(m.track_alias.into_inner()));
141 o.insert("expires".into(), vi(m.expires.into_inner()));
142 o.insert("group_order".into(), vi(m.group_order as u64));
143 o.insert("content_exists".into(), vi(m.content_exists as u64));
144 if let Some(loc) = &m.largest_location {
145 o.insert("largest_location".into(), loc_to_json(loc));
146 }
147 o.insert("parameters".into(), kvp_to_json_msg(&m.parameters));
148 o
149 }
150 ControlMessage::SubscribeError(m) => {
151 let mut o = Map::new();
152 o.insert("request_id".into(), vi(m.request_id.into_inner()));
153 o.insert("error_code".into(), vi(m.error_code.into_inner()));
154 o.insert(
155 "reason_phrase".into(),
156 Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
157 );
158 o
159 }
160 ControlMessage::SubscribeUpdate(m) => {
161 let mut o = Map::new();
162 o.insert("request_id".into(), vi(m.request_id.into_inner()));
163 o.insert("start_group".into(), vi(m.start_group.into_inner()));
164 o.insert("start_object".into(), vi(m.start_object.into_inner()));
165 o.insert("end_group".into(), vi(m.end_group.into_inner()));
166 o.insert("subscriber_priority".into(), vi(m.subscriber_priority as u64));
167 o.insert("forward".into(), vi(m.forward as u64));
168 o.insert("parameters".into(), kvp_to_json_msg(&m.parameters));
169 o
170 }
171 ControlMessage::SubscribeDone(m) => {
172 let mut o = Map::new();
173 o.insert("request_id".into(), vi(m.request_id.into_inner()));
174 o.insert("status_code".into(), vi(m.status_code.into_inner()));
175 o.insert("stream_count".into(), vi(m.stream_count.into_inner()));
176 o.insert(
177 "reason_phrase".into(),
178 Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
179 );
180 o
181 }
182 ControlMessage::Unsubscribe(m) => {
183 let mut o = Map::new();
184 o.insert("request_id".into(), vi(m.request_id.into_inner()));
185 o
186 }
187 ControlMessage::Announce(m) => {
188 let mut o = Map::new();
189 o.insert("request_id".into(), vi(m.request_id.into_inner()));
190 o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
191 o.insert("parameters".into(), kvp_to_json_msg(&m.parameters));
192 o
193 }
194 ControlMessage::AnnounceOk(m) => {
195 let mut o = Map::new();
196 o.insert("request_id".into(), vi(m.request_id.into_inner()));
197 o
198 }
199 ControlMessage::AnnounceError(m) => {
200 let mut o = Map::new();
201 o.insert("request_id".into(), vi(m.request_id.into_inner()));
202 o.insert("error_code".into(), vi(m.error_code.into_inner()));
203 o.insert(
204 "reason_phrase".into(),
205 Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
206 );
207 o
208 }
209 ControlMessage::AnnounceCancel(m) => {
210 let mut o = Map::new();
211 o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
212 o.insert("error_code".into(), vi(m.error_code.into_inner()));
213 o.insert(
214 "reason_phrase".into(),
215 Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
216 );
217 o
218 }
219 ControlMessage::Unannounce(m) => {
220 let mut o = Map::new();
221 o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
222 o
223 }
224 ControlMessage::SubscribeAnnounces(m) => {
225 let mut o = Map::new();
226 o.insert("request_id".into(), vi(m.request_id.into_inner()));
227 o.insert("track_namespace_prefix".into(), ns_to_json(&m.track_namespace_prefix));
228 o.insert("parameters".into(), kvp_to_json_msg(&m.parameters));
229 o
230 }
231 ControlMessage::SubscribeAnnouncesOk(m) => {
232 let mut o = Map::new();
233 o.insert("request_id".into(), vi(m.request_id.into_inner()));
234 o
235 }
236 ControlMessage::SubscribeAnnouncesError(m) => {
237 let mut o = Map::new();
238 o.insert("request_id".into(), vi(m.request_id.into_inner()));
239 o.insert("error_code".into(), vi(m.error_code.into_inner()));
240 o.insert(
241 "reason_phrase".into(),
242 Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
243 );
244 o
245 }
246 ControlMessage::UnsubscribeAnnounces(m) => {
247 let mut o = Map::new();
248 o.insert("track_namespace_prefix".into(), ns_to_json(&m.track_namespace_prefix));
249 o
250 }
251 ControlMessage::TrackStatusRequest(m) => {
252 let mut o = Map::new();
253 o.insert("request_id".into(), vi(m.request_id.into_inner()));
254 o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
255 o.insert(
256 "track_name".into(),
257 Value::Text(String::from_utf8_lossy(&m.track_name).into_owned()),
258 );
259 o.insert("parameters".into(), kvp_to_json_msg(&m.parameters));
260 o
261 }
262 ControlMessage::TrackStatus(m) => {
263 let mut o = Map::new();
264 o.insert("request_id".into(), vi(m.request_id.into_inner()));
265 o.insert("status_code".into(), vi(m.status_code.into_inner()));
266 o.insert("largest_location".into(), loc_to_json(&m.largest_location));
267 o.insert("parameters".into(), kvp_to_json_msg(&m.parameters));
268 o
269 }
270 ControlMessage::Fetch(m) => {
271 let mut o = Map::new();
272 o.insert("request_id".into(), vi(m.request_id.into_inner()));
273 o.insert("subscriber_priority".into(), vi(m.subscriber_priority as u64));
274 o.insert("group_order".into(), vi(m.group_order as u64));
275 o.insert("fetch_type".into(), vi(m.fetch_type as u64));
276 match &m.fetch_payload {
277 FetchPayload::Standalone {
278 track_namespace,
279 track_name,
280 start_group,
281 start_object,
282 end_group,
283 end_object,
284 } => {
285 o.insert("track_namespace".into(), ns_to_json(track_namespace));
286 o.insert(
287 "track_name".into(),
288 Value::Text(String::from_utf8_lossy(track_name).into_owned()),
289 );
290 o.insert("start_group".into(), vi(start_group.into_inner()));
291 o.insert("start_object".into(), vi(start_object.into_inner()));
292 o.insert("end_group".into(), vi(end_group.into_inner()));
293 o.insert("end_object".into(), vi(end_object.into_inner()));
294 }
295 FetchPayload::Joining { joining_request_id, joining_start } => {
296 o.insert("joining_subscribe_id".into(), vi(joining_request_id.into_inner()));
302 o.insert("joining_start".into(), vi(joining_start.into_inner()));
303 }
304 }
305 o.insert("parameters".into(), kvp_to_json_msg(&m.parameters));
306 o
307 }
308 ControlMessage::FetchOk(m) => {
309 let mut o = Map::new();
310 o.insert("request_id".into(), vi(m.request_id.into_inner()));
311 o.insert("group_order".into(), vi(m.group_order as u64));
312 o.insert("end_of_track".into(), vi(m.end_of_track as u64));
313 o.insert("end_location".into(), loc_to_json(&m.end_location));
314 o.insert("parameters".into(), kvp_to_json_msg(&m.parameters));
315 o
316 }
317 ControlMessage::FetchError(m) => {
318 let mut o = Map::new();
319 o.insert("request_id".into(), vi(m.request_id.into_inner()));
320 o.insert("error_code".into(), vi(m.error_code.into_inner()));
321 o.insert(
322 "reason_phrase".into(),
323 Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
324 );
325 o
326 }
327 ControlMessage::FetchCancel(m) => {
328 let mut o = Map::new();
329 o.insert("request_id".into(), vi(m.request_id.into_inner()));
330 o
331 }
332 ControlMessage::Publish(m) => {
333 let mut o = Map::new();
334 o.insert("request_id".into(), vi(m.request_id.into_inner()));
335 o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
336 o.insert(
337 "track_name".into(),
338 Value::Text(String::from_utf8_lossy(&m.track_name).into_owned()),
339 );
340 o.insert("track_alias".into(), vi(m.track_alias.into_inner()));
341 o.insert("group_order".into(), vi(m.group_order as u64));
342 o.insert("content_exists".into(), vi(m.content_exists as u64));
343 if let Some(loc) = &m.largest_location {
344 o.insert("largest_location".into(), loc_to_json(loc));
345 }
346 o.insert("forward".into(), vi(m.forward as u64));
347 o.insert("parameters".into(), kvp_to_json_msg(&m.parameters));
348 o
349 }
350 ControlMessage::PublishOk(m) => {
351 let mut o = Map::new();
352 o.insert("request_id".into(), vi(m.request_id.into_inner()));
353 o.insert("forward".into(), vi(m.forward as u64));
354 o.insert("subscriber_priority".into(), vi(m.subscriber_priority as u64));
355 o.insert("group_order".into(), vi(m.group_order as u64));
356 o.insert("filter_type".into(), vi(m.filter_type.into_inner()));
357 if let Some(sg) = &m.start_group {
358 o.insert("start_group".into(), vi(sg.into_inner()));
359 }
360 if let Some(so) = &m.start_object {
361 o.insert("start_object".into(), vi(so.into_inner()));
362 }
363 if let Some(eg) = &m.end_group {
364 o.insert("end_group".into(), vi(eg.into_inner()));
365 }
366 o.insert("parameters".into(), kvp_to_json_msg(&m.parameters));
367 o
368 }
369 ControlMessage::PublishError(m) => {
370 let mut o = Map::new();
371 o.insert("request_id".into(), vi(m.request_id.into_inner()));
372 o.insert("error_code".into(), vi(m.error_code.into_inner()));
373 o.insert(
374 "reason_phrase".into(),
375 Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
376 );
377 o
378 }
379 };
380 obj
381}