async fn run(&mut self) -> Result<Never, anyhow::Error> {
        let ethdev_dir_path = &format!("{}/{}", DEV_PATH, devices::EthernetDevice::PATH);
        let mut ethdev_dir_watcher_stream = fvfs_watcher::Watcher::new(
            open_directory_in_namespace(ethdev_dir_path, OPEN_RIGHT_READABLE)
                .context("error opening ethdev directory")?,
        )
        .await
        .with_context(|| format!("creating watcher for ethdevs {}", ethdev_dir_path))?
        .fuse();

        let netdev_dir_path = &format!("{}/{}", DEV_PATH, devices::NetworkDevice::PATH);
        let mut netdev_dir_watcher_stream = fvfs_watcher::Watcher::new(
            open_directory_in_namespace(netdev_dir_path, OPEN_RIGHT_READABLE)
                .context("error opening netdev directory")?,
        )
        .await
        .with_context(|| format!("error creating watcher for netdevs {}", netdev_dir_path))?
        .fuse();

        let if_watcher_event_stream =
            fnet_interfaces_ext::event_stream_from_state(&self.interface_state)
                .context("error creating interface watcher event stream")?
                .fuse();
        futures::pin_mut!(if_watcher_event_stream);

        let (dns_server_watcher, dns_server_watcher_req) =
            fidl::endpoints::create_proxy::<fnet_name::DnsServerWatcherMarker>()
                .context("error creating dns server watcher")?;
        let () = self
            .stack
            .get_dns_server_watcher(dns_server_watcher_req)
            .context("get dns server watcher")?;
        let netstack_dns_server_stream = dns_server_watcher::new_dns_server_stream(
            DnsServersUpdateSource::Netstack,
            dns_server_watcher,
        )
        .map(move |(s, r)| (s, r.context("error getting next Netstack DNS server update event")))
        .boxed();

        let dns_watchers = DnsServerWatchers::empty();
        // `Fuse` (the return of `fuse`) guarantees that once the underlying stream is
        // exhausted, future attempts to poll the stream will return `None`. This would
        // be undesirable if we needed to support a scenario where all streams are
        // exhausted before adding a new stream to the `StreamMap`. However,
        // `netstack_dns_server_stream` is not expected to end so we can fuse the
        // `StreamMap` without issue.
        let mut dns_watchers = dns_watchers.fuse();
        assert!(
            dns_watchers
                .get_mut()
                .insert(DnsServersUpdateSource::Netstack, netstack_dns_server_stream)
                .is_none(),
            "dns watchers should be empty"
        );

        debug!("starting eventloop...");

        loop {
            let () = futures::select! {
                ethdev_dir_watcher_res = ethdev_dir_watcher_stream.try_next() => {
                    let event = ethdev_dir_watcher_res
                        .with_context(|| format!("error watching ethdevs at {}", ethdev_dir_path))?
                        .ok_or_else(|| anyhow::anyhow!(
                            "ethdev directory {} watcher stream ended unexpectedly",
                            ethdev_dir_path
                        ))?;
                    self
                        .handle_dev_event::<devices::EthernetDevice>(ethdev_dir_path, event)
                        .await
                        .context("handle ethdev event")?
                }
                netdev_dir_watcher_res = netdev_dir_watcher_stream.try_next() => {
                    let event = netdev_dir_watcher_res
                        .with_context(|| format!("error watching netdevs at {}", netdev_dir_path))?
                        .ok_or_else(|| anyhow::anyhow!(
                            "netdev directory {} watcher stream ended unexpectedly",
                            netdev_dir_path
                        ))?;
                    self
                        .handle_dev_event::<devices::NetworkDevice>(netdev_dir_path, event)
                        .await
                        .context("handle netdev event")?
                }
                if_watcher_res = if_watcher_event_stream.try_next() => {
                    let event = if_watcher_res
                        .context("error watching interface property changes")?
                        .ok_or(
                            anyhow::anyhow!("interface watcher event stream ended unexpectedly")
                        )?;
                    trace!("got interfaces watcher event = {:?}", event);

                    self
                        .handle_interface_watcher_event(
                            event, dns_watchers.get_mut()
                        )
                        .await
                        .context("handle interface watcher event")
                        .or_else(|e| match e {
                            errors::Error::NonFatal(e) => {
                                error!("non-fatal error: {:?}", e);
                                Ok(())
                            }
                            errors::Error::Fatal(e) => Err(e),
                        })?
                }
                dns_watchers_res = dns_watchers.next() => {
                    let (source, res) = dns_watchers_res
                        .ok_or(anyhow::anyhow!("dns watchers stream should never be exhausted"))?;
                    let servers = match res {
                        Ok(s) => s,
                        Err(e) => {
                            // TODO(fxbug.dev/57484): Restart the DNS server watcher.
                            error!("non-fatal error getting next event \
                                from DNS server watcher stream with source = {:?}: {:?}", source, e);
                            let () = self
                                .handle_dns_server_watcher_done(source, dns_watchers.get_mut())
                                .await
                                .with_context(|| {
                                    format!("error handling completion of DNS server watcher for \
                                        {:?}", source)
                                })?;
                            continue;
                        }
                    };

                    self.update_dns_servers(source, servers).await.with_context(|| {
                        format!("error handling DNS servers update from {:?}", source)
                    }).or_else(|e| match e {
                        errors::Error::NonFatal(e) => {
                            error!("non-fatal error: {:?}", e);
                            Ok(())
                        }
                        errors::Error::Fatal(e) => Err(e),
                    })?
                }
                complete => break,
            };
        }

        Err(anyhow::anyhow!("eventloop ended unexpectedly"))
    }
/// Run the network configuration eventloop.
///
/// The device directory will be monitored for device events and the netstack will be
/// configured with a new interface on new device discovery.
Run the network configuration eventloop.
The device directory will be monitored for device events and the netstack will be
configured with a new interface on new device discovery.
Run the network configuration eventloop . The device directory will be monitored for device events and the netstack will be configured with a new interface on new device discovery .
async fn run(&mut self) -> Result<Never, anyhow::Error> {
        let ethdev_dir_path = &format!("{}/{}", DEV_PATH, devices::EthernetDevice::PATH);
        let mut ethdev_dir_watcher_stream = fvfs_watcher::Watcher::new(
            open_directory_in_namespace(ethdev_dir_path, OPEN_RIGHT_READABLE)
                .context("error opening ethdev directory")?,
        )
        .await
        .with_context(|| format!("creating watcher for ethdevs {}", ethdev_dir_path))?
        .fuse();
        let netdev_dir_path = &format!("{}/{}", DEV_PATH, devices::NetworkDevice::PATH);
        let mut netdev_dir_watcher_stream = fvfs_watcher::Watcher::new(
            open_directory_in_namespace(netdev_dir_path, OPEN_RIGHT_READABLE)
                .context("error opening netdev directory")?,
        )
        .await
        .with_context(|| format!("error creating watcher for netdevs {}", netdev_dir_path))?
        .fuse();
        let if_watcher_event_stream =
            fnet_interfaces_ext::event_stream_from_state(&self.interface_state)
                .context("error creating interface watcher event stream")?
                .fuse();
        futures::pin_mut!(if_watcher_event_stream);
        let (dns_server_watcher, dns_server_watcher_req) =
            fidl::endpoints::create_proxy::<fnet_name::DnsServerWatcherMarker>()
                .context("error creating dns server watcher")?;
        let () = self
            .stack
            .get_dns_server_watcher(dns_server_watcher_req)
            .context("get dns server watcher")?;
        let netstack_dns_server_stream = dns_server_watcher::new_dns_server_stream(
            DnsServersUpdateSource::Netstack,
            dns_server_watcher,
        )
        .map(move |(s, r)| (s, r.context("error getting next Netstack DNS server update event")))
        .boxed();
        let dns_watchers = DnsServerWatchers::empty();
        // `Fuse` (the return of `fuse`) guarantees that once the underlying stream is
        // exhausted, future attempts to poll the stream will return `None`. This would
        // be undesirable if we needed to support a scenario where all streams are
        // exhausted before adding a new stream to the `StreamMap`. However,
        // `netstack_dns_server_stream` is not expected to end so we can fuse the
        // `StreamMap` without issue.
        let mut dns_watchers = dns_watchers.fuse();
        assert!(
            dns_watchers
                .get_mut()
                .insert(DnsServersUpdateSource::Netstack, netstack_dns_server_stream)
                .is_none(),
            "dns watchers should be empty"
        );
        debug!("starting eventloop...");
        loop {
            let () = futures::select! {
                ethdev_dir_watcher_res = ethdev_dir_watcher_stream.try_next() => {
                    let event = ethdev_dir_watcher_res
                        .with_context(|| format!("error watching ethdevs at {}", ethdev_dir_path))?
                        .ok_or_else(|| anyhow::anyhow!(
                            "ethdev directory {} watcher stream ended unexpectedly",
                            ethdev_dir_path
                        ))?;
                    self
                        .handle_dev_event::<devices::EthernetDevice>(ethdev_dir_path, event)
                        .await
                        .context("handle ethdev event")?
                }
                netdev_dir_watcher_res = netdev_dir_watcher_stream.try_next() => {
                    let event = netdev_dir_watcher_res
                        .with_context(|| format!("error watching netdevs at {}", netdev_dir_path))?
                        .ok_or_else(|| anyhow::anyhow!(
                            "netdev directory {} watcher stream ended unexpectedly",
                            netdev_dir_path
                        ))?;
                    self
                        .handle_dev_event::<devices::NetworkDevice>(netdev_dir_path, event)
                        .await
                        .context("handle netdev event")?
                }
                if_watcher_res = if_watcher_event_stream.try_next() => {
                    let event = if_watcher_res
                        .context("error watching interface property changes")?
                        .ok_or(
                            anyhow::anyhow!("interface watcher event stream ended unexpectedly")
                        )?;
                    trace!("got interfaces watcher event = {:?}", event);
                    self
                        .handle_interface_watcher_event(
                            event, dns_watchers.get_mut()
                        )
                        .await
                        .context("handle interface watcher event")
                        .or_else(|e| match e {
                            errors::Error::NonFatal(e) => {
                                error!("non-fatal error: {:?}", e);
                                Ok(())
                            }
                            errors::Error::Fatal(e) => Err(e),
                        })?
                }
                dns_watchers_res = dns_watchers.next() => {
                    let (source, res) = dns_watchers_res
                        .ok_or(anyhow::anyhow!("dns watchers stream should never be exhausted"))?;
                    let servers = match res {
                        Ok(s) => s,
                        Err(e) => {
                            // TODO(fxbug.dev/57484): Restart the DNS server watcher.
                            error!("non-fatal error getting next event \
                                from DNS server watcher stream with source = {:?}: {:?}", source, e);
                            let () = self
                                .handle_dns_server_watcher_done(source, dns_watchers.get_mut())
                                .await
                                .with_context(|| {
                                    format!("error handling completion of DNS server watcher for \
                                        {:?}", source)
                                })?;
                            continue;
                        }
                    };
                    self.update_dns_servers(source, servers).await.with_context(|| {
                        format!("error handling DNS servers update from {:?}", source)
                    }).or_else(|e| match e {
                        errors::Error::NonFatal(e) => {
                            error!("non-fatal error: {:?}", e);
                            Ok(())
                        }
                        errors::Error::Fatal(e) => Err(e),
                    })?
                }
                complete => break,
            };
        }
        Err(anyhow::anyhow!("eventloop ended unexpectedly"))
    }
async fn run ( & mut self ) -> Result < Never , anyhow :: Error > { let ethdev_dir_path = & format ! ( " " DEV_PATH devices EthernetDevice PATH ) ; let mut ethdev_dir_watcher_stream = fvfs_watcher :: Watcher :: new ( open_directory_in_namespace ( ethdev_dir_path , OPEN_RIGHT_READABLE ) . context ( " " ) ? , ) . await . with_context ( | | format ! ( " " ethdev_dir_path ) ) ? . fuse ( ) ; let netdev_dir_path = & format ! ( " " DEV_PATH devices NetworkDevice PATH ) ; let mut netdev_dir_watcher_stream = fvfs_watcher :: Watcher :: new ( open_directory_in_namespace ( netdev_dir_path , OPEN_RIGHT_READABLE ) . context ( " " ) ? , ) . await . with_context ( | | format ! ( " " netdev_dir_path ) ) ? . fuse ( ) ; let if_watcher_event_stream = fnet_interfaces_ext :: event_stream_from_state ( & self . interface_state ) . context ( " " ) ? . fuse ( ) ; futures :: pin_mut ! ( if_watcher_event_stream ) ; let ( dns_server_watcher , dns_server_watcher_req ) = fidl :: endpoints :: create_proxy :: < fnet_name :: DnsServerWatcherMarker > ( ) . context ( " " ) ? ; let ( ) = self . stack . get_dns_server_watcher ( dns_server_watcher_req ) . context ( " " ) ? ; let netstack_dns_server_stream = dns_server_watcher :: new_dns_server_stream ( DnsServersUpdateSource :: Netstack , dns_server_watcher , ) . map ( move | ( s , r ) | ( s , r . context ( " " ) ) ) . boxed ( ) ; let dns_watchers = DnsServerWatchers :: empty ( ) ; // `Fuse` (the return of `fuse`) guarantees that once the underlying stream is // exhausted, future attempts to poll the stream will return `None`. This would // be undesirable if we needed to support a scenario where all streams are // exhausted before adding a new stream to the `StreamMap`. However, // `netstack_dns_server_stream` is not expected to end so we can fuse the // `StreamMap` without issue. let mut dns_watchers = dns_watchers . fuse ( ) ; assert ! ( dns_watchers get_mut ( ) insert ( DnsServersUpdateSource Netstack netstack_dns_server_stream ) is_none ( ) " " ) ; debug ! ( " " ) ; loop { let ( ) = futures :: select ! { ethdev_dir_watcher_res ethdev_dir_watcher_stream try_next ( ) { let event ethdev_dir_watcher_res with_context ( format ( " " ethdev_dir_path ) ) ok_or_else ( anyhow anyhow ( " " ethdev_dir_path ) ) self handle_dev_event devices EthernetDevice ( ethdev_dir_path event ) await context ( " " ) } netdev_dir_watcher_res netdev_dir_watcher_stream try_next ( ) { let event netdev_dir_watcher_res with_context ( format ( " " netdev_dir_path ) ) ok_or_else ( anyhow anyhow ( " " netdev_dir_path ) ) self handle_dev_event devices NetworkDevice ( netdev_dir_path event ) await context ( " " ) } if_watcher_res if_watcher_event_stream try_next ( ) { let event if_watcher_res context ( " " ) ok_or ( anyhow anyhow ( " " ) ) trace ( " " event ) self handle_interface_watcher_event ( event dns_watchers get_mut ( ) ) await context ( " " ) or_else ( e match e { errors Error NonFatal ( e ) { error ( " " e ) Ok ( ( ) ) } errors Error Fatal ( e ) Err ( e ) } ) } dns_watchers_res dns_watchers next ( ) { let ( source res ) dns_watchers_res ok_or ( anyhow anyhow ( " " ) ) let servers match res { Ok ( s ) s Err ( e ) { // TODO(fxbug.dev/57484): Restart the DNS server watcher. error ( " \
 " source e ) let ( ) self handle_dns_server_watcher_done ( source dns_watchers get_mut ( ) ) await with_context ( { format ( " \
 " source ) } ) continue } } self update_dns_servers ( source servers ) await with_context ( { format ( " " source ) } ) or_else ( e match e { errors Error NonFatal ( e ) { error ( " " e ) Ok ( ( ) ) } errors Error Fatal ( e ) Err ( e ) } ) } complete break } ; } Err ( anyhow :: anyhow ! ( " " ) ) }
Run the network configuration eventloop.
Run the network configuration eventloop .
&mut self
&mut self