dns_server_watcher/
stream.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//! DNS Server watcher stream.
6
7use fidl_fuchsia_net_name::{DnsServer_, DnsServerWatcherProxy};
8
9use async_utils::stream::WithTag as _;
10use futures::future::TryFutureExt as _;
11use futures::stream::Stream;
12
13/// The possible sources of DNS server updates.
14#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)]
15pub enum DnsServersUpdateSource {
16    Default,
17    Dhcpv4 { interface_id: u64 },
18    Dhcpv6 { interface_id: u64 },
19    Ndp { interface_id: u64 },
20    SocketProxy,
21}
22
23/// Returns a `Stream` of [`DnsServerWatcherEvent`]s from watching the server configuration
24/// provided by `proxy`.
25pub fn new_dns_server_stream(
26    source: DnsServersUpdateSource,
27    proxy: DnsServerWatcherProxy,
28) -> impl Stream<Item = (DnsServersUpdateSource, Result<Vec<DnsServer_>, fidl::Error>)> {
29    futures::stream::try_unfold(proxy, move |proxy| {
30        proxy.watch_servers().map_ok(move |s| Some((s, proxy)))
31    })
32    .tagged(source)
33}
34
35#[cfg(test)]
36mod tests {
37    use std::collections::VecDeque;
38    use std::sync::Arc;
39
40    use fidl_fuchsia_net_name::{
41        DnsServerWatcherMarker, DnsServerWatcherRequest, DnsServerWatcherRequestStream,
42        DnsServerWatcherWatchServersResponder,
43    };
44
45    use futures::lock::Mutex;
46    use futures::{FutureExt, StreamExt, TryStreamExt};
47
48    use super::*;
49    use crate::test_util::constants::*;
50
51    struct MockDnsServerWatcher {
52        configs: VecDeque<Vec<DnsServer_>>,
53        pending_request: Option<DnsServerWatcherWatchServersResponder>,
54    }
55
56    impl MockDnsServerWatcher {
57        fn new() -> Self {
58            Self { configs: VecDeque::new(), pending_request: None }
59        }
60
61        fn push_config(&mut self, config: Vec<DnsServer_>) {
62            match self.pending_request.take() {
63                Some(req) => {
64                    req.send(&config).expect("Failed to fulfill FIDL request");
65                }
66                None => self.configs.push_back(config),
67            }
68        }
69
70        async fn serve(
71            watcher: Arc<Mutex<Self>>,
72            rs: DnsServerWatcherRequestStream,
73        ) -> Result<(), fidl::Error> {
74            rs.try_for_each(move |r| {
75                let watcher = watcher.clone();
76                async move {
77                    match r {
78                        DnsServerWatcherRequest::WatchServers { responder } => {
79                            let mut w = watcher.lock().await;
80                            if w.pending_request.is_some() {
81                                panic!("No more than 1 pending requests allowed");
82                            }
83
84                            if let Some(config) = w.configs.pop_front() {
85                                responder.send(&config).expect("Failed to fulfill FIDL request");
86                            } else {
87                                w.pending_request = Some(responder)
88                            }
89                        }
90                    }
91                    Ok(())
92                }
93            })
94            .await
95        }
96    }
97
98    #[fuchsia::test]
99    async fn test_dns_server_stream() {
100        let watcher = Arc::new(Mutex::new(MockDnsServerWatcher::new()));
101        let (proxy, rs) = fidl::endpoints::create_proxy_and_stream::<DnsServerWatcherMarker>();
102        let (serve_fut, abort_handle) =
103            futures::future::abortable(MockDnsServerWatcher::serve(watcher.clone(), rs));
104
105        let (serve_result, mut stream) = futures::future::join(serve_fut, async move {
106            let mut stream = new_dns_server_stream(DnsServersUpdateSource::Default, proxy);
107            assert!(stream.next().now_or_never().is_none());
108            assert!(stream.next().now_or_never().is_none());
109            {
110                let mut w = watcher.lock().await;
111                w.push_config(vec![ndp_server()]);
112                w.push_config(vec![static_server()]);
113            }
114            let (source, res) = stream.next().await.expect("stream ended unexpectedly");
115            assert_eq!(source, DnsServersUpdateSource::Default);
116            assert_eq!(vec![ndp_server()], res.expect("FIDL error occurred"));
117
118            let (source, res) = stream.next().await.expect("stream ended unexpectedly");
119            assert_eq!(source, DnsServersUpdateSource::Default);
120            assert_eq!(vec![static_server()], res.expect("FIDL error occurred"));
121
122            // Abort the serving future so join will end.
123            abort_handle.abort();
124            stream
125        })
126        .await;
127        let _aborted = serve_result.expect_err("Future must've been aborted");
128        let (source, res) = stream.next().await.expect("Stream must yield a final value");
129        assert_eq!(source, DnsServersUpdateSource::Default);
130        let _fidl_error: fidl::Error = res.expect_err("Stream must yield an error");
131        assert!(stream.next().await.is_none(), "Stream must end after error");
132    }
133}
dns_server_watcher/stream.rs

dns_server_watcher/
stream.rs
