1use anyhow::{format_err, Error};
6use fidl::endpoints::{ProtocolMarker, ServerEnd};
7use fuchsia_component::client::connect_to_protocol_at_path;
8use lazy_static::lazy_static;
9use std::collections::VecDeque;
10use thiserror::Error;
11use {fidl_fuchsia_component as fcomponent, fidl_fuchsia_io as fio};
12
13lazy_static! {
14 pub static ref START_COMPONENT_TREE_STREAM: String = "StartComponentTree".into();
17}
18
19pub fn event_name(event_type: &fcomponent::EventType) -> String {
21 match event_type {
22 fcomponent::EventType::CapabilityRequested => "capability_requested",
23 fcomponent::EventType::Discovered => unreachable!("This isn't used anymore"),
24 fcomponent::EventType::Destroyed => "destroyed",
25 fcomponent::EventType::Resolved => "resolved",
26 fcomponent::EventType::Unresolved => "unresolved",
27 fcomponent::EventType::Started => "started",
28 fcomponent::EventType::Stopped => "stopped",
29 fcomponent::EventType::DebugStarted => "debug_started",
30 #[cfg(fuchsia_api_level_at_least = "HEAD")]
31 fcomponent::EventType::DirectoryReady => unreachable!("This isn't used anymore"),
32 }
33 .to_string()
34}
35
36enum InternalStream {
37 New(fcomponent::EventStreamProxy),
38}
39
40pub struct EventStream {
41 stream: InternalStream,
42 buffer: VecDeque<fcomponent::Event>,
43}
44
45#[derive(Debug, Error, Clone)]
46pub enum EventStreamError {
47 #[error("Stream terminated unexpectedly")]
48 StreamClosed,
49}
50
51impl EventStream {
52 pub fn new(stream: fcomponent::EventStreamProxy) -> Self {
53 Self { stream: InternalStream::New(stream), buffer: VecDeque::new() }
54 }
55
56 pub fn open_at_path_pipelined(path: impl Into<String>) -> Result<Self, Error> {
57 Ok(Self::new(connect_to_protocol_at_path::<fcomponent::EventStreamMarker>(&path.into())?))
58 }
59
60 pub async fn open_at_path(path: impl Into<String>) -> Result<Self, Error> {
61 let event_stream =
62 connect_to_protocol_at_path::<fcomponent::EventStreamMarker>(&path.into())?;
63 event_stream.wait_for_ready().await?;
64 Ok(Self::new(event_stream))
65 }
66
67 pub async fn open() -> Result<Self, Error> {
68 let event_stream = connect_to_protocol_at_path::<fcomponent::EventStreamMarker>(
69 "/svc/fuchsia.component.EventStream",
70 )?;
71 event_stream.wait_for_ready().await?;
72 Ok(Self::new(event_stream))
73 }
74
75 pub fn open_pipelined() -> Result<Self, Error> {
76 Ok(Self::new(connect_to_protocol_at_path::<fcomponent::EventStreamMarker>(
77 "/svc/fuchsia.component.EventStream",
78 )?))
79 }
80
81 pub async fn next(&mut self) -> Result<fcomponent::Event, EventStreamError> {
82 if let Some(event) = self.buffer.pop_front() {
83 return Ok(event);
84 }
85 match &mut self.stream {
86 InternalStream::New(stream) => {
87 match stream.get_next().await {
88 Ok(events) => {
89 let mut iter = events.into_iter();
90 if let Some(real_event) = iter.next() {
91 let ret = real_event;
92 while let Some(value) = iter.next() {
93 self.buffer.push_back(value);
94 }
95 return Ok(ret);
96 } else {
97 Err(EventStreamError::StreamClosed)
100 }
101 }
102 Err(_) => Err(EventStreamError::StreamClosed),
103 }
104 }
105 }
106 }
107}
108
109pub trait Event: TryFrom<fcomponent::Event, Error = anyhow::Error> {
111 const TYPE: fcomponent::EventType;
112 const NAME: &'static str;
113
114 fn target_moniker(&self) -> &str;
115 fn component_url(&self) -> &str;
116 fn timestamp(&self) -> zx::BootInstant;
117 fn is_ok(&self) -> bool;
118 fn is_err(&self) -> bool;
119}
120
121#[derive(Copy, Debug, PartialEq, Eq, Clone, Ord, PartialOrd)]
122pub enum ExitStatus {
125 Clean,
126 Crash(i32),
127}
128
129impl From<i32> for ExitStatus {
130 fn from(exit_status: i32) -> Self {
131 match exit_status {
132 0 => ExitStatus::Clean,
133 _ => ExitStatus::Crash(exit_status),
134 }
135 }
136}
137
138#[derive(Debug)]
139struct EventHeader {
140 event_type: fcomponent::EventType,
141 component_url: String,
142 moniker: String,
143 timestamp: zx::BootInstant,
144}
145
146impl TryFrom<fcomponent::EventHeader> for EventHeader {
147 type Error = anyhow::Error;
148
149 fn try_from(header: fcomponent::EventHeader) -> Result<Self, Self::Error> {
150 let event_type = header.event_type.ok_or_else(|| format_err!("No event type"))?;
151 let component_url = header.component_url.ok_or_else(|| format_err!("No component url"))?;
152 let moniker = header.moniker.ok_or_else(|| format_err!("No moniker"))?;
153 let timestamp = header
154 .timestamp
155 .ok_or_else(|| format_err!("Missing timestamp from the Event object"))?;
156 Ok(EventHeader { event_type, component_url, moniker, timestamp })
157 }
158}
159
160#[derive(Debug, PartialEq, Eq)]
161pub struct EventError {
162 pub description: String,
163}
164
165macro_rules! create_event {
187 (
189 event_type: $event_type:ident,
190 event_name: $event_name:ident,
191 payload: {
192 data: {$(
193 {
194 name: $data_name:ident,
195 ty: $data_ty:ty,
196 }
197 )*},
198 client_protocols: {$(
199 {
200 name: $client_protocol_name:ident,
201 ty: $client_protocol_ty:ty,
202 }
203 )*},
204 server_protocols: {$(
205 {
206 name: $server_protocol_name:ident,
207 }
208 )*},
209 },
210 error_payload: {
211 $(
212 {
213 name: $error_data_name:ident,
214 ty: $error_data_ty:ty,
215 }
216 )*
217 }
218 ) => {
219 paste::paste! {
220 #[derive(Debug)]
221 pub struct [<$event_type Payload>] {
222 $(pub $client_protocol_name: $client_protocol_ty,)*
223 $(pub $server_protocol_name: Option<zx::Channel>,)*
224 $(pub $data_name: $data_ty,)*
225 }
226
227 #[derive(Debug)]
228 pub struct [<$event_type Error>] {
229 $(pub $error_data_name: $error_data_ty,)*
230 pub description: String,
231 }
232
233 #[derive(Debug)]
234 pub struct $event_type {
235 header: EventHeader,
236 result: Result<[<$event_type Payload>], [<$event_type Error>]>,
237 }
238
239 impl $event_type {
240 pub fn result<'a>(&'a self) -> Result<&'a [<$event_type Payload>], &'a [<$event_type Error>]> {
241 self.result.as_ref()
242 }
243
244 $(
245 pub fn [<take_ $server_protocol_name>]<T: ProtocolMarker>(&mut self)
246 -> Option<T::RequestStream> {
247 self.result.as_mut()
248 .ok()
249 .and_then(|payload| payload.$server_protocol_name.take())
250 .map(|channel| {
251 let server_end = ServerEnd::<T>::new(channel);
252 server_end.into_stream()
253 })
254 }
255 )*
256 }
257
258 impl Event for $event_type {
259 const TYPE: fcomponent::EventType = fcomponent::EventType::$event_type;
260 const NAME: &'static str = stringify!($event_name);
261
262 fn target_moniker(&self) -> &str {
263 &self.header.moniker
264 }
265
266 fn component_url(&self) -> &str {
267 &self.header.component_url
268 }
269
270 fn timestamp(&self) -> zx::BootInstant {
271 self.header.timestamp
272 }
273
274 fn is_ok(&self) -> bool {
275 self.result.is_ok()
276 }
277
278 fn is_err(&self) -> bool {
279 self.result.is_err()
280 }
281 }
282
283 impl TryFrom<fcomponent::Event> for $event_type {
284 type Error = anyhow::Error;
285
286 fn try_from(event: fcomponent::Event) -> Result<Self, Self::Error> {
287 let result = match event.payload {
289 Some(payload) => {
290 #[allow(unused)]
293 let payload = match payload {
294 fcomponent::EventPayload::$event_type(payload) => Ok(payload),
295 _ => Err(format_err!("Incorrect payload type, {:?}", payload)),
296 }?;
297
298 $(
300 let $data_name: $data_ty = payload.$data_name.coerce().ok_or(
301 format_err!("Missing {} from {} object",
302 stringify!($data_name), stringify!($event_type))
303 )?;
304 )*
305
306 $(
308 let $client_protocol_name: $client_protocol_ty = payload.$client_protocol_name.ok_or(
309 format_err!("Missing {} from {} object",
310 stringify!($client_protocol_name), stringify!($event_type))
311 )?.into_proxy();
312 )*
313 $(
314 let $server_protocol_name: Option<zx::Channel> =
315 Some(payload.$server_protocol_name.ok_or(
316 format_err!("Missing {} from {} object",
317 stringify!($server_protocol_name), stringify!($event_type))
318 )?);
319 )*
320
321 #[allow(dead_code)]
322 let payload = paste::paste! {
323 [<$event_type Payload>] {
324 $($data_name,)*
325 $($client_protocol_name,)*
326 $($server_protocol_name,)*
327 }
328 };
329
330 Ok(Ok(payload))
331 },
332 None => Err(format_err!("Missing event_result from Event object")),
333 }?;
334
335 let event = {
336 let header = event.header
337 .ok_or(format_err!("Missing Event header"))
338 .and_then(|header| EventHeader::try_from(header))?;
339
340 if header.event_type != Self::TYPE {
341 return Err(format_err!("Incorrect event type"));
342 }
343
344 $event_type { header, result }
345 };
346 Ok(event)
347 }
348 }
349 }
350 };
351 ($event_type:ident, $event_name:ident) => {
352 create_event!(event_type: $event_type, event_name: $event_name,
353 payload: {
354 data: {},
355 client_protocols: {},
356 server_protocols: {},
357 },
358 error_payload: {});
359 };
360}
361
362create_event!(Destroyed, destroyed);
364create_event!(Resolved, resolved);
365create_event!(Unresolved, unresolved);
366create_event!(Started, started);
367create_event!(
368 event_type: Stopped,
369 event_name: stopped,
370 payload: {
371 data: {
372 {
373 name: status,
374 ty: ExitStatus,
375 }
376 {
377 name: exit_code,
378 ty: Option<i64>,
379 }
380 },
381 client_protocols: {},
382 server_protocols: {},
383 },
384 error_payload: {}
385);
386create_event!(
387 event_type: CapabilityRequested,
388 event_name: capability_requested,
389 payload: {
390 data: {
391 {
392 name: name,
393 ty: String,
394 }
395 },
396 client_protocols: {},
397 server_protocols: {
398 {
399 name: capability,
400 }
401 },
402 },
403 error_payload: {
404 {
405 name: name,
406 ty: String,
407 }
408 }
409);
410create_event!(
411 event_type: DebugStarted,
412 event_name: debug_started,
413 payload: {
414 data: {
415 {
416 name: break_on_start,
417 ty: zx::EventPair,
418 }
419 },
420 client_protocols: {
421 {
422 name: runtime_dir,
423 ty: fio::DirectoryProxy,
424 }
425 },
426 server_protocols: {},
427 },
428 error_payload: {}
429);
430
431trait Coerce<T> {
432 fn coerce(self) -> Option<T>;
433}
434
435impl<T> Coerce<T> for Option<T> {
436 fn coerce(self) -> Option<T> {
437 self
438 }
439}
440
441impl<T> Coerce<Option<T>> for Option<T> {
442 fn coerce(self) -> Option<Option<T>> {
443 Some(self)
444 }
445}
446
447impl Coerce<ExitStatus> for Option<i32> {
448 fn coerce(self) -> Option<ExitStatus> {
449 self.map(Into::into)
450 }
451}