src/connectivity/management/reachability/src/worker.rs

// Copyright 2019 The Fuchsia Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

use {
    crate::eventloop::Event,
    anyhow::Error,
    fidl_fuchsia_net_stack as stack, fidl_fuchsia_netstack as netstack, fuchsia_async as fasync,
    fuchsia_component::server::ServiceFs,
    fuchsia_inspect::{component, Inspector},
    futures::{channel::mpsc, StreamExt, TryFutureExt},
};

/// `EventWorker` waits for events from netstack and stack and sends them on the indicated
/// `event_chan`.
pub struct EventWorker;

impl EventWorker {
    pub fn spawn(
        self,
        streams: (stack::StackEventStream, netstack::NetstackEventStream),
        event_chan: mpsc::UnboundedSender<Event>,
    ) {
        fasync::Task::local(
            async move {
                let mut select_stream = futures::stream::select(
                    streams.0.map(|e| e.map(Event::StackEvent)),
                    streams.1.map(|e| e.map(Event::NetstackEvent)),
                );

                while let Some(e) = select_stream.next().await {
                    debug!("Sending event: {:?}", e);
                    match e {
                        Ok(e) => event_chan.unbounded_send(e)?,
                        Err(e) => error!("Fidl event error: {}", e),
                    }
                }
                Ok(())
            }
            .unwrap_or_else(|err: Error| error!("Sending event error {:?}", err)),
        )
        .detach();
    }
}

pub struct FidlWorker;

impl FidlWorker {
    pub fn spawn(self) -> Result<&'static Inspector, Error> {
        let mut fs = ServiceFs::new_local();
        fs.take_and_serve_directory_handle()?;

        let inspector = component::inspector();
        inspector.serve(&mut fs)?;

        fasync::Task::local(fs.collect()).detach();
        Ok(inspector)
    }
}

/// `TimerWorker` waits for timer events and sends them on the indicated
/// `event_chan`.
pub struct TimerWorker;

impl TimerWorker {
    pub fn spawn(
        self,
        mut timer: fasync::Interval,
        event_chan: mpsc::UnboundedSender<Event>,
        id: Option<u64>,
    ) {
        debug!("spawn periodic timer");
        fasync::Task::local(async move {
            while let Some(()) = (timer.next()).await {
                event_chan.unbounded_send(Event::TimerEvent(id)).unwrap_or_else(
                    |err: mpsc::TrySendError<Event>| error!("Sending event error {:?}", err),
                );
            }
        })
        .detach();
    }
}