fidl_fuchsia_net_neighbor_ext_fdomain/
lib.rs

1// Copyright 2020 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5//! Extensions for types in the `fidl_fuchsia_net_neighbor` crate.
6
7#![deny(missing_docs)]
8
9#[cfg(any(test, feature = "testutils"))]
10pub mod testutil;
11
12use async_utils::{fold, stream};
13use fidl_fuchsia_net_ext as fnet_ext;
14use fidl_table_validation::*;
15use flex_client::ProxyHasDomain;
16use flex_fuchsia_net as fnet;
17use flex_fuchsia_net_neighbor as fnet_neighbor;
18use futures::{Stream, TryStreamExt as _};
19use std::num::NonZeroU64;
20use thiserror::Error;
21use zx_types as zx;
22
23#[derive(Debug, Error)]
24enum ConversionError {
25    #[error("interface ID must be non-zero")]
26    ZeroInterface,
27}
28
29struct NonZeroU64Converter;
30
31impl Converter for NonZeroU64Converter {
32    type Fidl = u64;
33    type Validated = NonZeroU64;
34    type Error = ConversionError;
35    fn try_from_fidl(value: Self::Fidl) -> std::result::Result<Self::Validated, Self::Error> {
36        NonZeroU64::new(value).ok_or(ConversionError::ZeroInterface)
37    }
38    fn from_validated(validated: Self::Validated) -> Self::Fidl {
39        validated.get()
40    }
41}
42
43/// Information on a neighboring device in the local network.
44#[derive(Clone, Debug, Eq, Hash, PartialEq, ValidFidlTable)]
45#[fidl_table_src(fnet_neighbor::Entry)]
46#[fidl_table_strict]
47pub struct Entry {
48    /// Identifier for the interface used for communicating with the neighbor.
49    #[fidl_field_type(converter = NonZeroU64Converter)]
50    pub interface: NonZeroU64,
51    /// IP address of the neighbor.
52    pub neighbor: fnet::IpAddress,
53    /// State of the entry within the Neighbor Unreachability Detection (NUD)
54    /// state machine.
55    pub state: fnet_neighbor::EntryState,
56    /// MAC address of the neighboring device's network interface controller.
57    #[fidl_field_type(optional)]
58    pub mac: Option<fnet::MacAddress>,
59    /// Timestamp when this entry has changed `state`.
60    // TODO(https://fxbug.dev/42155335): Replace with zx::MonotonicInstant once there is
61    // support for custom conversion functions.
62    pub updated_at: zx::zx_time_t,
63}
64
65/// Returns a &str suitable for display representing the EntryState parameter.
66pub fn display_entry_state(state: &fnet_neighbor::EntryState) -> &'static str {
67    match state {
68        fnet_neighbor::EntryState::Incomplete => "INCOMPLETE",
69        fnet_neighbor::EntryState::Reachable => "REACHABLE",
70        fnet_neighbor::EntryState::Stale => "STALE",
71        fnet_neighbor::EntryState::Delay => "DELAY",
72        fnet_neighbor::EntryState::Probe => "PROBE",
73        fnet_neighbor::EntryState::Static => "STATIC",
74        fnet_neighbor::EntryState::Unreachable => "UNREACHABLE",
75    }
76}
77
78impl std::fmt::Display for Entry {
79    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> Result<(), std::fmt::Error> {
80        let Self { interface, neighbor, mac, state, updated_at: _ } = self;
81        write!(f, "Interface {} | IP {} | MAC ", interface, fnet_ext::IpAddress::from(*neighbor))?;
82        if let Some(mac) = mac {
83            write!(f, "{}", fnet_ext::MacAddress::from(*mac))?;
84        } else {
85            write!(f, "?")?;
86        }
87        write!(f, " | {}", display_entry_state(state))
88    }
89}
90
91/// Neighbor entry events.
92#[derive(Clone, Debug, PartialEq)]
93pub enum Event {
94    /// A neighbor entry that existed prior to watching.
95    Existing(Entry),
96    /// A neighbor entry that was added.
97    Added(Entry),
98    /// A neighbor entry that was changed.
99    Changed(Entry),
100    /// A neighbor entry that was removed.
101    Removed(Entry),
102    /// Sentinel value indicating that all existing entries have been sent.
103    Idle,
104}
105
106impl TryFrom<fnet_neighbor::EntryIteratorItem> for Event {
107    type Error = EntryValidationError;
108
109    fn try_from(value: fnet_neighbor::EntryIteratorItem) -> Result<Self, Self::Error> {
110        match value {
111            fnet_neighbor::EntryIteratorItem::Existing(e) => {
112                Ok(Event::Existing(Entry::try_from(e)?))
113            }
114            fnet_neighbor::EntryIteratorItem::Added(e) => Ok(Event::Added(Entry::try_from(e)?)),
115            fnet_neighbor::EntryIteratorItem::Changed(e) => Ok(Event::Changed(Entry::try_from(e)?)),
116            fnet_neighbor::EntryIteratorItem::Removed(e) => Ok(Event::Removed(Entry::try_from(e)?)),
117            fnet_neighbor::EntryIteratorItem::Idle(_) => Ok(Event::Idle),
118        }
119    }
120}
121
122/// Options for modifying the behavior of `EntryIterator`.
123#[derive(Clone, Debug, Default, Eq, PartialEq, ValidFidlTable)]
124#[fidl_table_src(fnet_neighbor::EntryIteratorOptions)]
125#[fidl_table_strict]
126pub struct EntryIteratorOptions {}
127
128/// Neighbor table entry iterator creation error.
129#[derive(Clone, Debug, Error)]
130#[error("failed to open neighbor entry iterator: {0}")]
131pub struct OpenEntryIteratorError(fidl::Error);
132
133/// Dispatches `OpenEntryIterator` on the view proxy.
134pub fn open_entry_iterator(
135    view_proxy: &fnet_neighbor::ViewProxy,
136    options: EntryIteratorOptions,
137) -> Result<fnet_neighbor::EntryIteratorProxy, OpenEntryIteratorError> {
138    let (neighbor_iter_proxy, entry_iter_server_end) =
139        view_proxy.domain().create_proxy::<fnet_neighbor::EntryIteratorMarker>();
140
141    view_proxy
142        .open_entry_iterator(entry_iter_server_end, &options.into())
143        .map_err(OpenEntryIteratorError)?;
144
145    Ok(neighbor_iter_proxy)
146}
147
148/// Neighbor table entry iterator `GetNext` errors.
149#[derive(Debug, Error)]
150pub enum EntryIteratorError {
151    /// The call to `GetNext` returned a FIDL error.
152    #[error("the call to `GetNext()` failed: {0}")]
153    Fidl(fidl::Error),
154    /// The event returned by `GetNext` encountered a conversion error.
155    #[error("failed to convert event returned by `GetNext()`: {0:?}")]
156    Conversion(EntryValidationError),
157    /// The server returned an empty batch of events.
158    #[error("the call to `GetNext()` returned an empty batch of events")]
159    EmptyEventBatch,
160}
161
162/// [`event_stream_from_view_with_options`] with default [`EntryIteratorOptions`].
163pub fn event_stream_from_view(
164    neighbors_view: &fnet_neighbor::ViewProxy,
165) -> Result<impl Stream<Item = Result<Event, EntryIteratorError>> + 'static, OpenEntryIteratorError>
166{
167    event_stream_from_view_with_options(neighbors_view, Default::default())
168}
169
170/// Connects to the neighbor table entry iterator protocol with
171/// [`EntryIteratorOptions`] and converts the Hanging-Get style API into an
172/// Event stream.
173///
174/// Each call to `GetNext` returns a batch of events, which are flattened into a
175/// single stream. If an error is encountered while calling `GetNext` or while
176/// converting the event, the stream is immediately terminated.
177pub fn event_stream_from_view_with_options(
178    neighbors_view: &fnet_neighbor::ViewProxy,
179    options: EntryIteratorOptions,
180) -> Result<impl Stream<Item = Result<Event, EntryIteratorError>> + 'static, OpenEntryIteratorError>
181{
182    let neighbor_iterator = open_entry_iterator(neighbors_view, options)?;
183    Ok(event_stream_from_iterator(neighbor_iterator))
184}
185
186/// Turns the provided neighbor table entry iterator into an [`Event`] stream by applying
187/// Hanging-Get watch.
188///
189/// Each call to `GetNext` returns a batch of events, which are flattened into a
190/// single stream. If an error is encountered while calling `GetNext` or while
191/// converting the event, the stream is immediately terminated.
192pub fn event_stream_from_iterator(
193    neighbor_iterator: fnet_neighbor::EntryIteratorProxy,
194) -> impl Stream<Item = Result<Event, EntryIteratorError>> {
195    stream::ShortCircuit::new(
196        futures::stream::try_unfold(neighbor_iterator, |iter| async {
197            let events_batch = iter.get_next().await.map_err(EntryIteratorError::Fidl)?;
198            if events_batch.is_empty() {
199                return Err(EntryIteratorError::EmptyEventBatch);
200            }
201            let events_batch = events_batch
202                .into_iter()
203                .map(|event| event.try_into().map_err(EntryIteratorError::Conversion));
204            let event_stream = futures::stream::iter(events_batch);
205            Ok(Some((event_stream, iter)))
206        })
207        // Flatten the stream of event streams into a single event stream.
208        .try_flatten(),
209    )
210}
211
212/// Errors returned by [`collect_neighbors_until_idle`].
213#[derive(Debug, Error)]
214pub enum CollectNeighborsUntilIdleError {
215    /// There was an error in the event stream.
216    #[error("there was an error in the event stream: {0}")]
217    ErrorInStream(EntryIteratorError),
218    /// There was an unexpected event in the event stream. Only `existing` or
219    /// `idle` events are expected.
220    #[error("there was an unexpected event in the event stream: {0:?}")]
221    UnexpectedEvent(Event),
222    /// The event stream unexpectedly ended.
223    #[error("the event stream unexpectedly ended")]
224    StreamEnded,
225}
226
227/// Collects all `existing` events from the stream, stopping once the `idle`
228/// event is observed.
229pub async fn collect_neighbors_until_idle<C: Extend<Entry> + Default>(
230    event_stream: impl futures::Stream<Item = Result<Event, EntryIteratorError>> + Unpin,
231) -> Result<C, CollectNeighborsUntilIdleError> {
232    fold::fold_while(
233        event_stream,
234        Ok(C::default()),
235        |existing_neighbors: Result<C, CollectNeighborsUntilIdleError>, event| {
236            futures::future::ready(match existing_neighbors {
237                Err(_) => {
238                    unreachable!(
239                        "`existing_neighbors` must be `Ok`, because we stop folding on err"
240                    )
241                }
242                Ok(mut existing_neighbors) => match event {
243                    Err(e) => {
244                        fold::FoldWhile::Done(Err(CollectNeighborsUntilIdleError::ErrorInStream(e)))
245                    }
246                    Ok(e) => match e {
247                        Event::Existing(e) => {
248                            existing_neighbors.extend([e]);
249                            fold::FoldWhile::Continue(Ok(existing_neighbors))
250                        }
251                        Event::Idle => fold::FoldWhile::Done(Ok(existing_neighbors)),
252                        e @ Event::Added(_) | e @ Event::Changed(_) | e @ Event::Removed(_) => {
253                            fold::FoldWhile::Done(Err(
254                                CollectNeighborsUntilIdleError::UnexpectedEvent(e),
255                            ))
256                        }
257                    },
258                },
259            })
260        },
261    )
262    .await
263    .short_circuited()
264    .map_err(|_accumulated_thus_far: Result<C, CollectNeighborsUntilIdleError>| {
265        CollectNeighborsUntilIdleError::StreamEnded
266    })?
267}
268
269#[cfg(test)]
270mod tests {
271    use super::*;
272
273    use assert_matches::assert_matches;
274    use futures::{FutureExt, StreamExt as _};
275    use test_case::test_case;
276
277    fn valid_fidl_entry(interface: NonZeroU64) -> fnet_neighbor::Entry {
278        fnet_neighbor::Entry {
279            interface: Some(interface.get()),
280            neighbor: Some(fnet::IpAddress::Ipv4(fnet::Ipv4Address { addr: [192, 168, 0, 1] })),
281            state: Some(fnet_neighbor::EntryState::Reachable),
282            mac: Some(fnet::MacAddress { octets: [0, 1, 2, 3, 4, 5] }),
283            updated_at: Some(123456),
284            ..Default::default()
285        }
286    }
287
288    #[test]
289    fn event_try_from_success() {
290        let fidl_entry = valid_fidl_entry(NonZeroU64::new(1).unwrap());
291        let local_entry: Entry = fidl_entry.clone().try_into().unwrap();
292        assert_matches!(
293            fnet_neighbor::EntryIteratorItem::Existing(fidl_entry.clone()).try_into(),
294            Ok(Event::Existing(entry)) if entry == local_entry
295        );
296        assert_matches!(
297            fnet_neighbor::EntryIteratorItem::Added(fidl_entry.clone()).try_into(),
298            Ok(Event::Added(entry)) if entry == local_entry
299        );
300        assert_matches!(
301            fnet_neighbor::EntryIteratorItem::Changed(fidl_entry.clone()).try_into(),
302            Ok(Event::Changed(entry)) if entry == local_entry
303        );
304        assert_matches!(
305            fnet_neighbor::EntryIteratorItem::Removed(fidl_entry.clone()).try_into(),
306            Ok(Event::Removed(entry)) if entry == local_entry
307        );
308        assert_matches!(
309            fnet_neighbor::EntryIteratorItem::Idle(fnet_neighbor::IdleEvent).try_into(),
310            Ok(Event::Idle)
311        );
312    }
313
314    #[test]
315    fn event_try_from_missing_field() {
316        let fidl_entry = fnet_neighbor::Entry {
317            interface: None, // Required field omitted.
318            ..valid_fidl_entry(NonZeroU64::new(1).unwrap())
319        };
320        assert_matches!(
321            Event::try_from(fnet_neighbor::EntryIteratorItem::Existing(fidl_entry.clone())),
322            Err(EntryValidationError::MissingField(_))
323        );
324        assert_matches!(
325            Event::try_from(fnet_neighbor::EntryIteratorItem::Added(fidl_entry.clone())),
326            Err(EntryValidationError::MissingField(_))
327        );
328        assert_matches!(
329            Event::try_from(fnet_neighbor::EntryIteratorItem::Changed(fidl_entry.clone())),
330            Err(EntryValidationError::MissingField(_))
331        );
332        assert_matches!(
333            Event::try_from(fnet_neighbor::EntryIteratorItem::Removed(fidl_entry.clone())),
334            Err(EntryValidationError::MissingField(_))
335        );
336    }
337
338    #[test]
339    fn entry_try_from_zero_interface() {
340        let fidl_entry = fnet_neighbor::Entry {
341            interface: Some(0),
342            ..valid_fidl_entry(NonZeroU64::new(1).unwrap())
343        };
344        assert_matches!(Entry::try_from(fidl_entry), Err(EntryValidationError::InvalidField(_)));
345    }
346
347    // Tests `event_stream_from_view` with various "shapes". The test
348    // parameter is a vec of ranges, where each range corresponds to the batch
349    // of events that will be sent in response to a single call to `GetNext().
350    #[test_case(Vec::new(); "no events")]
351    #[test_case(vec![0..1]; "single_batch_single_event")]
352    #[test_case(vec![0..10]; "single_batch_many_events")]
353    #[test_case(vec![0..10, 10..20, 20..30]; "many_batches_many_events")]
354    #[fuchsia::test]
355    async fn event_stream_from_view_against_shape(test_shape: Vec<std::ops::Range<u8>>) {
356        // Build the event stream based on the `test_shape`. Use a channel
357        // so that the stream stays open until `close_channel` is called later.
358        let (batches_sender, batches_receiver) =
359            futures::channel::mpsc::unbounded::<Vec<fnet_neighbor::EntryIteratorItem>>();
360        for batch_shape in &test_shape {
361            batches_sender
362                .unbounded_send(testutil::generate_events_in_range(batch_shape.clone()))
363                .expect("failed to send event batch");
364        }
365
366        #[cfg(feature = "fdomain")]
367        let client = fdomain_local::local_client_empty();
368        #[cfg(not(feature = "fdomain"))]
369        let client = fidl::endpoints::ZirconClient;
370        let (view, view_server_end) = client.create_proxy::<fnet_neighbor::ViewMarker>();
371        let entry_iter_fut = testutil::serve_view_impl(batches_receiver, view_server_end);
372
373        let event_stream =
374            event_stream_from_view(&view).expect("failed to open entry iterator").fuse();
375
376        futures::pin_mut!(entry_iter_fut, event_stream);
377
378        for batch_shape in test_shape {
379            for event_idx in batch_shape.into_iter() {
380                futures::select! {
381                    () = entry_iter_fut => panic!(
382                        "fake entry iterator implementation unexpectedly finished"
383                    ),
384                    event = event_stream.next() => {
385                        let actual_event = event
386                            .expect("event stream unexpectedly empty")
387                            .expect("error processing event");
388                        let expected_event = testutil::generate_event(event_idx)
389                                .try_into()
390                                .expect("test event is unexpectedly invalid");
391                        assert_eq!(actual_event, expected_event);
392                    }
393                };
394            }
395        }
396
397        // Close `batches_sender` and observe that the `event_stream` ends.
398        batches_sender.close_channel();
399        let ((), mut events) = futures::join!(entry_iter_fut, event_stream.collect::<Vec<_>>());
400        assert_matches!(
401            events.pop(),
402            Some(Err(EntryIteratorError::Fidl(fidl::Error::ClientChannelClosed {
403                epitaph: fidl::Epitaph::PeerClosed,
404                ..
405            })))
406        );
407        assert_matches!(events[..], []);
408    }
409
410    // Verify that calling `event_stream_from_view` multiple times with the
411    // same `View` proxy, results in independent `EntryIterator` clients.
412    #[fuchsia::test]
413    async fn event_stream_from_view_multiple_iterators() {
414        // Events for 3 iterators. Each receives one batch containing 10 events.
415        let test_data = vec![
416            vec![testutil::generate_events_in_range(0..10)],
417            vec![testutil::generate_events_in_range(10..20)],
418            vec![testutil::generate_events_in_range(20..30)],
419        ];
420
421        // Instantiate the fake EntryIterator implementations.
422        #[cfg(feature = "fdomain")]
423        let client = fdomain_local::local_client_empty();
424        #[cfg(not(feature = "fdomain"))]
425        let client = fidl::endpoints::ZirconClient;
426        let (view, view_server_end) = client.create_proxy::<fnet_neighbor::ViewMarker>();
427        let view_request_stream = view_server_end.into_stream();
428        let entry_iters_fut = view_request_stream
429            .zip(futures::stream::iter(test_data.clone()))
430            .for_each_concurrent(usize::MAX, |(request, event_data)| {
431                testutil::serve_view_request(
432                    request.expect("failed to receive `OpenEntryIterator` request"),
433                    futures::stream::iter(event_data),
434                )
435            });
436
437        let validate_event_streams_fut =
438            futures::future::join_all(test_data.into_iter().map(|event_data| {
439                let events_fut = event_stream_from_view(&view)
440                    .expect("failed to create entry iterator")
441                    .collect::<std::collections::VecDeque<_>>();
442                events_fut.then(|mut events| {
443                    for expected_event in event_data.into_iter().flatten() {
444                        assert_eq!(
445                            events
446                                .pop_front()
447                                .expect("event_stream unexpectedly empty")
448                                .expect("error processing event"),
449                            expected_event.try_into().expect("test event is unexpectedly invalid"),
450                        );
451                    }
452                    assert_matches!(
453                        events.pop_front(),
454                        Some(Err(EntryIteratorError::Fidl(fidl::Error::ClientChannelClosed {
455                            epitaph: fidl::Epitaph::PeerClosed,
456                            ..
457                        })))
458                    );
459                    assert_matches!(events.make_contiguous(), []);
460                    futures::future::ready(())
461                })
462            }));
463
464        futures::join!(entry_iters_fut, validate_event_streams_fut);
465    }
466
467    // Verify that failing to convert an event results in an error and closes
468    // the event stream. `trailing_event` and `trailing_batch` control whether
469    // a good event is sent after the bad event, either as part of the same
470    // batch or in a subsequent batch. The test expects this data to be
471    // truncated from the resulting event_stream.
472    #[test_case(false, false; "no_trailing")]
473    #[test_case(true, false; "trailing_event")]
474    #[test_case(false, true; "trailing_batch")]
475    #[test_case(true, true; "trailing_event_and_batch")]
476    #[fuchsia::test]
477    async fn event_stream_from_view_conversion_error(trailing_event: bool, trailing_batch: bool) {
478        let bad_event = fnet_neighbor::EntryIteratorItem::Added(fnet_neighbor::Entry {
479            interface: None, // Required field omitted.
480            neighbor: Some(fnet::IpAddress::Ipv4(fnet::Ipv4Address { addr: [192, 168, 0, 1] })),
481            state: Some(fnet_neighbor::EntryState::Reachable),
482            mac: Some(fnet::MacAddress { octets: [0, 1, 2, 3, 4, 5] }),
483            updated_at: Some(123456),
484            ..Default::default()
485        });
486
487        let batch = std::iter::once(bad_event)
488            // Optionally append a known good event to the batch.
489            .chain(trailing_event.then(|| testutil::generate_event(0)).into_iter())
490            .collect::<Vec<_>>();
491        let batches = std::iter::once(batch)
492            // Optionally append a known good batch to the sequence of batches.
493            .chain(trailing_batch.then(|| vec![testutil::generate_event(1)]))
494            .collect::<Vec<_>>();
495
496        // Instantiate the fake entry iterator implementation.
497        #[cfg(feature = "fdomain")]
498        let client = fdomain_local::local_client_empty();
499        #[cfg(not(feature = "fdomain"))]
500        let client = fidl::endpoints::ZirconClient;
501        let (view, view_server_end) = client.create_proxy::<fnet_neighbor::ViewMarker>();
502        let entry_iter_fut =
503            testutil::serve_view_impl(futures::stream::iter(batches), view_server_end);
504
505        let event_stream = event_stream_from_view(&view).expect("failed to connect to view").fuse();
506
507        futures::pin_mut!(entry_iter_fut, event_stream);
508        let ((), events) = futures::join!(entry_iter_fut, event_stream.collect::<Vec<_>>());
509        assert_matches!(&events[..], &[Err(EntryIteratorError::Conversion(_))]);
510    }
511
512    // Verify that iterator returning an empty batch results in an error and
513    // closes the event stream. When `trailing_batch` is true, an additional
514    // "good" batch will be sent after the empty batch; the test expects this
515    // data to be truncated from the resulting event_stream.
516    #[test_case(false; "no_trailing_batch")]
517    #[test_case(true; "trailing_batch")]
518    #[fuchsia::test]
519    async fn event_stream_from_view_empty_batch_error(trailing_batch: bool) {
520        let batches = std::iter::once(Vec::new())
521            // Optionally append a known good batch to the sequence of batches.
522            .chain(trailing_batch.then(|| vec![testutil::generate_event(0)]))
523            .collect::<Vec<_>>();
524
525        // Instantiate the fake EntryIterator implementation.
526        #[cfg(feature = "fdomain")]
527        let client = fdomain_local::local_client_empty();
528        #[cfg(not(feature = "fdomain"))]
529        let client = fidl::endpoints::ZirconClient;
530        let (view, view_server_end) = client.create_proxy::<fnet_neighbor::ViewMarker>();
531        let entry_iter_fut =
532            testutil::serve_view_impl(futures::stream::iter(batches), view_server_end);
533
534        let event_stream =
535            event_stream_from_view(&view).expect("failed to create entry iterator").fuse();
536
537        futures::pin_mut!(entry_iter_fut, event_stream);
538        let ((), events) = futures::join!(entry_iter_fut, event_stream.collect::<Vec<_>>());
539        assert_matches!(&events[..], &[Err(EntryIteratorError::EmptyEventBatch)]);
540    }
541
542    #[fuchsia::test]
543    async fn collect_neighbors_until_idle_error_error_in_stream() {
544        let event = Err(EntryIteratorError::EmptyEventBatch);
545        let event_stream = futures::stream::once(futures::future::ready(event));
546        assert_matches!(
547            collect_neighbors_until_idle::<Vec<_>>(event_stream).await,
548            Err(CollectNeighborsUntilIdleError::ErrorInStream(_))
549        );
550    }
551
552    #[fuchsia::test]
553    async fn collect_neighbors_until_idle_error_unexpected_event() {
554        let event =
555            Ok(Event::Added(valid_fidl_entry(NonZeroU64::new(1).unwrap()).try_into().unwrap()));
556        let event_stream = futures::stream::once(futures::future::ready(event));
557        assert_matches!(
558            collect_neighbors_until_idle::<Vec<_>>(event_stream).await,
559            Err(CollectNeighborsUntilIdleError::UnexpectedEvent(_))
560        );
561    }
562
563    #[fuchsia::test]
564    async fn collect_neighbors_until_idle_error_stream_ended() {
565        let event =
566            Ok(Event::Existing(valid_fidl_entry(NonZeroU64::new(1).unwrap()).try_into().unwrap()));
567        let event_stream = futures::stream::once(futures::future::ready(event));
568        assert_matches!(
569            collect_neighbors_until_idle::<Vec<_>>(event_stream).await,
570            Err(CollectNeighborsUntilIdleError::StreamEnded)
571        );
572    }
573
574    #[fuchsia::test]
575    async fn collect_neighbors_until_idle_success() {
576        let entry: Entry = valid_fidl_entry(NonZeroU64::new(1).unwrap()).try_into().unwrap();
577        let mut event_stream = futures::stream::iter([
578            Ok(Event::Existing(entry.clone())),
579            Ok(Event::Idle),
580            Ok(Event::Added(entry.clone())),
581        ]);
582
583        let existing: Vec<_> = collect_neighbors_until_idle(&mut event_stream)
584            .await
585            .expect("failed to collect existing neighbors");
586        assert_eq!(existing, &[entry.clone()]);
587
588        let trailing_events: Vec<_> = event_stream.collect().await;
589        assert_matches!(
590            &trailing_events[..],
591            [Ok(Event::Added(found_entry))] if *found_entry == entry
592        );
593    }
594}
fidl_fuchsia_net_neighbor_ext_fdomain/lib.rs

fidl_fuchsia_net_neighbor_ext_fdomain/
lib.rs
