// 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::host_identifier::HostIdentifier,
    anyhow::{Context as _, Result},
    fidl::prelude::*,
    fidl_fuchsia_developer_remotecontrol as rcs,
    fidl_fuchsia_diagnostics::Selector,
    fidl_fuchsia_io as io,
    fidl_fuchsia_net_ext::SocketAddress as SocketAddressExt,
    fuchsia_async as fasync, fuchsia_zircon as zx,
    futures::future::join,
    futures::prelude::*,
    selector_maps::{MappingError, SelectorMappingList},
    std::{cell::RefCell, net::SocketAddr, rc::Rc},
    tracing::*,
};

mod host_identifier;
mod service_discovery;

const HUB_ROOT: &str = "/discovery_root";

pub struct RemoteControlService {
    ids: RefCell<Vec<u64>>,
    id_allocator: fn() -> Result<HostIdentifier>,
    maps: SelectorMappingList,
}

impl RemoteControlService {
    pub async fn new() -> Self {
        let f = match fuchsia_fs::open_file_in_namespace(
            "/config/data/selector-maps.json",
            io::OpenFlags::RIGHT_READABLE,
        ) {
            Ok(f) => f,
            Err(e) => {
                error!(%e, "failed to open selector maps json file");
                return Self::new_with_allocator_and_maps(
                    || HostIdentifier::new(),
                    SelectorMappingList::default(),
                );
            }
        };
        let bytes = match fuchsia_fs::read_file_bytes(&f).await {
            Ok(b) => b,
            Err(e) => {
                error!(?e, "failed to read bytes from selector maps json");
                return Self::new_with_allocator_and_maps(
                    || HostIdentifier::new(),
                    SelectorMappingList::default(),
                );
            }
        };
        let list: SelectorMappingList = match serde_json::from_slice(bytes.as_slice()) {
            Ok(m) => m,
            Err(e) => {
                error!(?e, "failed to parse selector map json");
                return Self::new_with_allocator_and_maps(
                    || HostIdentifier::new(),
                    SelectorMappingList::default(),
                );
            }
        };
        return Self::new_with_allocator_and_maps(|| HostIdentifier::new(), list);
    }

    pub(crate) fn new_with_allocator_and_maps(
        id_allocator: fn() -> Result<HostIdentifier>,
        maps: SelectorMappingList,
    ) -> Self {
        return Self { id_allocator, ids: Default::default(), maps };
    }

    async fn handle(self: &Rc<Self>, request: rcs::RemoteControlRequest) -> Result<()> {
        match request {
            rcs::RemoteControlRequest::EchoString { value, responder } => {
                log::info!("Received echo string {}", value);
                responder.send(&value)?;
                Ok(())
            }
            rcs::RemoteControlRequest::AddId { id, responder } => {
                self.ids.borrow_mut().push(id);
                responder.send()?;
                Ok(())
            }
            rcs::RemoteControlRequest::IdentifyHost { responder } => {
                self.clone().identify_host(responder).await?;
                Ok(())
            }
            rcs::RemoteControlRequest::Connect { selector, service_chan, responder } => {
                responder
                    .send(&mut self.clone().connect_to_service(selector, service_chan).await)?;
                Ok(())
            }
            rcs::RemoteControlRequest::Select { selector, responder } => {
                responder.send(&mut self.clone().select(selector).await)?;
                Ok(())
            }
            rcs::RemoteControlRequest::OpenHub { server, responder } => {
                responder.send(
                    &mut fuchsia_fs::connect_in_namespace(
                        HUB_ROOT,
                        server.into_channel(),
                        io::OpenFlags::RIGHT_READABLE | io::OpenFlags::RIGHT_WRITABLE,
                    )
                    .map_err(|i| i.into_raw()),
                )?;
                Ok(())
            }
            rcs::RemoteControlRequest::RootRealmExplorer { server, responder } => {
                responder.send(
                    &mut fuchsia_fs::connect_in_namespace(
                        &format!(
                            "/svc/{}.root",
                            fidl_fuchsia_sys2::RealmExplorerMarker::PROTOCOL_NAME
                        ),
                        server.into_channel(),
                        io::OpenFlags::RIGHT_READABLE | io::OpenFlags::RIGHT_WRITABLE,
                    )
                    .map_err(|i| i.into_raw()),
                )?;
                Ok(())
            }
            rcs::RemoteControlRequest::RootRealmQuery { server, responder } => {
                responder.send(
                    &mut fuchsia_fs::connect_in_namespace(
                        &format!(
                            "/svc/{}.root",
                            fidl_fuchsia_sys2::RealmQueryMarker::PROTOCOL_NAME
                        ),
                        server.into_channel(),
                        io::OpenFlags::RIGHT_READABLE | io::OpenFlags::RIGHT_WRITABLE,
                    )
                    .map_err(|i| i.into_raw()),
                )?;
                Ok(())
            }
            rcs::RemoteControlRequest::RootLifecycleController { server, responder } => {
                responder.send(
                    &mut fuchsia_fs::connect_in_namespace(
                        &format!(
                            "/svc/{}.root",
                            fidl_fuchsia_sys2::LifecycleControllerMarker::PROTOCOL_NAME
                        ),
                        server.into_channel(),
                        io::OpenFlags::RIGHT_READABLE | io::OpenFlags::RIGHT_WRITABLE,
                    )
                    .map_err(|i| i.into_raw()),
                )?;
                Ok(())
            }
            rcs::RemoteControlRequest::RootRouteValidator { server, responder } => {
                responder.send(
                    &mut fuchsia_fs::connect_in_namespace(
                        &format!(
                            "/svc/{}.root",
                            fidl_fuchsia_sys2::RouteValidatorMarker::PROTOCOL_NAME
                        ),
                        server.into_channel(),
                        io::OpenFlags::RIGHT_READABLE | io::OpenFlags::RIGHT_WRITABLE,
                    )
                    .map_err(|i| i.into_raw()),
                )?;
                Ok(())
            }
            rcs::RemoteControlRequest::ForwardTcp { addr, socket, responder } => {
                let addr: SocketAddressExt = addr.into();
                let addr = addr.0;
                let mut result = match fasync::Socket::from_socket(socket) {
                    Ok(socket) => match self.connect_forwarded_port(addr, socket).await {
                        Ok(()) => Ok(()),
                        Err(e) => {
                            log::error!("Port forward connection failed: {:?}", e);
                            Err(rcs::TunnelError::ConnectFailed)
                        }
                    },
                    Err(e) => {
                        log::error!("Could not use socket asynchronously: {:?}", e);
                        Err(rcs::TunnelError::SocketFailed)
                    }
                };
                responder.send(&mut result)?;
                Ok(())
            }
            rcs::RemoteControlRequest::ReverseTcp { addr, client, responder } => {
                let addr: SocketAddressExt = addr.into();
                let addr = addr.0;
                let client = match client.into_proxy() {
                    Ok(proxy) => proxy,
                    Err(e) => {
                        log::error!("Could not communicate with callback: {:?}", e);
                        responder.send(&mut Err(rcs::TunnelError::CallbackError))?;
                        return Ok(());
                    }
                };
                let mut result = match self.listen_reversed_port(addr, client).await {
                    Ok(()) => Ok(()),
                    Err(e) => {
                        log::error!("Port forward connection failed: {:?}", e);
                        Err(rcs::TunnelError::ConnectFailed)
                    }
                };
                responder.send(&mut result)?;
                Ok(())
            }
        }
    }

    pub async fn serve_stream(self: Rc<Self>, stream: rcs::RemoteControlRequestStream) {
        stream
            .for_each_concurrent(None, |request| async {
                match request {
                    Ok(request) => {
                        let _ = self
                            .handle(request)
                            .await
                            .map_err(|e| log::warn!("stream request handling error: {:?}", e));
                    }
                    Err(e) => log::warn!("stream error: {:?}", e),
                }
            })
            .await
    }

    async fn listen_reversed_port(
        &self,
        listen_addr: SocketAddr,
        client: rcs::ForwardCallbackProxy,
    ) -> Result<(), std::io::Error> {
        let mut listener = fasync::net::TcpListener::bind(&listen_addr)?.accept_stream();

        fasync::Task::local(async move {
            let mut client_closed = client.on_closed().fuse();

            loop {
                // Listen for a connection, or exit if the client has gone away.
                let (stream, addr) = futures::select! {
                    result = listener.next().fuse() => {
                        match result {
                            Some(Ok(x)) => x,
                            Some(Err(e)) => {
                                log::warn!("Error accepting connection: {:?}", e);
                                continue;
                            }
                            None => {
                                log::warn!("reverse tunnel to {:?} listener socket closed", listen_addr);
                                break;
                            }
                        }
                    }
                    _ = client_closed => {
                        log::info!("reverse tunnel {:?} client has closed", listen_addr);
                        break;
                    }
                };

                log::info!("reverse tunnel connection from {:?} to {:?}", addr, listen_addr);

                let (local, remote) = match zx::Socket::create(zx::SocketOpts::STREAM) {
                    Ok(x) => x,
                    Err(e) => {
                        log::warn!("Error creating socket: {:?}", e);
                        continue;
                    }
                };

                let local = match fasync::Socket::from_socket(local) {
                    Ok(x) => x,
                    Err(e) => {
                        log::warn!("Error converting socket to async: {:?}", e);
                        continue;
                    }
                };

                spawn_forward_traffic(stream, local);

                // Send the socket to the client.
                if let Err(e) = client.forward(remote, &mut SocketAddressExt(addr).into()) {
                    // The client has gone away, so stop the task.
                    if let fidl::Error::ClientChannelClosed { .. } = e {
                        log::warn!("tunnel client channel closed while forwarding socket");
                        break;
                    }

                    log::warn!("Could not return forwarded socket to client: {:?}", e);
                }
            }
        })
        .detach();

        Ok(())
    }

    async fn connect_forwarded_port(
        &self,
        addr: SocketAddr,
        socket: fasync::Socket,
    ) -> Result<(), std::io::Error> {
        let tcp_conn = fasync::net::TcpStream::connect(addr)?.await?;

        spawn_forward_traffic(tcp_conn, socket);

        Ok(())
    }

    async fn connect_with_matcher(
        self: &Rc<Self>,
        selector: &Selector,
        service_chan: zx::Channel,
        matcher_fut: impl Future<Output = Result<Vec<service_discovery::PathEntry>>>,
    ) -> Result<rcs::ServiceMatch, rcs::ConnectError> {
        let paths = matcher_fut.await.map_err(|err| {
            warn!(?selector, %err, "error looking for matching services for selector");
            rcs::ConnectError::ServiceDiscoveryFailed
        })?;
        if paths.is_empty() {
            return Err(rcs::ConnectError::NoMatchingServices);
        } else if paths.len() > 1 {
            // TODO(jwing): we should be able to communicate this to the FE somehow.
            warn!(
                ?paths,
                "Selector must match exactly one service. Provided selector matched all of the following");
            return Err(rcs::ConnectError::MultipleMatchingServices);
        }
        let svc_match = paths.get(0).unwrap();
        let hub_path = svc_match.hub_path.to_str().unwrap();
        info!(hub_path, "attempting to connect");
        fuchsia_fs::connect_in_namespace(
            hub_path,
            service_chan,
            io::OpenFlags::RIGHT_READABLE | io::OpenFlags::RIGHT_WRITABLE,
        )
        .map_err(|err| {
            error!(?selector, %err, "error connecting to selector");
            rcs::ConnectError::ServiceConnectFailed
        })?;

        Ok(svc_match.into())
    }

    pub(crate) fn map_selector(
        self: &Rc<Self>,
        selector: Selector,
    ) -> Result<Selector, rcs::ConnectError> {
        self.maps.map_selector(selector.clone()).map_err(|e| {
            match e {
                MappingError::BadSelector(selector_str, err) => {
                    error!(?selector, ?selector_str, %err, "got invalid selector mapping");
                }
                MappingError::BadInputSelector(err) => {
                    error!(%err, "input selector invalid");
                }
                MappingError::Unbounded => {
                    error!(?selector, %e, "got a cycle in mapping selector");
                }
            }
            rcs::ConnectError::ServiceRerouteFailed
        })
    }

    pub async fn connect_to_service(
        self: &Rc<Self>,
        selector: Selector,
        service_chan: zx::Channel,
    ) -> Result<rcs::ServiceMatch, rcs::ConnectError> {
        let selector = self.map_selector(selector.clone())?;
        self.connect_with_matcher(
            &selector,
            service_chan,
            service_discovery::get_matching_paths(HUB_ROOT, &selector),
        )
        .await
    }

    async fn select_with_matcher(
        self: &Rc<Self>,
        selector: &Selector,
        matcher_fut: impl Future<Output = Result<Vec<service_discovery::PathEntry>>>,
    ) -> Result<Vec<rcs::ServiceMatch>, rcs::SelectError> {
        let paths = matcher_fut.await.map_err(|err| {
            warn!(?selector, %err, "error looking for matching services for selector");
            rcs::SelectError::ServiceDiscoveryFailed
        })?;

        Ok(paths.iter().map(|p| p.into()).collect::<Vec<rcs::ServiceMatch>>())
    }

    pub async fn select(
        self: &Rc<Self>,
        selector: Selector,
    ) -> Result<Vec<rcs::ServiceMatch>, rcs::SelectError> {
        self.select_with_matcher(
            &selector,
            service_discovery::get_matching_paths(HUB_ROOT, &selector),
        )
        .await
    }

    pub async fn identify_host(
        self: &Rc<Self>,
        responder: rcs::RemoteControlIdentifyHostResponder,
    ) -> Result<()> {
        let identifier = match (self.id_allocator)() {
            Ok(i) => i,
            Err(e) => {
                error!(%e, "Allocating host identifier");
                return responder
                    .send(&mut Err(rcs::IdentifyHostError::ProxyConnectionFailed))
                    .context("responding to client");
            }
        };

        // TODO(raggi): limit size to stay under message size limit.
        let ids = self.ids.borrow().clone();
        let mut target_identity = identifier.identify().await.map(move |mut i| {
            i.ids = Some(ids);
            i
        });
        responder.send(&mut target_identity).context("responding to client")?;
        Ok(())
    }
}

#[derive(Debug)]
enum ForwardError {
    TcpToZx(anyhow::Error),
    ZxToTcp(anyhow::Error),
    Both { tcp_to_zx: anyhow::Error, zx_to_tcp: anyhow::Error },
}

fn spawn_forward_traffic(tcp_side: fasync::net::TcpStream, zx_side: fasync::Socket) {
    fasync::Task::local(async move {
        match forward_traffic(tcp_side, zx_side).await {
            Ok(()) => {}
            Err(ForwardError::TcpToZx(err)) => {
                log::error!("error forwarding from tcp to zx socket: {:#}", err);
            }
            Err(ForwardError::ZxToTcp(err)) => {
                log::error!("error forwarding from zx to tcp socket: {:#}", err);
            }
            Err(ForwardError::Both { tcp_to_zx, zx_to_tcp }) => {
                log::error!(
                    "error forwarding from zx to tcp socket:\n{:#}\n{:#}",
                    tcp_to_zx,
                    zx_to_tcp
                );
            }
        }
    })
    .detach()
}

async fn forward_traffic(
    tcp_side: fasync::net::TcpStream,
    zx_side: fasync::Socket,
) -> Result<(), ForwardError> {
    // We will forward traffic with two sub-tasks. One to stream bytes from the
    // tcp socket to the zircon socket, and vice versa. Since we have two tasks,
    // we need to handle how we exit the loops, otherwise we risk leaking
    // resource.
    //
    // To handle this, we'll create two promises that will resolve upon the
    // stream closing. For the zircon socket, we can use a native signal, but
    // unfortunately fasync::net::TcpStream doesn't support listening for
    // closure, so we'll just use a oneshot channel to signal to the other task
    // when the tcp stream closes.
    let (tcp_closed_tx, mut tcp_closed_rx) = futures::channel::oneshot::channel::<()>();
    let mut zx_closed =
        fasync::OnSignals::new(&zx_side, zx::Signals::SOCKET_PEER_CLOSED).extend_lifetime().fuse();

    let (mut tcp_read, mut tcp_write) = tcp_side.split();
    let (mut zx_read, mut zx_write) = zx_side.split();

    let tcp_to_zx = async move {
        let res = async move {
            // TODO(84188): Use a buffer pool once we have them.
            let mut buf = [0; 4096];
            loop {
                futures::select! {
                    res = tcp_read.read(&mut buf).fuse() => {
                        let num_bytes = res.context("read tcp socket")?;
                        if num_bytes == 0 {
                            return Ok(());
                        }

                        zx_write.write_all(&mut buf[..num_bytes]).await.context("write zx socket")?;
                        zx_write.flush().await.context("flush zx socket")?;
                    }
                    _ = zx_closed => {
                        return Ok(());
                    }
                }
            }
        }
        .await;

        // Let the other task know the tcp stream has shut down. If the other
        // task finished before this one, this send could fail. That's okay, so
        // just ignore the result.
        let _ = tcp_closed_tx.send(());

        res
    };

    let zx_to_tcp = async move {
        // TODO(84188): Use a buffer pool once we have them.
        let mut buf = [0; 4096];
        loop {
            futures::select! {
                res = zx_read.read(&mut buf).fuse() => {
                    let num_bytes = res.context("read zx socket")?;
                    if num_bytes == 0 {
                        return Ok(());
                    }
                    tcp_write.write_all(&mut buf[..num_bytes]).await.context("write tcp socket")?;
                    tcp_write.flush().await.context("flush tcp socket")?;
                }
                _ = tcp_closed_rx => {
                    break Ok(());
                }
            }
        }
    };

    match join(tcp_to_zx, zx_to_tcp).await {
        (Ok(()), Ok(())) => Ok(()),
        (Err(tcp_to_zx), Err(zx_to_tcp)) => Err(ForwardError::Both { tcp_to_zx, zx_to_tcp }),
        (Err(tcp_to_zx), Ok(())) => Err(ForwardError::TcpToZx(tcp_to_zx)),
        (Ok(()), Err(zx_to_tcp)) => Err(ForwardError::ZxToTcp(zx_to_tcp)),
    }
}

#[cfg(test)]
mod tests {
    use {
        super::*,
        assert_matches::assert_matches,
        fidl_fuchsia_buildinfo as buildinfo, fidl_fuchsia_developer_remotecontrol as rcs,
        fidl_fuchsia_device as fdevice, fidl_fuchsia_hwinfo as hwinfo, fidl_fuchsia_io as fio,
        fidl_fuchsia_net as fnet, fidl_fuchsia_net_interfaces as fnet_interfaces,
        fuchsia_zircon as zx,
        selectors::{parse_selector, VerboseError},
        service_discovery::PathEntry,
        std::net::Ipv4Addr,
        std::path::PathBuf,
    };

    const NODENAME: &'static str = "thumb-set-human-shred";
    const BOOT_TIME: u64 = 123456789000000000;
    const SERIAL: &'static str = "test_serial";
    const BOARD_CONFIG: &'static str = "test_board_name";
    const PRODUCT_CONFIG: &'static str = "core";
    const FAKE_SERVICE_SELECTOR: &'static str = "my/component:expose:some.fake.Service";
    const MAPPED_SERVICE_SELECTOR: &'static str = "my/other/component:out:some.fake.mapped.Service";

    const IPV4_ADDR: [u8; 4] = [127, 0, 0, 1];
    const IPV6_ADDR: [u8; 16] = [127, 1, 2, 3, 4, 5, 6, 7, 8, 9, 1, 2, 3, 4, 5, 6];

    fn setup_fake_device_service() -> hwinfo::DeviceProxy {
        let (proxy, mut stream) =
            fidl::endpoints::create_proxy_and_stream::<hwinfo::DeviceMarker>().unwrap();
        fasync::Task::spawn(async move {
            while let Ok(Some(req)) = stream.try_next().await {
                match req {
                    hwinfo::DeviceRequest::GetInfo { responder } => {
                        let _ = responder.send(hwinfo::DeviceInfo {
                            serial_number: Some(String::from(SERIAL)),
                            ..hwinfo::DeviceInfo::EMPTY
                        });
                    }
                }
            }
        })
        .detach();

        proxy
    }

    fn setup_fake_build_info_service() -> buildinfo::ProviderProxy {
        let (proxy, mut stream) =
            fidl::endpoints::create_proxy_and_stream::<buildinfo::ProviderMarker>().unwrap();
        fasync::Task::spawn(async move {
            while let Ok(Some(req)) = stream.try_next().await {
                match req {
                    buildinfo::ProviderRequest::GetBuildInfo { responder } => {
                        let _ = responder.send(buildinfo::BuildInfo {
                            board_config: Some(String::from(BOARD_CONFIG)),
                            product_config: Some(String::from(PRODUCT_CONFIG)),
                            ..buildinfo::BuildInfo::EMPTY
                        });
                    }
                }
            }
        })
        .detach();

        proxy
    }

    fn setup_fake_name_provider_service() -> fdevice::NameProviderProxy {
        let (proxy, mut stream) =
            fidl::endpoints::create_proxy_and_stream::<fdevice::NameProviderMarker>().unwrap();

        fasync::Task::spawn(async move {
            while let Ok(Some(req)) = stream.try_next().await {
                match req {
                    fdevice::NameProviderRequest::GetDeviceName { responder } => {
                        let _ = responder.send(&mut Ok(String::from(NODENAME)));
                    }
                }
            }
        })
        .detach();

        proxy
    }

    fn setup_fake_interface_state_service() -> fnet_interfaces::StateProxy {
        let (proxy, mut stream) =
            fidl::endpoints::create_proxy_and_stream::<fnet_interfaces::StateMarker>().unwrap();

        fasync::Task::spawn(async move {
            while let Ok(Some(req)) = stream.try_next().await {
                match req {
                    fnet_interfaces::StateRequest::GetWatcher {
                        options: _,
                        watcher,
                        control_handle: _,
                    } => {
                        let mut stream = watcher.into_stream().unwrap();
                        let mut first = true;
                        while let Ok(Some(req)) = stream.try_next().await {
                            match req {
                                fnet_interfaces::WatcherRequest::Watch { responder } => {
                                    let mut event = if first {
                                        first = false;
                                        fnet_interfaces::Event::Existing(
                                            fnet_interfaces::Properties {
                                                id: Some(1),
                                                addresses: Some(
                                                    IntoIterator::into_iter([
                                                        fnet::Subnet {
                                                            addr: fnet::IpAddress::Ipv4(
                                                                fnet::Ipv4Address {
                                                                    addr: IPV4_ADDR,
                                                                },
                                                            ),
                                                            prefix_len: 4,
                                                        },
                                                        fnet::Subnet {
                                                            addr: fnet::IpAddress::Ipv6(
                                                                fnet::Ipv6Address {
                                                                    addr: IPV6_ADDR,
                                                                },
                                                            ),
                                                            prefix_len: 110,
                                                        },
                                                    ])
                                                    .map(Some)
                                                    .map(|addr| fnet_interfaces::Address {
                                                        addr,
                                                        valid_until: Some(1),
                                                        ..fnet_interfaces::Address::EMPTY
                                                    })
                                                    .collect(),
                                                ),
                                                online: Some(true),
                                                device_class: Some(
                                                    fnet_interfaces::DeviceClass::Loopback(
                                                        fnet_interfaces::Empty {},
                                                    ),
                                                ),
                                                has_default_ipv4_route: Some(false),
                                                has_default_ipv6_route: Some(false),
                                                name: Some(String::from("eth0")),
                                                ..fnet_interfaces::Properties::EMPTY
                                            },
                                        )
                                    } else {
                                        fnet_interfaces::Event::Idle(fnet_interfaces::Empty {})
                                    };
                                    let () = responder.send(&mut event).unwrap();
                                }
                            }
                        }
                    }
                }
            }
        })
        .detach();

        proxy
    }

    fn make_rcs() -> Rc<RemoteControlService> {
        make_rcs_with_maps(vec![])
    }

    fn make_rcs_with_maps(maps: Vec<(&str, &str)>) -> Rc<RemoteControlService> {
        Rc::new(RemoteControlService::new_with_allocator_and_maps(
            || {
                Ok(HostIdentifier {
                    interface_state_proxy: setup_fake_interface_state_service(),
                    name_provider_proxy: setup_fake_name_provider_service(),
                    device_info_proxy: setup_fake_device_service(),
                    build_info_proxy: setup_fake_build_info_service(),
                    boot_timestamp_nanos: BOOT_TIME,
                })
            },
            SelectorMappingList::new(
                maps.iter().map(|s| (s.0.to_string(), s.1.to_string())).collect(),
            ),
        ))
    }

    fn setup_rcs_proxy() -> rcs::RemoteControlProxy {
        let service = make_rcs();

        let (rcs_proxy, stream) =
            fidl::endpoints::create_proxy_and_stream::<rcs::RemoteControlMarker>().unwrap();
        fasync::Task::local(async move {
            service.serve_stream(stream).await;
        })
        .detach();

        return rcs_proxy;
    }

    #[fasync::run_singlethreaded(test)]
    async fn test_identify_host() -> Result<()> {
        let rcs_proxy = setup_rcs_proxy();

        let resp = rcs_proxy.identify_host().await.unwrap().unwrap();

        assert_eq!(resp.serial_number.unwrap(), SERIAL);
        assert_eq!(resp.board_config.unwrap(), BOARD_CONFIG);
        assert_eq!(resp.product_config.unwrap(), PRODUCT_CONFIG);
        assert_eq!(resp.nodename.unwrap(), NODENAME);

        let addrs = resp.addresses.unwrap();
        assert_eq!(
            addrs[..],
            [
                fnet::Subnet {
                    addr: fnet::IpAddress::Ipv4(fnet::Ipv4Address { addr: IPV4_ADDR }),
                    prefix_len: 4,
                },
                fnet::Subnet {
                    addr: fnet::IpAddress::Ipv6(fnet::Ipv6Address { addr: IPV6_ADDR }),
                    prefix_len: 110,
                }
            ]
        );

        assert_eq!(resp.boot_timestamp_nanos.unwrap(), BOOT_TIME);

        Ok(())
    }

    #[fasync::run_singlethreaded(test)]
    async fn test_ids_in_host_identify() -> Result<()> {
        let rcs_proxy = setup_rcs_proxy();

        let ident = rcs_proxy.identify_host().await.unwrap().unwrap();
        assert_eq!(ident.ids, Some(vec![]));

        rcs_proxy.add_id(1234).await.unwrap();
        rcs_proxy.add_id(4567).await.unwrap();

        let ident = rcs_proxy.identify_host().await.unwrap().unwrap();
        let ids = ident.ids.unwrap();
        assert_eq!(ids.len(), 2);
        assert_eq!(1234u64, ids[0]);
        assert_eq!(4567u64, ids[1]);

        Ok(())
    }

    fn wildcard_selector() -> Selector {
        parse_selector::<VerboseError>("*:*:*").unwrap()
    }

    fn service_selector() -> Selector {
        parse_selector::<VerboseError>(FAKE_SERVICE_SELECTOR).unwrap()
    }

    fn mapped_service_selector() -> Selector {
        parse_selector::<VerboseError>(MAPPED_SERVICE_SELECTOR).unwrap()
    }

    async fn no_paths_matcher() -> Result<Vec<PathEntry>> {
        Ok(vec![])
    }

    async fn two_paths_matcher() -> Result<Vec<PathEntry>> {
        Ok(vec![
            PathEntry {
                hub_path: PathBuf::from("/"),
                moniker: PathBuf::from("/a/b/c"),
                component_subdir: "out".to_string(),
                service: "myservice".to_string(),
            },
            PathEntry {
                hub_path: PathBuf::from("/"),
                moniker: PathBuf::from("/a/b/c"),
                component_subdir: "out".to_string(),
                service: "myservice2".to_string(),
            },
        ])
    }

    async fn single_path_matcher() -> Result<Vec<PathEntry>> {
        Ok(vec![PathEntry {
            hub_path: PathBuf::from("/tmp"),
            moniker: PathBuf::from("/tmp"),
            component_subdir: "out".to_string(),
            service: "myservice".to_string(),
        }])
    }

    #[fasync::run_singlethreaded(test)]
    async fn test_connect_no_matches() -> Result<()> {
        let service = make_rcs();
        let (_, server_end) = zx::Channel::create().unwrap();

        let result = service
            .connect_with_matcher(&wildcard_selector(), server_end, no_paths_matcher())
            .await;

        assert!(result.is_err());
        assert_eq!(result.unwrap_err(), rcs::ConnectError::NoMatchingServices);
        Ok(())
    }

    #[fasync::run_singlethreaded(test)]
    async fn test_connect_multiple_matches() -> Result<()> {
        let service = make_rcs();
        let (_, server_end) = zx::Channel::create().unwrap();

        let result = service
            .connect_with_matcher(&wildcard_selector(), server_end, two_paths_matcher())
            .await;

        assert!(result.is_err());
        assert_eq!(result.unwrap_err(), rcs::ConnectError::MultipleMatchingServices);
        Ok(())
    }

    #[fasync::run_singlethreaded(test)]
    async fn test_connect_single_match() -> Result<()> {
        let service = make_rcs();
        let (client_end, server_end) =
            fidl::endpoints::create_endpoints::<fio::NodeMarker>().unwrap();

        service
            .connect_with_matcher(
                &wildcard_selector(),
                server_end.into_channel(),
                single_path_matcher(),
            )
            .await
            .unwrap();

        // Make a dummy call to verify that the channel did get hooked up.
        assert!(client_end.into_proxy().unwrap().describe().await.is_ok());
        Ok(())
    }

    #[fasync::run_singlethreaded(test)]
    async fn test_map_selector() -> Result<()> {
        let service = make_rcs_with_maps(vec![(FAKE_SERVICE_SELECTOR, MAPPED_SERVICE_SELECTOR)]);

        assert_eq!(service.map_selector(service_selector()).unwrap(), mapped_service_selector());
        Ok(())
    }

    #[fasync::run_singlethreaded(test)]
    async fn test_map_selector_broken_mapping() -> Result<()> {
        let service = make_rcs_with_maps(vec![(FAKE_SERVICE_SELECTOR, "not_a_selector:::::")]);

        assert_matches!(
            service.map_selector(service_selector()).unwrap_err(),
            rcs::ConnectError::ServiceRerouteFailed
        );
        Ok(())
    }

    #[fasync::run_singlethreaded(test)]
    async fn test_map_selector_unbounded_mapping() -> Result<()> {
        let service = make_rcs_with_maps(vec![
            (FAKE_SERVICE_SELECTOR, MAPPED_SERVICE_SELECTOR),
            (MAPPED_SERVICE_SELECTOR, FAKE_SERVICE_SELECTOR),
        ]);

        assert_matches!(
            service.map_selector(service_selector()).unwrap_err(),
            rcs::ConnectError::ServiceRerouteFailed
        );
        Ok(())
    }

    #[fasync::run_singlethreaded(test)]
    async fn test_map_selector_no_matches() -> Result<()> {
        let service =
            make_rcs_with_maps(vec![("not/a/match:out:some.Service", MAPPED_SERVICE_SELECTOR)]);

        assert_eq!(service.map_selector(service_selector()).unwrap(), service_selector());
        Ok(())
    }

    #[fasync::run_singlethreaded(test)]
    async fn test_select_multiple_matches() -> Result<()> {
        let service = make_rcs();

        let result =
            service.select_with_matcher(&wildcard_selector(), two_paths_matcher()).await.unwrap();

        assert_eq!(result.len(), 2);
        assert!(result.iter().any(|p| *p
            == rcs::ServiceMatch {
                moniker: vec!["a".to_string(), "b".to_string(), "c".to_string()],
                subdir: "out".to_string(),
                service: "myservice".to_string()
            }));
        assert!(result.iter().any(|p| *p
            == rcs::ServiceMatch {
                moniker: vec!["a".to_string(), "b".to_string(), "c".to_string()],
                subdir: "out".to_string(),
                service: "myservice2".to_string()
            }));
        Ok(())
    }

    async fn create_forward_tunnel(
    ) -> (fasync::net::TcpStream, fasync::Socket, fasync::Task<Result<(), ForwardError>>) {
        let addr = (Ipv4Addr::LOCALHOST, 0).into();
        let listener = fasync::net::TcpListener::bind(&addr).unwrap();
        let listen_addr = listener.local_addr().unwrap();
        let mut listener_stream = listener.accept_stream();

        let (remote_tx, remote_rx) = futures::channel::oneshot::channel();

        // Run the listener in a background task so it can forward traffic in
        // parallel with the test.
        let forward_task = fasync::Task::local(async move {
            let (stream, _) = listener_stream.next().await.unwrap().unwrap();

            let (local, remote) = zx::Socket::create(zx::SocketOpts::STREAM).unwrap();
            let local = fasync::Socket::from_socket(local).unwrap();
            let remote = fasync::Socket::from_socket(remote).unwrap();

            remote_tx.send(remote).unwrap();

            forward_traffic(stream, local).await
        });

        // We should connect to the TCP socket, which should set us up a zircon socket.
        let tcp_stream = fasync::net::TcpStream::connect(listen_addr).unwrap().await.unwrap();
        let zx_socket = remote_rx.await.unwrap();

        (tcp_stream, zx_socket, forward_task)
    }

    #[fasync::run_singlethreaded(test)]
    async fn test_forward_traffic_tcp_closes_first() {
        let (mut tcp_stream, mut zx_socket, forward_task) = create_forward_tunnel().await;

        // Now any traffic that is sent to the tcp stream should come out of the zx socket.
        let msg = b"ping";
        tcp_stream.write_all(msg).await.unwrap();

        let mut buf = [0; 4096];
        zx_socket.read_exact(&mut buf[..msg.len()]).await.unwrap();
        assert_eq!(&buf[..msg.len()], msg);

        // Send a reply from the zx socket to the tcp stream.
        let msg = b"pong";
        zx_socket.write_all(msg).await.unwrap();

        tcp_stream.read_exact(&mut buf[..msg.len()]).await.unwrap();
        assert_eq!(&buf[..msg.len()], msg);

        // Now, close the tcp stream, this should cause the zx socket to close as well.
        std::mem::drop(tcp_stream);

        let mut buf = vec![];
        zx_socket.read_to_end(&mut buf).await.unwrap();
        assert_eq!(&buf, &Vec::<u8>::default());

        // Make sure the forward task shuts down as well.
        assert_matches!(forward_task.await, Ok(()));
    }

    #[fasync::run_singlethreaded(test)]
    async fn test_forward_traffic_zx_socket_closes_first() {
        let (mut tcp_stream, mut zx_socket, forward_task) = create_forward_tunnel().await;

        // Check that the zx socket can send the first data.
        let msg = b"ping";
        zx_socket.write_all(msg).await.unwrap();

        let mut buf = [0; 4096];
        tcp_stream.read_exact(&mut buf[..msg.len()]).await.unwrap();
        assert_eq!(&buf[..msg.len()], msg);

        let msg = b"pong";
        tcp_stream.write_all(msg).await.unwrap();

        zx_socket.read_exact(&mut buf[..msg.len()]).await.unwrap();
        assert_eq!(&buf[..msg.len()], msg);

        // Now, close the zx socket, this should cause the tcp stream to close as well.
        std::mem::drop(zx_socket);

        let mut buf = vec![];
        tcp_stream.read_to_end(&mut buf).await.unwrap();
        assert_eq!(&buf, &Vec::<u8>::default());

        // Make sure the forward task shuts down as well.
        assert_matches!(forward_task.await, Ok(()));
    }
}