1use crate::draft08::message::ControlMessage;
2use crate::fields::{FieldMap as Map, FieldValue as Value};
3use crate::types::*;
4
5use crate::fields::params::{kvp_to_json_d07, kvp_to_json_d08_setup};
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
17pub fn message_fields(msg: &ControlMessage) -> Map {
23 let obj = match msg {
24 ControlMessage::ClientSetup(m) => {
25 let mut o = Map::new();
26 o.insert(
27 "supported_versions".into(),
28 Value::Array(m.supported_versions.iter().map(|v| vi(v.into_inner())).collect()),
29 );
30 o.insert("parameters".into(), kvp_to_json_d08_setup(&m.parameters));
31 o
32 }
33 ControlMessage::ServerSetup(m) => {
34 let mut o = Map::new();
35 o.insert("selected_version".into(), vi(m.selected_version.into_inner()));
36 o.insert("parameters".into(), kvp_to_json_d08_setup(&m.parameters));
37 o
38 }
39 ControlMessage::GoAway(m) => {
40 let mut o = Map::new();
41 o.insert(
42 "new_session_uri".into(),
43 Value::Text(String::from_utf8_lossy(&m.new_session_uri).into_owned()),
44 );
45 o
46 }
47 ControlMessage::MaxSubscribeId(m) => {
48 let mut o = Map::new();
49 o.insert("subscribe_id".into(), vi(m.subscribe_id.into_inner()));
50 o
51 }
52 ControlMessage::SubscribesBlocked(m) => {
53 let mut o = Map::new();
54 o.insert("maximum_subscribe_id".into(), vi(m.maximum_subscribe_id.into_inner()));
55 o
56 }
57 ControlMessage::Subscribe(m) => {
58 let mut o = Map::new();
59 o.insert("subscribe_id".into(), vi(m.subscribe_id.into_inner()));
60 o.insert("track_alias".into(), vi(m.track_alias.into_inner()));
61 o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
62 o.insert(
63 "track_name".into(),
64 Value::Text(String::from_utf8_lossy(&m.track_name).into_owned()),
65 );
66 o.insert("subscriber_priority".into(), vi(m.subscriber_priority as u64));
67 o.insert("group_order".into(), vi(m.group_order as u64));
68 o.insert("filter_type".into(), vi(m.filter_type as u64));
69 if let Some(loc) = &m.start_location {
70 o.insert("start_group".into(), vi(loc.group.into_inner()));
71 o.insert("start_object".into(), vi(loc.object.into_inner()));
72 }
73 if let Some(eg) = &m.end_group {
74 o.insert("end_group".into(), vi(eg.into_inner()));
75 }
76 o.insert("parameters".into(), kvp_to_json_d07(&m.parameters));
77 o
78 }
79 ControlMessage::SubscribeOk(m) => {
80 let mut o = Map::new();
81 o.insert("subscribe_id".into(), vi(m.subscribe_id.into_inner()));
82 o.insert("expires".into(), vi(m.expires.into_inner()));
83 o.insert("group_order".into(), vi(m.group_order as u64));
84 o.insert("content_exists".into(), vi(m.content_exists as u64));
85 if let Some(gid) = &m.largest_group_id {
86 o.insert("largest_group_id".into(), vi(gid.into_inner()));
87 }
88 if let Some(oid) = &m.largest_object_id {
89 o.insert("largest_object_id".into(), vi(oid.into_inner()));
90 }
91 o.insert("parameters".into(), kvp_to_json_d07(&m.parameters));
92 o
93 }
94 ControlMessage::SubscribeError(m) => {
95 let mut o = Map::new();
96 o.insert("subscribe_id".into(), vi(m.subscribe_id.into_inner()));
97 o.insert("error_code".into(), vi(m.error_code.into_inner()));
98 o.insert(
99 "reason_phrase".into(),
100 Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
101 );
102 o.insert("track_alias".into(), vi(m.track_alias.into_inner()));
103 o
104 }
105 ControlMessage::SubscribeUpdate(m) => {
106 let mut o = Map::new();
107 o.insert("subscribe_id".into(), vi(m.subscribe_id.into_inner()));
108 o.insert("start_group".into(), vi(m.start_group.into_inner()));
109 o.insert("start_object".into(), vi(m.start_object.into_inner()));
110 o.insert("end_group".into(), vi(m.end_group.into_inner()));
111 o.insert("subscriber_priority".into(), vi(m.subscriber_priority as u64));
112 o.insert("parameters".into(), kvp_to_json_d07(&m.parameters));
113 o
114 }
115 ControlMessage::SubscribeDone(m) => {
116 let mut o = Map::new();
117 o.insert("subscribe_id".into(), vi(m.subscribe_id.into_inner()));
118 o.insert("status_code".into(), vi(m.status_code.into_inner()));
119 o.insert("stream_count".into(), vi(m.stream_count.into_inner()));
120 o.insert(
121 "reason_phrase".into(),
122 Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
123 );
124 o
125 }
126 ControlMessage::Unsubscribe(m) => {
127 let mut o = Map::new();
128 o.insert("subscribe_id".into(), vi(m.subscribe_id.into_inner()));
129 o
130 }
131 ControlMessage::Announce(m) => {
132 let mut o = Map::new();
133 o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
134 o.insert("parameters".into(), kvp_to_json_d07(&m.parameters));
135 o
136 }
137 ControlMessage::AnnounceOk(m) => {
138 let mut o = Map::new();
139 o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
140 o
141 }
142 ControlMessage::AnnounceError(m) => {
143 let mut o = Map::new();
144 o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
145 o.insert("error_code".into(), vi(m.error_code.into_inner()));
146 o.insert(
147 "reason_phrase".into(),
148 Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
149 );
150 o
151 }
152 ControlMessage::AnnounceCancel(m) => {
153 let mut o = Map::new();
154 o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
155 o.insert("error_code".into(), vi(m.error_code.into_inner()));
156 o.insert(
157 "reason_phrase".into(),
158 Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
159 );
160 o
161 }
162 ControlMessage::Unannounce(m) => {
163 let mut o = Map::new();
164 o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
165 o
166 }
167 ControlMessage::SubscribeAnnounces(m) => {
168 let mut o = Map::new();
169 o.insert("track_namespace_prefix".into(), ns_to_json(&m.track_namespace_prefix));
170 o.insert("parameters".into(), kvp_to_json_d07(&m.parameters));
171 o
172 }
173 ControlMessage::SubscribeAnnouncesOk(m) => {
174 let mut o = Map::new();
175 o.insert("track_namespace_prefix".into(), ns_to_json(&m.track_namespace_prefix));
176 o
177 }
178 ControlMessage::SubscribeAnnouncesError(m) => {
179 let mut o = Map::new();
180 o.insert("track_namespace_prefix".into(), ns_to_json(&m.track_namespace_prefix));
181 o.insert("error_code".into(), vi(m.error_code.into_inner()));
182 o.insert(
183 "reason_phrase".into(),
184 Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
185 );
186 o
187 }
188 ControlMessage::UnsubscribeAnnounces(m) => {
189 let mut o = Map::new();
190 o.insert("track_namespace_prefix".into(), ns_to_json(&m.track_namespace_prefix));
191 o
192 }
193 ControlMessage::TrackStatusRequest(m) => {
194 let mut o = Map::new();
195 o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
196 o.insert(
197 "track_name".into(),
198 Value::Text(String::from_utf8_lossy(&m.track_name).into_owned()),
199 );
200 o
201 }
202 ControlMessage::TrackStatus(m) => {
203 let mut o = Map::new();
204 o.insert("track_namespace".into(), ns_to_json(&m.track_namespace));
205 o.insert(
206 "track_name".into(),
207 Value::Text(String::from_utf8_lossy(&m.track_name).into_owned()),
208 );
209 o.insert("status_code".into(), vi(m.status_code.into_inner()));
210 o.insert("last_group_id".into(), vi(m.last_group_id.into_inner()));
211 o.insert("last_object_id".into(), vi(m.last_object_id.into_inner()));
212 o
213 }
214 ControlMessage::Fetch(m) => {
215 let mut o = Map::new();
216 o.insert("subscribe_id".into(), vi(m.subscribe_id.into_inner()));
217 o.insert("subscriber_priority".into(), vi(m.subscriber_priority as u64));
218 o.insert("group_order".into(), vi(m.group_order as u64));
219 o.insert("fetch_type".into(), vi(m.fetch_type as u64));
220 if let Some(ns) = &m.track_namespace {
221 o.insert("track_namespace".into(), ns_to_json(ns));
222 }
223 if let Some(name) = &m.track_name {
224 o.insert(
225 "track_name".into(),
226 Value::Text(String::from_utf8_lossy(name).into_owned()),
227 );
228 }
229 if let Some(sg) = &m.start_group {
230 o.insert("start_group".into(), vi(sg.into_inner()));
231 }
232 if let Some(so) = &m.start_object {
233 o.insert("start_object".into(), vi(so.into_inner()));
234 }
235 if let Some(eg) = &m.end_group {
236 o.insert("end_group".into(), vi(eg.into_inner()));
237 }
238 if let Some(eo) = &m.end_object {
239 o.insert("end_object".into(), vi(eo.into_inner()));
240 }
241 if let Some(jsi) = &m.joining_subscribe_id {
242 o.insert("joining_subscribe_id".into(), vi(jsi.into_inner()));
243 }
244 if let Some(pgo) = &m.preceding_group_offset {
245 o.insert("preceding_group_offset".into(), vi(pgo.into_inner()));
246 }
247 o.insert("parameters".into(), kvp_to_json_d07(&m.parameters));
248 o
249 }
250 ControlMessage::FetchOk(m) => {
251 let mut o = Map::new();
252 o.insert("subscribe_id".into(), vi(m.subscribe_id.into_inner()));
253 o.insert("group_order".into(), vi(m.group_order as u64));
254 o.insert("end_of_track".into(), vi(m.end_of_track as u64));
255 o.insert("largest_group_id".into(), vi(m.largest_group_id.into_inner()));
256 o.insert("largest_object_id".into(), vi(m.largest_object_id.into_inner()));
257 o.insert("parameters".into(), kvp_to_json_d07(&m.parameters));
258 o
259 }
260 ControlMessage::FetchError(m) => {
261 let mut o = Map::new();
262 o.insert("subscribe_id".into(), vi(m.subscribe_id.into_inner()));
263 o.insert("error_code".into(), vi(m.error_code.into_inner()));
264 o.insert(
265 "reason_phrase".into(),
266 Value::Text(String::from_utf8_lossy(&m.reason_phrase).into_owned()),
267 );
268 o
269 }
270 ControlMessage::FetchCancel(m) => {
271 let mut o = Map::new();
272 o.insert("subscribe_id".into(), vi(m.subscribe_id.into_inner()));
273 o
274 }
275 };
276 obj
277}