fidl_fuchsia_net_sockets_ext/
lib.rs

1// Copyright 2025 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 the fuchsia.sockets FIDL library.
6
7#![warn(
8    missing_docs,
9    unreachable_patterns,
10    clippy::useless_conversion,
11    clippy::redundant_clone,
12    clippy::precedence
13)]
14
15use fidl_fuchsia_net as fnet;
16use fidl_fuchsia_net_ext::{IntoExt, Marks};
17use fidl_fuchsia_net_matchers as fnet_matchers;
18use fidl_fuchsia_net_matchers_ext as fnet_matchers_ext;
19use fidl_fuchsia_net_sockets as fnet_sockets;
20use fidl_fuchsia_net_tcp as fnet_tcp;
21use fidl_fuchsia_net_udp as fnet_udp;
22use futures::{Stream, TryStreamExt as _};
23use net_types::ip::{self, GenericOverIp, Ip, IpInvariant, Ipv4, Ipv6};
24use thiserror::Error;
25
26/// An extension type for [`fnet_sockets::IpSocketMatcher`].
27#[derive(Debug, Clone, PartialEq, Eq)]
28pub enum IpSocketMatcher {
29    /// Matches against the IP version of the socket.
30    Family(ip::IpVersion),
31    /// Matches against the source address of the socket.
32    SrcAddr(fnet_matchers_ext::BoundAddress),
33    /// Matches against the destination address of the socket.
34    DstAddr(fnet_matchers_ext::BoundAddress),
35    /// Matches against transport protocol fields of the socket.
36    Proto(fnet_matchers_ext::SocketTransportProtocol),
37    /// Matches against the (bound, i.e. SO_BINDTODEVICE) interface of the
38    /// socket.
39    BoundInterface(fnet_matchers_ext::BoundInterface),
40    /// Matches against the cookie of the socket (i.e. SO_COOKIE)
41    Cookie(fnet_matchers::SocketCookie),
42    /// Matches against one mark of the socket.
43    Mark(fnet_matchers_ext::MarkInDomain),
44}
45
46/// Errors returned by the conversion from [`fnet_sockets::IpSocketMatcher`]
47/// to [`IpSocketMatcher`].
48#[derive(Debug, PartialEq, Error)]
49pub enum IpSocketMatcherError {
50    /// A union type was unknown.
51    #[error("got unexpected union variant: {0}")]
52    UnknownUnionVariant(u64),
53    /// An error was encountered when converting one of the address matchers.
54    #[error("address matcher conversion failure: {0}")]
55    Address(fnet_matchers_ext::BoundAddressError),
56    /// An error was encountered when converting the transport protocol
57    /// matcher.
58    #[error("protocol matcher conversion failure: {0}")]
59    TransportProtocol(fnet_matchers_ext::SocketTransportProtocolError),
60    /// An error was encountered while converting the interface matcher.
61    #[error("bound interface matcher conversion failure: {0}")]
62    BoundInterface(fnet_matchers_ext::BoundInterfaceError),
63    /// An error was encountered when converting one of the mark matchers.
64    #[error("mark matcher conversion failure: {0}")]
65    Mark(fnet_matchers_ext::MarkInDomainError),
66}
67
68impl TryFrom<fnet_sockets::IpSocketMatcher> for IpSocketMatcher {
69    type Error = IpSocketMatcherError;
70
71    fn try_from(matcher: fnet_sockets::IpSocketMatcher) -> Result<Self, Self::Error> {
72        match matcher {
73            fnet_sockets::IpSocketMatcher::Family(ip_version) => {
74                Ok(Self::Family(ip_version.into_ext()))
75            }
76            fnet_sockets::IpSocketMatcher::SrcAddr(addr) => {
77                Ok(Self::SrcAddr(addr.try_into().map_err(|e| IpSocketMatcherError::Address(e))?))
78            }
79            fnet_sockets::IpSocketMatcher::DstAddr(addr) => {
80                Ok(Self::DstAddr(addr.try_into().map_err(|e| IpSocketMatcherError::Address(e))?))
81            }
82            fnet_sockets::IpSocketMatcher::Proto(proto) => Ok(Self::Proto(
83                proto.try_into().map_err(|e| IpSocketMatcherError::TransportProtocol(e))?,
84            )),
85            fnet_sockets::IpSocketMatcher::BoundInterface(bound_interface) => {
86                Ok(Self::BoundInterface(
87                    bound_interface
88                        .try_into()
89                        .map_err(|e| IpSocketMatcherError::BoundInterface(e))?,
90                ))
91            }
92            fnet_sockets::IpSocketMatcher::Cookie(cookie) => Ok(Self::Cookie(cookie)),
93            fnet_sockets::IpSocketMatcher::Mark(mark) => {
94                Ok(Self::Mark(mark.try_into().map_err(|e| IpSocketMatcherError::Mark(e))?))
95            }
96            fnet_sockets::IpSocketMatcher::__SourceBreaking { unknown_ordinal } => {
97                Err(IpSocketMatcherError::UnknownUnionVariant(unknown_ordinal))
98            }
99        }
100    }
101}
102
103impl From<IpSocketMatcher> for fnet_sockets::IpSocketMatcher {
104    fn from(value: IpSocketMatcher) -> Self {
105        match value {
106            IpSocketMatcher::Family(ip_version) => {
107                fnet_sockets::IpSocketMatcher::Family(ip_version.into_ext())
108            }
109            IpSocketMatcher::SrcAddr(address) => {
110                fnet_sockets::IpSocketMatcher::SrcAddr(address.into())
111            }
112            IpSocketMatcher::DstAddr(address) => {
113                fnet_sockets::IpSocketMatcher::DstAddr(address.into())
114            }
115            IpSocketMatcher::Proto(socket_transport_protocol) => {
116                fnet_sockets::IpSocketMatcher::Proto(socket_transport_protocol.into())
117            }
118            IpSocketMatcher::BoundInterface(mark) => {
119                fnet_sockets::IpSocketMatcher::BoundInterface(mark.into())
120            }
121            IpSocketMatcher::Cookie(socket_cookie) => {
122                fnet_sockets::IpSocketMatcher::Cookie(socket_cookie)
123            }
124            IpSocketMatcher::Mark(mark) => fnet_sockets::IpSocketMatcher::Mark(mark.into()),
125        }
126    }
127}
128
129/// Extension type for [`fnet_sockets::IpSocketState`].
130#[derive(Debug, PartialEq, Eq, Clone)]
131pub enum IpSocketState {
132    /// IPv4 socket state.
133    V4(IpSocketStateSpecific<Ipv4>),
134    /// IPv6 socket state.
135    V6(IpSocketStateSpecific<Ipv6>),
136}
137
138/// Error type for [`IpSocketState`] conversion.
139#[derive(Debug, Error, PartialEq)]
140pub enum IpSocketStateError {
141    /// Missing a required field.
142    #[error("missing field: {0}")]
143    MissingField(&'static str),
144    /// The socket address version does not match the expected version.
145    #[error("version mismatch")]
146    VersionMismatch,
147    /// The transport state is invalid.
148    #[error("transport state error: {0}")]
149    Transport(IpSocketTransportStateError),
150}
151
152impl TryFrom<fnet_sockets::IpSocketState> for IpSocketState {
153    type Error = IpSocketStateError;
154
155    fn try_from(value: fnet_sockets::IpSocketState) -> Result<Self, Self::Error> {
156        fn convert_address<I: Ip>(addr: fnet::IpAddress) -> Result<I::Addr, IpSocketStateError> {
157            I::map_ip::<_, Option<I::Addr>>(
158                IpInvariant(addr.into_ext()),
159                |IpInvariant(addr)| match addr {
160                    net_types::ip::IpAddr::V4(addr) => Some(addr),
161                    _ => None,
162                },
163                |IpInvariant(addr)| match addr {
164                    net_types::ip::IpAddr::V6(addr) => Some(addr),
165                    _ => None,
166                },
167            )
168            .ok_or(IpSocketStateError::VersionMismatch)
169        }
170
171        fn to_ip_socket_specific<I: Ip>(
172            src_addr: Option<fnet::IpAddress>,
173            dst_addr: Option<fnet::IpAddress>,
174            cookie: u64,
175            marks: fnet::Marks,
176            transport: fnet_sockets::IpSocketTransportState,
177        ) -> Result<IpSocketStateSpecific<I>, IpSocketStateError> {
178            let src_addr: Option<I::Addr> = src_addr.map(convert_address::<I>).transpose()?;
179            let dst_addr: Option<I::Addr> = dst_addr.map(convert_address::<I>).transpose()?;
180
181            Ok(IpSocketStateSpecific {
182                src_addr,
183                dst_addr,
184                cookie,
185                marks: marks.into(),
186                transport: transport.try_into().map_err(IpSocketStateError::Transport)?,
187            })
188        }
189
190        let fnet_sockets::IpSocketState {
191            family,
192            src_addr,
193            dst_addr,
194            cookie,
195            marks,
196            transport,
197            __source_breaking,
198        } = value;
199
200        let family = family.ok_or(IpSocketStateError::MissingField("family"))?;
201        let cookie = cookie.ok_or(IpSocketStateError::MissingField("cookie"))?;
202        let marks = marks.ok_or(IpSocketStateError::MissingField("marks"))?;
203        let transport = transport.ok_or(IpSocketStateError::MissingField("transport"))?;
204
205        match family {
206            fnet::IpVersion::V4 => Ok(IpSocketState::V4(to_ip_socket_specific(
207                src_addr, dst_addr, cookie, marks, transport,
208            )?)),
209            fnet::IpVersion::V6 => Ok(IpSocketState::V6(to_ip_socket_specific(
210                src_addr, dst_addr, cookie, marks, transport,
211            )?)),
212        }
213    }
214}
215
216impl From<IpSocketState> for fnet_sockets::IpSocketState {
217    fn from(state: IpSocketState) -> Self {
218        match state {
219            IpSocketState::V4(state) => state.into(),
220            IpSocketState::V6(state) => state.into(),
221        }
222    }
223}
224
225/// Lowest-level socket state information that ensures all fields are for the
226/// same IP version.
227#[derive(Debug, PartialEq, Eq, Clone, GenericOverIp)]
228#[generic_over_ip(I, Ip)]
229pub struct IpSocketStateSpecific<I: Ip> {
230    /// The source address of the socket.
231    pub src_addr: Option<I::Addr>,
232    /// The destination address of the socket.
233    pub dst_addr: Option<I::Addr>,
234    /// The cookie of the socket.
235    pub cookie: u64,
236    /// The marks of the socket.
237    pub marks: Marks,
238    /// The transport state of the socket.
239    pub transport: IpSocketTransportState,
240}
241
242impl<I: Ip> From<IpSocketStateSpecific<I>> for fnet_sockets::IpSocketState {
243    fn from(value: IpSocketStateSpecific<I>) -> Self {
244        let IpSocketStateSpecific { src_addr, dst_addr, cookie, marks, transport } = value;
245
246        fnet_sockets::IpSocketState {
247            family: Some(I::VERSION.into_ext()),
248            src_addr: src_addr.map(|a| net_types::ip::IpAddr::from(a).into_ext()),
249            dst_addr: dst_addr.map(|a| net_types::ip::IpAddr::from(a).into_ext()),
250            cookie: Some(cookie),
251            marks: Some(marks.into()),
252            transport: Some(transport.into()),
253            __source_breaking: fidl::marker::SourceBreaking,
254        }
255    }
256}
257
258/// Extension type for [`fnet_sockets::IpSocketTransportState`].
259#[derive(Debug, PartialEq, Eq, Clone)]
260pub enum IpSocketTransportState {
261    /// TCP socket state.
262    Tcp(IpSocketTcpState),
263    /// UDP socket state.
264    Udp(IpSocketUdpState),
265}
266
267/// Error type for [`IpSocketTransportState`] conversion.
268#[derive(Debug, PartialEq, Error)]
269pub enum IpSocketTransportStateError {
270    /// Error converting a TCP socket state.
271    #[error("tcp validation error: {0}")]
272    Tcp(IpSocketTcpStateError),
273    /// Error converting a UDP socket state.
274    #[error("udp validation error: {0}")]
275    Udp(IpSocketUdpStateError),
276    /// A union type was unknown.
277    #[error("got unexpected union variant: {0}")]
278    UnknownUnionVariant(u64),
279}
280
281impl TryFrom<fnet_sockets::IpSocketTransportState> for IpSocketTransportState {
282    type Error = IpSocketTransportStateError;
283
284    fn try_from(value: fnet_sockets::IpSocketTransportState) -> Result<Self, Self::Error> {
285        match value {
286            fnet_sockets::IpSocketTransportState::Tcp(tcp) => Ok(IpSocketTransportState::Tcp(
287                tcp.try_into().map_err(IpSocketTransportStateError::Tcp)?,
288            )),
289            fnet_sockets::IpSocketTransportState::Udp(udp) => Ok(IpSocketTransportState::Udp(
290                udp.try_into().map_err(IpSocketTransportStateError::Udp)?,
291            )),
292            fnet_sockets::IpSocketTransportState::__SourceBreaking { unknown_ordinal } => {
293                Err(IpSocketTransportStateError::UnknownUnionVariant(unknown_ordinal))
294            }
295        }
296    }
297}
298
299impl From<IpSocketTransportState> for fnet_sockets::IpSocketTransportState {
300    fn from(state: IpSocketTransportState) -> Self {
301        match state {
302            IpSocketTransportState::Tcp(tcp) => {
303                fnet_sockets::IpSocketTransportState::Tcp(tcp.into())
304            }
305            IpSocketTransportState::Udp(udp) => {
306                fnet_sockets::IpSocketTransportState::Udp(udp.into())
307            }
308        }
309    }
310}
311
312/// Extension type for [`fnet_sockets::IpSocketTcpState`].
313#[derive(Debug, PartialEq, Eq, Clone)]
314pub struct IpSocketTcpState {
315    /// The source port of the socket.
316    pub src_port: Option<u16>,
317    /// The destination port of the socket.
318    pub dst_port: Option<u16>,
319    /// The TCP state machine state for the socket.
320    pub state: fnet_tcp::State,
321    /// Extended TCP information if the TCP_INFO extension was requested.
322    pub tcp_info: Option<TcpInfo>,
323}
324
325/// Error type for [`IpSocketTcpState`] conversion.
326#[derive(Debug, PartialEq, Error)]
327pub enum IpSocketTcpStateError {
328    /// Missing a required field.
329    #[error("missing field: {0}")]
330    MissingField(&'static str),
331    /// Error converting a [`TcpInfo`].
332    #[error("tcp info error: {0}")]
333    TcpInfo(TcpInfoError),
334}
335
336impl TryFrom<fnet_sockets::IpSocketTcpState> for IpSocketTcpState {
337    type Error = IpSocketTcpStateError;
338
339    fn try_from(value: fnet_sockets::IpSocketTcpState) -> Result<Self, Self::Error> {
340        let fnet_sockets::IpSocketTcpState {
341            src_port,
342            dst_port,
343            state,
344            tcp_info,
345            __source_breaking,
346        } = value;
347
348        let state = state.ok_or(IpSocketTcpStateError::MissingField("state"))?;
349
350        Ok(IpSocketTcpState {
351            src_port,
352            dst_port,
353            state,
354            tcp_info: tcp_info
355                .map(|t| t.try_into())
356                .transpose()
357                .map_err(|e| IpSocketTcpStateError::TcpInfo(e))?,
358        })
359    }
360}
361
362impl From<IpSocketTcpState> for fnet_sockets::IpSocketTcpState {
363    fn from(state: IpSocketTcpState) -> Self {
364        let IpSocketTcpState { src_port, dst_port, state, tcp_info } = state;
365        fnet_sockets::IpSocketTcpState {
366            src_port,
367            dst_port,
368            state: Some(state),
369            tcp_info: tcp_info.map(Into::into),
370            __source_breaking: fidl::marker::SourceBreaking,
371        }
372    }
373}
374
375/// Extension type for [`fnet_tcp::Info`].
376#[derive(Debug, PartialEq, Eq, Clone)]
377pub struct TcpInfo {
378    /// The state of the TCP connection.
379    pub state: fnet_tcp::State,
380    /// The congestion control state of the TCP connection.
381    pub ca_state: fnet_tcp::CongestionControlState,
382    /// The retransmission timeout of the TCP connection in microseconds.
383    pub rto_usec: Option<u32>,
384    /// The time since the most recent data was sent on the connection in milliseconds.
385    pub tcpi_last_data_sent_msec: Option<u32>,
386    /// The time since the most recent ACK was received in milliseconds.
387    pub tcpi_last_ack_recv_msec: Option<u32>,
388    /// The estimated smoothed roundtrip time in microseconds.
389    pub rtt_usec: Option<u32>,
390    /// The smoothed mean deviation of the roundtrip time in microseconds.
391    pub rtt_var_usec: Option<u32>,
392    /// The sending slow start threshold in segments.
393    pub snd_ssthresh: u32,
394    /// The current sending congestion window in segments.
395    pub snd_cwnd: u32,
396    /// The total number of retransmissions.
397    pub tcpi_total_retrans: u32,
398    /// The total number of segments sent.
399    pub tcpi_segs_out: u64,
400    /// The total number of segments received.
401    pub tcpi_segs_in: u64,
402    /// Whether reordering has been seen on the connection.
403    pub reorder_seen: bool,
404    /// The send MSS for this endpoint.
405    pub tcpi_snd_mss: Option<u32>,
406    /// The receive MSS for this endpoint.
407    pub tcpi_rcv_mss: Option<u32>,
408}
409
410/// Error type for [`TcpInfo`] conversion.
411#[derive(Debug, PartialEq, Error)]
412pub enum TcpInfoError {
413    /// Missing a required field.
414    #[error("missing field: {0}")]
415    MissingField(&'static str),
416}
417
418impl TryFrom<fnet_tcp::Info> for TcpInfo {
419    type Error = TcpInfoError;
420
421    fn try_from(value: fnet_tcp::Info) -> Result<Self, Self::Error> {
422        let fnet_tcp::Info {
423            state,
424            ca_state,
425            rto_usec,
426            tcpi_last_data_sent_msec,
427            tcpi_last_ack_recv_msec,
428            rtt_usec,
429            rtt_var_usec,
430            snd_ssthresh,
431            snd_cwnd,
432            tcpi_total_retrans,
433            tcpi_segs_out,
434            tcpi_segs_in,
435            reorder_seen,
436            tcpi_snd_mss,
437            tcpi_rcv_mss,
438            __source_breaking,
439        } = value;
440
441        Ok(TcpInfo {
442            state: state.ok_or(TcpInfoError::MissingField("state"))?,
443            ca_state: ca_state.ok_or(TcpInfoError::MissingField("ca_state"))?,
444            rto_usec,
445            tcpi_last_data_sent_msec,
446            tcpi_last_ack_recv_msec,
447            rtt_usec,
448            rtt_var_usec,
449            snd_ssthresh: snd_ssthresh.ok_or(TcpInfoError::MissingField("snd_ssthresh"))?,
450            snd_cwnd: snd_cwnd.ok_or(TcpInfoError::MissingField("snd_cwnd"))?,
451            tcpi_total_retrans: tcpi_total_retrans
452                .ok_or(TcpInfoError::MissingField("tcpi_total_retrans"))?,
453            tcpi_segs_out: tcpi_segs_out.ok_or(TcpInfoError::MissingField("tcpi_segs_out"))?,
454            tcpi_segs_in: tcpi_segs_in.ok_or(TcpInfoError::MissingField("tcpi_segs_in"))?,
455            reorder_seen: reorder_seen.ok_or(TcpInfoError::MissingField("reorder_seen"))?,
456            tcpi_snd_mss,
457            tcpi_rcv_mss,
458        })
459    }
460}
461
462impl From<TcpInfo> for fnet_tcp::Info {
463    fn from(info: TcpInfo) -> Self {
464        let TcpInfo {
465            state,
466            ca_state,
467            rto_usec,
468            tcpi_last_data_sent_msec,
469            tcpi_last_ack_recv_msec,
470            rtt_usec,
471            rtt_var_usec,
472            snd_ssthresh,
473            snd_cwnd,
474            tcpi_total_retrans,
475            tcpi_segs_out,
476            tcpi_segs_in,
477            reorder_seen,
478            tcpi_snd_mss,
479            tcpi_rcv_mss,
480        } = info;
481        fnet_tcp::Info {
482            state: Some(state),
483            ca_state: Some(ca_state),
484            rto_usec: rto_usec,
485            tcpi_last_data_sent_msec,
486            tcpi_last_ack_recv_msec,
487            rtt_usec: rtt_usec,
488            rtt_var_usec: rtt_var_usec,
489            snd_ssthresh: Some(snd_ssthresh),
490            snd_cwnd: Some(snd_cwnd),
491            tcpi_total_retrans: Some(tcpi_total_retrans),
492            tcpi_segs_out: Some(tcpi_segs_out),
493            tcpi_segs_in: Some(tcpi_segs_in),
494            reorder_seen: Some(reorder_seen),
495            tcpi_snd_mss,
496            tcpi_rcv_mss,
497            __source_breaking: fidl::marker::SourceBreaking,
498        }
499    }
500}
501
502/// Extension type for [`fnet_sockets::IpSocketUdpState`].
503#[derive(Debug, PartialEq, Eq, Clone)]
504pub struct IpSocketUdpState {
505    /// The source port of the socket.
506    pub src_port: Option<u16>,
507    /// The destination port of the socket.
508    pub dst_port: Option<u16>,
509    /// The UDP pseudo-state machine state for the socket.
510    pub state: fnet_udp::State,
511}
512
513/// Error type for [`IpSocketUdpState`] conversion.
514#[derive(Debug, PartialEq, Error)]
515pub enum IpSocketUdpStateError {
516    /// Missing a required field.
517    #[error("missing field: {0}")]
518    MissingField(&'static str),
519}
520
521impl TryFrom<fnet_sockets::IpSocketUdpState> for IpSocketUdpState {
522    type Error = IpSocketUdpStateError;
523
524    fn try_from(value: fnet_sockets::IpSocketUdpState) -> Result<Self, Self::Error> {
525        let fnet_sockets::IpSocketUdpState { src_port, dst_port, state, __source_breaking } = value;
526
527        let state = state.ok_or(IpSocketUdpStateError::MissingField("state"))?;
528
529        Ok(IpSocketUdpState { src_port, dst_port, state })
530    }
531}
532
533impl From<IpSocketUdpState> for fnet_sockets::IpSocketUdpState {
534    fn from(state: IpSocketUdpState) -> Self {
535        let IpSocketUdpState { src_port, dst_port, state } = state;
536        fnet_sockets::IpSocketUdpState {
537            src_port,
538            dst_port,
539            state: Some(state),
540            __source_breaking: fidl::marker::SourceBreaking,
541        }
542    }
543}
544
545/// Errors returned by [`iterate_ip`]
546#[derive(Debug, Error)]
547pub enum IterateIpError {
548    /// The specified matcher was the first invalid one.
549    #[error("invalid matcher at position {0}")]
550    InvalidMatcher(usize),
551    /// An unknown response was received on the call to `Diagnostics.IterateIp`
552    #[error("unknown ordinal on Diagnostics.IterateIp call: {0}")]
553    UnknownOrdinal(u64),
554    /// A low-level FIDL error was encountered on the call to
555    /// `Diagnostics.IterateIp`.
556    #[error("fidl error during Diagnostics.IterateIp call: {0}")]
557    Fidl(fidl::Error),
558}
559
560impl From<fidl::Error> for IterateIpError {
561    fn from(e: fidl::Error) -> Self {
562        IterateIpError::Fidl(e)
563    }
564}
565
566/// Errors returned by the stream returned from [`iterate_ip`].
567#[derive(Debug, Error)]
568pub enum IpIteratorError {
569    /// The netstack returned an empty batch of sockets
570    #[error("received empty batch of sockets")]
571    EmptyBatch,
572    /// A low-level FIDL error was encountered on the call to
573    /// `Diagnostics.IterateIp`.
574    #[error("fidl error during Diagnostics.IterateIp call: {0}")]
575    Fidl(fidl::Error),
576    /// An error was encountered while converting a socket state.
577    #[error("error converting socket state: {0}")]
578    Conversion(IpSocketStateError),
579}
580
581impl From<fidl::Error> for IpIteratorError {
582    fn from(e: fidl::Error) -> Self {
583        IpIteratorError::Fidl(e)
584    }
585}
586
587/// Send a request to `Diagnostics.IterateIp` and drive the resulting
588/// `IpIterator`.
589///
590/// `IpIterator` returns a series of batches of sockets matching the query, the
591/// returned stream flattens those batches into individual sockets. If an error
592/// is encuontered during iteration, it is returned and iteration halts.
593//
594// TODO(https://github.com/rust-lang/rust/issues/130043): Remove types from the
595// precise capturing clause on the stream.
596pub async fn iterate_ip<M, I>(
597    diagnostics: &fnet_sockets::DiagnosticsProxy,
598    extensions: fnet_sockets::Extensions,
599    matchers: M,
600) -> Result<impl Stream<Item = Result<IpSocketState, IpIteratorError>> + use<M, I>, IterateIpError>
601where
602    M: IntoIterator<Item = I>,
603    I: Into<fnet_sockets::IpSocketMatcher>,
604{
605    let (proxy, server_end) = fidl::endpoints::create_proxy::<fnet_sockets::IpIteratorMarker>();
606    match diagnostics
607        .iterate_ip(
608            server_end,
609            extensions,
610            &matchers.into_iter().map(Into::into).collect::<Vec<_>>()[..],
611        )
612        .await?
613    {
614        fnet_sockets::IterateIpResult::Ok(_empty) => Ok(()),
615        fnet_sockets::IterateIpResult::InvalidMatcher(fnet_sockets::InvalidMatcher { index }) => {
616            Err(IterateIpError::InvalidMatcher(index as usize))
617        }
618        fnet_sockets::IterateIpResult::__SourceBreaking { unknown_ordinal } => {
619            Err(IterateIpError::UnknownOrdinal(unknown_ordinal))
620        }
621    }?;
622
623    Ok(futures::stream::try_unfold((proxy, true), |(proxy, has_more)| async move {
624        if !has_more {
625            return Ok(None);
626        }
627
628        let (batch, has_more) = proxy.next().await?;
629        if batch.is_empty() && has_more {
630            Err(IpIteratorError::EmptyBatch)
631        } else {
632            Ok(Some((
633                futures::stream::iter(
634                    batch
635                        .into_iter()
636                        .map(|s| s.try_into().map_err(|e| IpIteratorError::Conversion(e))),
637                ),
638                (proxy, has_more),
639            )))
640        }
641    })
642    .try_flatten())
643}
644
645/// Errors returned by [`disconnect_ip`]
646#[derive(Debug, Error)]
647pub enum DisconnectIpError {
648    /// The specified matcher was the first invalid one.
649    #[error("invalid matcher at position {0}")]
650    InvalidMatcher(usize),
651    /// Specified matchers would a priori match all sockets.
652    #[error("matchers were unconstrained")]
653    UnconstrainedMatchers,
654    /// An unknown response was received on the call to `Control.DisconnectIp`
655    #[error("unknown ordinal on Control.DisconnectIp call: {0}")]
656    UnknownOrdinal(u64),
657    /// A low-level FIDL error was encountered on the call to
658    /// `Control.DisconnectIp`.
659    #[error("fidl error during Control.DisconnectIp call: {0}")]
660    Fidl(fidl::Error),
661}
662
663/// Send a request to `Control.DisconnectIp` with the provided matchers.
664pub async fn disconnect_ip<M, I>(
665    control: &fnet_sockets::ControlProxy,
666    matchers: M,
667) -> Result<usize, DisconnectIpError>
668where
669    M: IntoIterator<Item = I>,
670    I: Into<fnet_sockets::IpSocketMatcher>,
671{
672    match control
673        .disconnect_ip(&fnet_sockets::ControlDisconnectIpRequest {
674            matchers: Some(matchers.into_iter().map(Into::into).collect()),
675            __source_breaking: fidl::marker::SourceBreaking,
676        })
677        .await
678    {
679        Ok(r) => match r {
680            fnet_sockets::DisconnectIpResult::Ok(fnet_sockets::DisconnectIpResponse {
681                disconnected,
682            }) => {
683                // Unwrap is safe because usize is always at least u32.
684                Ok(disconnected.try_into().unwrap())
685            }
686            fnet_sockets::DisconnectIpResult::InvalidMatcher(fnet_sockets::InvalidMatcher {
687                index,
688            }) => {
689                // Unwrap is safe because usize is always at least u32.
690                Err(DisconnectIpError::InvalidMatcher(index.try_into().unwrap()))
691            }
692            fnet_sockets::DisconnectIpResult::UnconstrainedMatchers(fnet_sockets::Empty) => {
693                Err(DisconnectIpError::UnconstrainedMatchers)
694            }
695            fnet_sockets::DisconnectIpResult::__SourceBreaking { unknown_ordinal } => {
696                Err(DisconnectIpError::UnknownOrdinal(unknown_ordinal))
697            }
698        },
699        Err(e) => Err(DisconnectIpError::Fidl(e)),
700    }
701}
702
703/// Errors returned by the stream returned from [`watch_destruction`].
704#[derive(Debug, Error)]
705pub enum DestructionWatcherError {
706    /// A low-level FIDL error was encountered on the call to
707    /// `DestructionWatcher.Watch`.
708    #[error("fidl error during DestructionWatcher.Watch call: {0}")]
709    Fidl(fidl::Error),
710    /// An error was encountered while converting a socket state.
711    #[error("error converting socket state: {0}")]
712    Conversion(IpSocketStateError),
713    /// The netstack returned an empty batch of sockets.
714    #[error("received empty batch of sockets")]
715    EmptyBatch,
716    /// Events were dropped
717    #[error("events were dropped")]
718    DroppedEvents(u64),
719}
720
721impl From<fidl::Error> for DestructionWatcherError {
722    fn from(e: fidl::Error) -> Self {
723        DestructionWatcherError::Fidl(e)
724    }
725}
726
727/// Get a destruction watcher and drive it to yield individual sockets.
728pub async fn watch_destruction(
729    diagnostics: &fnet_sockets::DiagnosticsProxy,
730) -> Result<impl Stream<Item = Result<IpSocketState, DestructionWatcherError>> + use<>, fidl::Error>
731{
732    let (proxy, server_end) =
733        fidl::endpoints::create_proxy::<fnet_sockets::DestructionWatcherMarker>();
734    diagnostics.get_destruction_watcher(server_end).await?;
735
736    Ok(futures::stream::try_unfold(proxy, |proxy| async {
737        let (sockets, dropped_events) = proxy.watch().await?;
738
739        let items = if sockets.is_empty() {
740            vec![Err(DestructionWatcherError::EmptyBatch)]
741        } else {
742            let dropped_events_iter = if dropped_events > 0 {
743                Some(Err(DestructionWatcherError::DroppedEvents(dropped_events)))
744            } else {
745                None
746            };
747
748            sockets
749                .into_iter()
750                .map(|s| s.try_into().map_err(DestructionWatcherError::Conversion))
751                .chain(dropped_events_iter)
752                .collect()
753        };
754
755        Ok::<_, fidl::Error>(Some((futures::stream::iter(items), proxy)))
756    })
757    .try_flatten())
758}
759
760#[cfg(test)]
761mod tests {
762    use super::*;
763
764    use std::num::NonZeroU64;
765
766    use assert_matches::assert_matches;
767    use fidl_fuchsia_net as fnet;
768    use fidl_fuchsia_net_tcp as fnet_tcp;
769    use futures::{FutureExt as _, StreamExt as _, future, pin_mut};
770    use net_declare::{fidl_ip, fidl_subnet, net_ip_v4, net_ip_v6};
771    use test_case::test_case;
772
773    #[test_case(
774        fnet_sockets::IpSocketMatcher::Family(fnet::IpVersion::V4),
775        IpSocketMatcher::Family(ip::IpVersion::V4);
776        "FamilyIpv4"
777    )]
778    #[test_case(
779        fnet_sockets::IpSocketMatcher::Family(fnet::IpVersion::V6),
780        IpSocketMatcher::Family(ip::IpVersion::V6);
781        "FamilyIpv6"
782    )]
783    #[test_case(
784        fnet_sockets::IpSocketMatcher::SrcAddr(fnet_matchers::BoundAddress::Bound(
785            fnet_matchers::Address {
786                matcher: fnet_matchers::AddressMatcherType::Subnet(fidl_subnet!("192.0.2.0/24")),
787                invert: true,
788            }
789        )),
790        IpSocketMatcher::SrcAddr(fnet_matchers_ext::BoundAddress::Bound(
791            fnet_matchers_ext::Address {
792                matcher: fnet_matchers_ext::AddressMatcherType::Subnet(
793                    fnet_matchers_ext::Subnet::try_from(fidl_subnet!("192.0.2.0/24")).unwrap()
794                ),
795                invert: true,
796            }
797        ));
798        "SrcAddr"
799    )]
800    #[test_case(
801        fnet_sockets::IpSocketMatcher::DstAddr(fnet_matchers::BoundAddress::Bound(
802            fnet_matchers::Address {
803                matcher: fnet_matchers::AddressMatcherType::Subnet(fidl_subnet!("2001:db8::/32")),
804                invert: false,
805            }
806        )),
807        IpSocketMatcher::DstAddr(fnet_matchers_ext::BoundAddress::Bound(
808            fnet_matchers_ext::Address {
809                matcher: fnet_matchers_ext::AddressMatcherType::Subnet(
810                    fnet_matchers_ext::Subnet::try_from(fidl_subnet!("2001:db8::/32")).unwrap()
811                ),
812                invert: false,
813            }
814        ));
815        "DstAddr"
816    )]
817    #[test_case(
818        fnet_sockets::IpSocketMatcher::Proto(fnet_matchers::SocketTransportProtocol::Tcp(
819            fnet_matchers::TcpSocket::Empty(fnet_matchers::Empty)
820        )),
821        IpSocketMatcher::Proto(fnet_matchers_ext::SocketTransportProtocol::Tcp(
822            fnet_matchers_ext::TcpSocket::Empty
823        ));
824        "ProtoTcp"
825    )]
826    #[test_case(
827        fnet_sockets::IpSocketMatcher::Proto(fnet_matchers::SocketTransportProtocol::Udp(
828            fnet_matchers::UdpSocket::Empty(fnet_matchers::Empty)
829        )),
830        IpSocketMatcher::Proto(fnet_matchers_ext::SocketTransportProtocol::Udp(
831            fnet_matchers_ext::UdpSocket::Empty
832        ));
833        "ProtoUdp"
834    )]
835    #[test_case(
836        fnet_sockets::IpSocketMatcher::BoundInterface(fnet_matchers::BoundInterface::Unbound(
837            fnet_matchers::Empty
838        )),
839        IpSocketMatcher::BoundInterface(fnet_matchers_ext::BoundInterface::Unbound);
840        "BoundInterfaceUnbound"
841    )]
842    #[test_case(
843        fnet_sockets::IpSocketMatcher::BoundInterface(fnet_matchers::BoundInterface::Bound(
844            fnet_matchers::Interface::Id(1)
845        )),
846        IpSocketMatcher::BoundInterface(fnet_matchers_ext::BoundInterface::Bound(
847            fnet_matchers_ext::Interface::Id(NonZeroU64::new(1).unwrap())
848        ));
849        "BoundInterfaceBound"
850    )]
851    #[test_case(
852        fnet_sockets::IpSocketMatcher::Cookie(fnet_matchers::SocketCookie {
853            cookie: 12345,
854            invert: false,
855        }),
856        IpSocketMatcher::Cookie(fnet_matchers::SocketCookie {
857            cookie: 12345,
858            invert: false,
859        });
860        "Cookie"
861    )]
862    #[test_case(
863        fnet_sockets::IpSocketMatcher::Mark(fnet_matchers::MarkInDomain {
864            domain: fnet::MarkDomain::Mark1,
865            mark: fnet_matchers::Mark::Unmarked(fnet_matchers::Unmarked),
866        }),
867        IpSocketMatcher::Mark(fnet_matchers_ext::MarkInDomain {
868            domain: fnet::MarkDomain::Mark1,
869            mark: fnet_matchers_ext::Mark::Unmarked,
870        });
871        "Mark"
872    )]
873    #[test_case(
874        fnet_sockets::IpSocketMatcher::SrcAddr(fnet_matchers::BoundAddress::Unbound(fnet_matchers::Empty)),
875        IpSocketMatcher::SrcAddr(fnet_matchers_ext::BoundAddress::Unbound);
876        "SrcAddrUnbound"
877    )]
878    #[test_case(
879        fnet_sockets::IpSocketMatcher::DstAddr(fnet_matchers::BoundAddress::Unbound(fnet_matchers::Empty)),
880        IpSocketMatcher::DstAddr(fnet_matchers_ext::BoundAddress::Unbound);
881        "DstAddrUnbound"
882    )]
883    #[test_case(
884        fnet_sockets::IpSocketMatcher::Proto(fnet_matchers::SocketTransportProtocol::Tcp(
885            fnet_matchers::TcpSocket::SrcPort(fnet_matchers::BoundPort::Unbound(fnet_matchers::Empty))
886        )),
887        IpSocketMatcher::Proto(fnet_matchers_ext::SocketTransportProtocol::Tcp(
888            fnet_matchers_ext::TcpSocket::SrcPort(fnet_matchers_ext::BoundPort::Unbound)
889        ));
890        "ProtoTcpSrcPortUnbound"
891    )]
892    #[test_case(
893        fnet_sockets::IpSocketMatcher::Proto(fnet_matchers::SocketTransportProtocol::Tcp(
894            fnet_matchers::TcpSocket::DstPort(fnet_matchers::BoundPort::Unbound(fnet_matchers::Empty))
895        )),
896        IpSocketMatcher::Proto(fnet_matchers_ext::SocketTransportProtocol::Tcp(
897            fnet_matchers_ext::TcpSocket::DstPort(fnet_matchers_ext::BoundPort::Unbound)
898        ));
899        "ProtoTcpDstPortUnbound"
900    )]
901    #[test_case(
902        fnet_sockets::IpSocketMatcher::Proto(fnet_matchers::SocketTransportProtocol::Udp(
903            fnet_matchers::UdpSocket::SrcPort(fnet_matchers::BoundPort::Unbound(fnet_matchers::Empty))
904        )),
905        IpSocketMatcher::Proto(fnet_matchers_ext::SocketTransportProtocol::Udp(
906            fnet_matchers_ext::UdpSocket::SrcPort(fnet_matchers_ext::BoundPort::Unbound)
907        ));
908        "ProtoUdpSrcPortUnbound"
909    )]
910    #[test_case(
911        fnet_sockets::IpSocketMatcher::Proto(fnet_matchers::SocketTransportProtocol::Udp(
912            fnet_matchers::UdpSocket::DstPort(fnet_matchers::BoundPort::Unbound(fnet_matchers::Empty))
913        )),
914        IpSocketMatcher::Proto(fnet_matchers_ext::SocketTransportProtocol::Udp(
915            fnet_matchers_ext::UdpSocket::DstPort(fnet_matchers_ext::BoundPort::Unbound)
916        ));
917        "ProtoUdpDstPortUnbound"
918    )]
919    #[test_case(
920        fnet_tcp::Info {
921            state: Some(fnet_tcp::State::Established),
922            ca_state: Some(fnet_tcp::CongestionControlState::Open),
923            rto_usec: Some(1),
924            tcpi_last_data_sent_msec: Some(2),
925            tcpi_last_ack_recv_msec: Some(3),
926            rtt_usec: Some(4),
927            rtt_var_usec: Some(5),
928            snd_ssthresh: Some(6),
929            snd_cwnd: Some(7),
930            tcpi_total_retrans: Some(8),
931            tcpi_segs_out: Some(9),
932            tcpi_segs_in: Some(10),
933            reorder_seen: Some(true),
934            tcpi_snd_mss: Some(11),
935            tcpi_rcv_mss: Some(12),
936            __source_breaking: fidl::marker::SourceBreaking,
937        },
938        TcpInfo {
939            state: fnet_tcp::State::Established,
940            ca_state: fnet_tcp::CongestionControlState::Open,
941            rto_usec: Some(1),
942            tcpi_last_data_sent_msec: Some(2),
943            tcpi_last_ack_recv_msec: Some(3),
944            rtt_usec: Some(4),
945            rtt_var_usec: Some(5),
946            snd_ssthresh: 6,
947            snd_cwnd: 7,
948            tcpi_total_retrans: 8,
949            tcpi_segs_out: 9,
950            tcpi_segs_in: 10,
951            reorder_seen: true,
952            tcpi_snd_mss: Some(11),
953            tcpi_rcv_mss: Some(12),
954        };
955        "TcpInfo"
956    )]
957    #[test_case(
958        fnet_sockets::IpSocketState {
959            family: Some(fnet::IpVersion::V4),
960            src_addr: Some(fidl_ip!("192.168.1.1")),
961            dst_addr: Some(fidl_ip!("192.168.1.2")),
962            cookie: Some(1234),
963            marks: Some(fnet::Marks {
964                mark_1: Some(1111),
965                mark_2: None,
966                __source_breaking: fidl::marker::SourceBreaking,
967            }),
968            transport: Some(fnet_sockets::IpSocketTransportState::Tcp(
969                fnet_sockets::IpSocketTcpState {
970                    src_port: Some(1111),
971                    dst_port: Some(2222),
972                    state: Some(fnet_tcp::State::Established),
973                    tcp_info: None,
974                    __source_breaking: fidl::marker::SourceBreaking,
975                },
976            )),
977            __source_breaking: fidl::marker::SourceBreaking,
978        },
979        IpSocketState::V4(IpSocketStateSpecific {
980            src_addr: Some(net_ip_v4!("192.168.1.1")),
981            dst_addr: Some(net_ip_v4!("192.168.1.2")),
982            cookie: 1234,
983            marks: fnet::Marks {
984                mark_1: Some(1111),
985                mark_2: None,
986                __source_breaking: fidl::marker::SourceBreaking,
987            }.into(),
988            transport: IpSocketTransportState::Tcp(IpSocketTcpState {
989                src_port: Some(1111),
990                dst_port: Some(2222),
991                state: fnet_tcp::State::Established,
992                tcp_info: None,
993            }),
994        });
995        "IpSocketStateV4"
996    )]
997    #[test_case(
998        fnet_sockets::IpSocketState {
999            family: Some(fnet::IpVersion::V6),
1000            src_addr: Some(fidl_ip!("2001:db8::1")),
1001            dst_addr: Some(fidl_ip!("2001:db8::2")),
1002            cookie: Some(1234),
1003            marks: Some(fnet::Marks {
1004                mark_1: Some(1111),
1005                mark_2: None,
1006                __source_breaking: fidl::marker::SourceBreaking,
1007            }),
1008            transport: Some(fnet_sockets::IpSocketTransportState::Udp(
1009                fnet_sockets::IpSocketUdpState {
1010                    src_port: Some(3333),
1011                    dst_port: Some(4444),
1012                    state: Some(fnet_udp::State::Connected),
1013                    __source_breaking: fidl::marker::SourceBreaking,
1014                },
1015            )),
1016            __source_breaking: fidl::marker::SourceBreaking,
1017        },
1018        IpSocketState::V6(IpSocketStateSpecific {
1019            src_addr: Some(net_ip_v6!("2001:db8::1")),
1020            dst_addr: Some(net_ip_v6!("2001:db8::2")),
1021            cookie: 1234,
1022            marks: fnet::Marks {
1023                mark_1: Some(1111),
1024                mark_2: None,
1025                __source_breaking: fidl::marker::SourceBreaking,
1026            }.into(),
1027            transport: IpSocketTransportState::Udp(IpSocketUdpState {
1028                src_port: Some(3333),
1029                dst_port: Some(4444),
1030                state: fnet_udp::State::Connected,
1031            }),
1032        });
1033        "IpSocketStateV6"
1034    )]
1035    fn convert_from_fidl_and_back<F, E>(fidl_type: F, local_type: E)
1036    where
1037        E: TryFrom<F> + Clone + std::fmt::Debug + PartialEq,
1038        <E as TryFrom<F>>::Error: std::fmt::Debug + PartialEq,
1039        F: From<E> + Clone + std::fmt::Debug + PartialEq,
1040    {
1041        assert_eq!(fidl_type.clone().try_into(), Ok(local_type.clone()));
1042        assert_eq!(<_ as Into<F>>::into(local_type), fidl_type);
1043    }
1044
1045    #[test_case(
1046        fnet_sockets::IpSocketMatcher::__SourceBreaking { unknown_ordinal: 100 } =>
1047            Err(IpSocketMatcherError::UnknownUnionVariant(100));
1048        "UnknownUnionVariant"
1049    )]
1050    #[test_case(
1051        fnet_sockets::IpSocketMatcher::SrcAddr(fnet_matchers::BoundAddress::Bound(
1052            fnet_matchers::Address {
1053                matcher: fnet_matchers::AddressMatcherType::__SourceBreaking { unknown_ordinal: 100 },
1054                invert: false,
1055            }
1056        )) => Err(IpSocketMatcherError::Address(fnet_matchers_ext::BoundAddressError::Address(
1057            fnet_matchers_ext::AddressError::AddressMatcherType(
1058                fnet_matchers_ext::AddressMatcherTypeError::UnknownUnionVariant
1059            )
1060        )));
1061        "AddressError"
1062    )]
1063    #[test_case(
1064        fnet_sockets::IpSocketMatcher::Proto(
1065            fnet_matchers::SocketTransportProtocol::__SourceBreaking { unknown_ordinal: 100 }
1066        ) => Err(IpSocketMatcherError::TransportProtocol(
1067            fnet_matchers_ext::SocketTransportProtocolError::UnknownUnionVariant(100)
1068        ));
1069        "TransportProtocolError"
1070    )]
1071    #[test_case(
1072        fnet_sockets::IpSocketMatcher::BoundInterface(
1073            fnet_matchers::BoundInterface::__SourceBreaking { unknown_ordinal: 100 }
1074        ) => Err(IpSocketMatcherError::BoundInterface(
1075            fnet_matchers_ext::BoundInterfaceError::UnknownUnionVariant(100)
1076        ));
1077        "BoundInterfaceError"
1078    )]
1079    #[test_case(
1080        fnet_sockets::IpSocketMatcher::Mark(fnet_matchers::MarkInDomain {
1081            domain: fnet::MarkDomain::Mark1,
1082            mark: fnet_matchers::Mark::__SourceBreaking { unknown_ordinal: 100 },
1083        }) => Err(IpSocketMatcherError::Mark(
1084            fnet_matchers_ext::MarkInDomainError::Mark(
1085                fnet_matchers_ext::MarkError::UnknownUnionVariant(100)
1086            )
1087        ));
1088        "MarkError"
1089    )]
1090    fn ip_socket_matcher_try_from_error(
1091        fidl: fnet_sockets::IpSocketMatcher,
1092    ) -> Result<IpSocketMatcher, IpSocketMatcherError> {
1093        IpSocketMatcher::try_from(fidl)
1094    }
1095
1096    #[test_case(
1097        fnet_sockets::IpSocketState {
1098            family: None,
1099            src_addr: Some(fidl_ip!("192.168.1.1")),
1100            dst_addr: Some(fidl_ip!("192.168.1.2")),
1101            cookie: Some(1234),
1102            marks: Some(fnet::Marks {
1103                mark_1: Some(1111),
1104                mark_2: None,
1105                __source_breaking: fidl::marker::SourceBreaking,
1106            }),
1107            transport: Some(fnet_sockets::IpSocketTransportState::Tcp(
1108                fnet_sockets::IpSocketTcpState {
1109                    src_port: Some(1111),
1110                    dst_port: Some(2222),
1111                    state: Some(fnet_tcp::State::Established),
1112                    tcp_info: None,
1113                    __source_breaking: fidl::marker::SourceBreaking,
1114                },
1115            )),
1116            __source_breaking: fidl::marker::SourceBreaking,
1117        } => Err(IpSocketStateError::MissingField("family"));
1118        "MissingFamily"
1119    )]
1120    #[test_case(
1121        fnet_sockets::IpSocketState {
1122            family: Some(fnet::IpVersion::V4),
1123            src_addr: Some(fidl_ip!("192.168.1.1")),
1124            dst_addr: Some(fidl_ip!("192.168.1.2")),
1125            cookie: None,
1126            marks: Some(fnet::Marks {
1127                mark_1: Some(1111),
1128                mark_2: None,
1129                __source_breaking: fidl::marker::SourceBreaking,
1130            }),
1131            transport: Some(fnet_sockets::IpSocketTransportState::Tcp(
1132                fnet_sockets::IpSocketTcpState {
1133                    src_port: Some(1111),
1134                    dst_port: Some(2222),
1135                    state: Some(fnet_tcp::State::Established),
1136                    tcp_info: None,
1137                    __source_breaking: fidl::marker::SourceBreaking,
1138                },
1139            )),
1140            __source_breaking: fidl::marker::SourceBreaking,
1141        } => Err(IpSocketStateError::MissingField("cookie"));
1142        "MissingCookie"
1143    )]
1144    #[test_case(
1145        fnet_sockets::IpSocketState {
1146            family: Some(fnet::IpVersion::V4),
1147            src_addr: Some(fidl_ip!("192.168.1.1")),
1148            dst_addr: Some(fidl_ip!("192.168.1.2")),
1149            cookie: Some(1234),
1150            marks: None,
1151            transport: Some(fnet_sockets::IpSocketTransportState::Tcp(
1152                fnet_sockets::IpSocketTcpState {
1153                    src_port: Some(1111),
1154                    dst_port: Some(2222),
1155                    state: Some(fnet_tcp::State::Established),
1156                    tcp_info: None,
1157                    __source_breaking: fidl::marker::SourceBreaking,
1158                },
1159            )),
1160            __source_breaking: fidl::marker::SourceBreaking,
1161        } => Err(IpSocketStateError::MissingField("marks"));
1162        "MissingMarks"
1163    )]
1164    #[test_case(
1165        fnet_sockets::IpSocketState {
1166            family: Some(fnet::IpVersion::V4),
1167            src_addr: Some(fidl_ip!("192.168.1.1")),
1168            dst_addr: Some(fidl_ip!("192.168.1.2")),
1169            cookie: Some(1234),
1170            marks: Some(fnet::Marks {
1171                mark_1: Some(1111),
1172                mark_2: None,
1173                __source_breaking: fidl::marker::SourceBreaking,
1174            }),
1175            transport: None,
1176            __source_breaking: fidl::marker::SourceBreaking,
1177        } => Err(IpSocketStateError::MissingField("transport"));
1178        "MissingTransport"
1179    )]
1180    #[test_case(
1181        fnet_sockets::IpSocketState {
1182            family: Some(fnet::IpVersion::V4),
1183            src_addr: Some(fidl_ip!("192.168.1.1")),
1184            dst_addr: Some(fidl_ip!("2001:db8::2")),
1185            cookie: Some(1234),
1186            marks: Some(fnet::Marks {
1187                mark_1: Some(1111),
1188                mark_2: None,
1189                __source_breaking: fidl::marker::SourceBreaking,
1190            }),
1191            transport: Some(fnet_sockets::IpSocketTransportState::Tcp(
1192                fnet_sockets::IpSocketTcpState {
1193                    src_port: Some(1111),
1194                    dst_port: Some(2222),
1195                    state: Some(fnet_tcp::State::Established),
1196                    tcp_info: None,
1197                    __source_breaking: fidl::marker::SourceBreaking,
1198                },
1199            )),
1200            __source_breaking: fidl::marker::SourceBreaking,
1201        } => Err(IpSocketStateError::VersionMismatch);
1202        "VersionMismatchV4"
1203    )]
1204    #[test_case(
1205        fnet_sockets::IpSocketState {
1206            family: Some(fnet::IpVersion::V6),
1207            src_addr: Some(fidl_ip!("192.168.1.1")),
1208            dst_addr: Some(fidl_ip!("2001:db8::2")),
1209            cookie: Some(1234),
1210            marks: Some(fnet::Marks {
1211                mark_1: Some(1111),
1212                mark_2: None,
1213                __source_breaking: fidl::marker::SourceBreaking,
1214            }),
1215            transport: Some(fnet_sockets::IpSocketTransportState::Tcp(
1216                fnet_sockets::IpSocketTcpState {
1217                    src_port: Some(1111),
1218                    dst_port: Some(2222),
1219                    state: Some(fnet_tcp::State::Established),
1220                    tcp_info: None,
1221                    __source_breaking: fidl::marker::SourceBreaking,
1222                },
1223            )),
1224            __source_breaking: fidl::marker::SourceBreaking,
1225        } => Err(IpSocketStateError::VersionMismatch);
1226        "VersionMismatchV6"
1227    )]
1228    #[test_case(
1229        fnet_sockets::IpSocketState {
1230            family: Some(fnet::IpVersion::V4),
1231            src_addr: Some(fidl_ip!("192.168.1.1")),
1232            dst_addr: Some(fidl_ip!("192.168.1.2")),
1233            cookie: Some(1234),
1234            marks: Some(fnet::Marks {
1235                mark_1: Some(1111),
1236                mark_2: None,
1237                __source_breaking: fidl::marker::SourceBreaking,
1238            }),
1239            transport: Some(fnet_sockets::IpSocketTransportState::Tcp(
1240                fnet_sockets::IpSocketTcpState {
1241                    src_port: Some(1111),
1242                    dst_port: Some(2222),
1243                    state: None,
1244                    tcp_info: None,
1245                    __source_breaking: fidl::marker::SourceBreaking,
1246                },
1247            )),
1248            __source_breaking: fidl::marker::SourceBreaking,
1249        } => Err(IpSocketStateError::Transport(IpSocketTransportStateError::Tcp(
1250                IpSocketTcpStateError::MissingField("state"),
1251        )));
1252        "MissingTcpState"
1253    )]
1254    #[test_case(
1255        fnet_sockets::IpSocketState {
1256            family: Some(fnet::IpVersion::V6),
1257            src_addr: Some(fidl_ip!("2001:db8::1")),
1258            dst_addr: Some(fidl_ip!("2001:db8::2")),
1259            cookie: Some(1234),
1260            marks: Some(fnet::Marks {
1261                mark_1: Some(1111),
1262                mark_2: None,
1263                __source_breaking: fidl::marker::SourceBreaking,
1264            }),
1265            transport: Some(fnet_sockets::IpSocketTransportState::Udp(
1266                fnet_sockets::IpSocketUdpState {
1267                    src_port: Some(3333),
1268                    dst_port: Some(4444),
1269                    state: None,
1270                    __source_breaking: fidl::marker::SourceBreaking,
1271                },
1272            )),
1273            __source_breaking: fidl::marker::SourceBreaking,
1274        } => Err(IpSocketStateError::Transport(IpSocketTransportStateError::Udp(
1275                IpSocketUdpStateError::MissingField("state"),
1276        )));
1277        "MissingUdpState"
1278    )]
1279    fn ip_socket_state_try_from_error(
1280        fidl: fnet_sockets::IpSocketState,
1281    ) -> Result<IpSocketState, IpSocketStateError> {
1282        IpSocketState::try_from(fidl)
1283    }
1284
1285    #[fuchsia_async::run_singlethreaded(test)]
1286    async fn iterate_ip_diagnostics_iterate_ip_error() {
1287        async fn serve_matcher_error(req: fnet_sockets::DiagnosticsRequest) {
1288            match req {
1289                fnet_sockets::DiagnosticsRequest::IterateIp {
1290                    s: _,
1291                    extensions: _,
1292                    matchers: _,
1293                    responder,
1294                } => responder
1295                    .send(&fnet_sockets::IterateIpResult::InvalidMatcher(
1296                        fnet_sockets::InvalidMatcher { index: 0 },
1297                    ))
1298                    .unwrap(),
1299                fnet_sockets::DiagnosticsRequest::GetDestructionWatcher { .. } => unreachable!(),
1300            };
1301        }
1302
1303        let (diagnostics, diagnostics_server_end) =
1304            fidl::endpoints::create_proxy::<fnet_sockets::DiagnosticsMarker>();
1305
1306        let (mut diagnostics_request_stream, _control_handle) =
1307            diagnostics_server_end.into_stream_and_control_handle();
1308        let server_fut = diagnostics_request_stream
1309            .next()
1310            .then(|req| {
1311                serve_matcher_error(req.expect("Request stream ended unexpectedly").unwrap())
1312            })
1313            .fuse();
1314        let client_fut = iterate_ip::<[IpSocketMatcher; 0], _>(
1315            &diagnostics,
1316            fnet_sockets::Extensions::empty(),
1317            [],
1318        );
1319
1320        pin_mut!(server_fut);
1321        pin_mut!(client_fut);
1322
1323        let ((), resp) = future::join(server_fut, client_fut).await;
1324
1325        assert_matches!(
1326            // Discard the stream because it can't be formatted.
1327            resp.map(|_| ()),
1328            Err(IterateIpError::InvalidMatcher(0))
1329        );
1330    }
1331
1332    #[fuchsia_async::run_singlethreaded(test)]
1333    async fn iterate_ip_next_error() {
1334        async fn serve_matcher(req: fnet_sockets::DiagnosticsRequest) {
1335            match req {
1336                fnet_sockets::DiagnosticsRequest::IterateIp {
1337                    s,
1338                    extensions: _,
1339                    matchers: _,
1340                    responder,
1341                } => {
1342                    s.close_with_epitaph(zx_status::Status::PEER_CLOSED).unwrap();
1343                    responder.send(&fnet_sockets::IterateIpResult::Ok(fnet_sockets::Empty)).unwrap()
1344                }
1345                fnet_sockets::DiagnosticsRequest::GetDestructionWatcher { .. } => unreachable!(),
1346            }
1347        }
1348
1349        let (diagnostics, diagnostics_server_end) =
1350            fidl::endpoints::create_proxy::<fnet_sockets::DiagnosticsMarker>();
1351
1352        let (mut diagnostics_request_stream, _control_handle) =
1353            diagnostics_server_end.into_stream_and_control_handle();
1354        let server_fut = diagnostics_request_stream
1355            .next()
1356            .then(|req| serve_matcher(req.expect("Request stream ended unexpectedly").unwrap()))
1357            .fuse();
1358
1359        let client_fut = iterate_ip::<[IpSocketMatcher; 0], _>(
1360            &diagnostics,
1361            fnet_sockets::Extensions::empty(),
1362            [],
1363        );
1364
1365        let ((), resp) = future::join(server_fut, client_fut).await;
1366        let stream = resp.unwrap();
1367        pin_mut!(stream);
1368
1369        assert_matches!(
1370            stream.try_next().await,
1371            Err(IpIteratorError::Fidl(fidl::Error::ClientChannelClosed { .. }))
1372        );
1373    }
1374
1375    #[fuchsia_async::run_singlethreaded(test)]
1376    async fn iterate_ip_empty_batch() {
1377        async fn serve_matcher(req: fnet_sockets::DiagnosticsRequest) {
1378            match req {
1379                fnet_sockets::DiagnosticsRequest::IterateIp {
1380                    s,
1381                    extensions: _,
1382                    matchers: _,
1383                    responder,
1384                } => {
1385                    responder
1386                        .send(&fnet_sockets::IterateIpResult::Ok(fnet_sockets::Empty))
1387                        .unwrap();
1388
1389                    let (mut stream, _control) = s.into_stream_and_control_handle();
1390                    match stream.next().await.unwrap().unwrap() {
1391                        fidl_fuchsia_net_sockets::IpIteratorRequest::Next { responder } => {
1392                            // Send an empty batch but indicate there's more to come.
1393                            responder.send(&[], true).unwrap();
1394                        }
1395                        fidl_fuchsia_net_sockets::IpIteratorRequest::_UnknownMethod { .. } => {
1396                            unreachable!()
1397                        }
1398                    }
1399                }
1400                fnet_sockets::DiagnosticsRequest::GetDestructionWatcher { .. } => unreachable!(),
1401            }
1402        }
1403
1404        let (diagnostics, diagnostics_server_end) =
1405            fidl::endpoints::create_proxy::<fnet_sockets::DiagnosticsMarker>();
1406
1407        let (mut diagnostics_request_stream, _control_handle) =
1408            diagnostics_server_end.into_stream_and_control_handle();
1409        let server_fut = diagnostics_request_stream
1410            .next()
1411            .then(|req| serve_matcher(req.expect("Request stream ended unexpectedly").unwrap()))
1412            .fuse();
1413
1414        let client_fut = async {
1415            let stream = iterate_ip::<[IpSocketMatcher; 0], _>(
1416                &diagnostics,
1417                fnet_sockets::Extensions::empty(),
1418                [],
1419            )
1420            .await
1421            .unwrap();
1422            pin_mut!(stream);
1423            stream.try_next().await
1424        };
1425
1426        let ((), resp) = future::join(server_fut, client_fut).await;
1427        assert_matches!(resp, Err(IpIteratorError::EmptyBatch));
1428    }
1429
1430    #[fuchsia_async::run_singlethreaded(test)]
1431    async fn iterate_ip_success() {
1432        let socket_1 = fnet_sockets::IpSocketState {
1433            family: Some(fnet::IpVersion::V4),
1434            src_addr: Some(fidl_ip!("192.168.1.1")),
1435            dst_addr: Some(fidl_ip!("192.168.1.2")),
1436            cookie: Some(1234),
1437            marks: Some(fnet::Marks {
1438                mark_1: Some(1111),
1439                mark_2: None,
1440                __source_breaking: fidl::marker::SourceBreaking,
1441            }),
1442            transport: Some(fnet_sockets::IpSocketTransportState::Tcp(
1443                fnet_sockets::IpSocketTcpState {
1444                    src_port: Some(1111),
1445                    dst_port: Some(2222),
1446                    state: Some(fnet_tcp::State::Established),
1447                    tcp_info: None,
1448                    __source_breaking: fidl::marker::SourceBreaking,
1449                },
1450            )),
1451            __source_breaking: fidl::marker::SourceBreaking,
1452        };
1453
1454        let socket_2 = fnet_sockets::IpSocketState {
1455            family: Some(fnet::IpVersion::V4),
1456            src_addr: Some(fidl_ip!("192.168.8.1")),
1457            dst_addr: Some(fidl_ip!("192.168.8.2")),
1458            cookie: Some(9876),
1459            marks: Some(fnet::Marks {
1460                mark_1: None,
1461                mark_2: Some(2222),
1462                __source_breaking: fidl::marker::SourceBreaking,
1463            }),
1464            transport: Some(fnet_sockets::IpSocketTransportState::Tcp(
1465                fnet_sockets::IpSocketTcpState {
1466                    src_port: Some(3333),
1467                    dst_port: Some(4444),
1468                    state: Some(fnet_tcp::State::TimeWait),
1469                    tcp_info: None,
1470                    __source_breaking: fidl::marker::SourceBreaking,
1471                },
1472            )),
1473            __source_breaking: fidl::marker::SourceBreaking,
1474        };
1475
1476        let socket_3 = fnet_sockets::IpSocketState {
1477            family: Some(fnet::IpVersion::V6),
1478            src_addr: Some(fidl_ip!("2001:db8::1")),
1479            dst_addr: Some(fidl_ip!("2001:db8::2")),
1480            cookie: Some(5678),
1481            marks: Some(fnet::Marks {
1482                mark_1: None,
1483                mark_2: None,
1484                __source_breaking: fidl::marker::SourceBreaking,
1485            }),
1486            transport: Some(fnet_sockets::IpSocketTransportState::Tcp(
1487                fnet_sockets::IpSocketTcpState {
1488                    src_port: Some(5555),
1489                    dst_port: Some(6666),
1490                    state: Some(fnet_tcp::State::TimeWait),
1491                    tcp_info: None,
1492                    __source_breaking: fidl::marker::SourceBreaking,
1493                },
1494            )),
1495            __source_breaking: fidl::marker::SourceBreaking,
1496        };
1497
1498        let (diagnostics, diagnostics_server_end) =
1499            fidl::endpoints::create_proxy::<fnet_sockets::DiagnosticsMarker>();
1500
1501        let serve_matcher = async |req: fnet_sockets::DiagnosticsRequest| {
1502            let responses = &[vec![socket_1.clone()], vec![socket_2.clone(), socket_3.clone()]];
1503
1504            match req {
1505                fnet_sockets::DiagnosticsRequest::IterateIp {
1506                    s,
1507                    extensions: _,
1508                    matchers: _,
1509                    responder,
1510                } => {
1511                    responder
1512                        .send(&fnet_sockets::IterateIpResult::Ok(fnet_sockets::Empty))
1513                        .unwrap();
1514
1515                    let (mut stream, _control) = s.into_stream_and_control_handle();
1516                    for (i, resp) in responses.iter().enumerate() {
1517                        match stream.next().await.unwrap().unwrap() {
1518                            fidl_fuchsia_net_sockets::IpIteratorRequest::Next { responder } => {
1519                                let has_more = i < responses.len() - 1;
1520                                responder.send(&resp, has_more).unwrap();
1521                            }
1522                            fidl_fuchsia_net_sockets::IpIteratorRequest::_UnknownMethod {
1523                                ..
1524                            } => {
1525                                unreachable!()
1526                            }
1527                        }
1528                    }
1529                }
1530                fnet_sockets::DiagnosticsRequest::GetDestructionWatcher { .. } => unreachable!(),
1531            };
1532        };
1533
1534        let (mut diagnostics_request_stream, _control_handle) =
1535            diagnostics_server_end.into_stream_and_control_handle();
1536
1537        let server_fut = diagnostics_request_stream
1538            .next()
1539            .then(|req| serve_matcher(req.expect("Request stream ended unexpectedly").unwrap()))
1540            .fuse();
1541
1542        let client_fut = async {
1543            iterate_ip::<[IpSocketMatcher; 0], _>(
1544                &diagnostics,
1545                fnet_sockets::Extensions::empty(),
1546                [],
1547            )
1548            .await
1549            .unwrap()
1550            .try_collect::<Vec<_>>()
1551            .await
1552            .unwrap()
1553        };
1554
1555        let ((), sockets) = future::join(server_fut, client_fut).await;
1556        assert_eq!(
1557            sockets,
1558            vec![
1559                socket_1.clone().try_into().unwrap(),
1560                socket_2.clone().try_into().unwrap(),
1561                socket_3.clone().try_into().unwrap()
1562            ]
1563        );
1564    }
1565
1566    #[fuchsia_async::run_singlethreaded(test)]
1567    async fn watch_destruction_success() {
1568        let socket_1 = fnet_sockets::IpSocketState {
1569            family: Some(fnet::IpVersion::V4),
1570            src_addr: Some(fidl_ip!("192.168.1.1")),
1571            dst_addr: Some(fidl_ip!("192.168.1.2")),
1572            cookie: Some(1234),
1573            marks: Some(fnet::Marks {
1574                mark_1: Some(1111),
1575                mark_2: None,
1576                __source_breaking: fidl::marker::SourceBreaking,
1577            }),
1578            transport: Some(fnet_sockets::IpSocketTransportState::Tcp(
1579                fnet_sockets::IpSocketTcpState {
1580                    src_port: Some(1111),
1581                    dst_port: Some(2222),
1582                    state: Some(fnet_tcp::State::Established),
1583                    tcp_info: None,
1584                    __source_breaking: fidl::marker::SourceBreaking,
1585                },
1586            )),
1587            __source_breaking: fidl::marker::SourceBreaking,
1588        };
1589
1590        let socket_2 = fnet_sockets::IpSocketState {
1591            family: Some(fnet::IpVersion::V4),
1592            src_addr: Some(fidl_ip!("192.168.8.1")),
1593            dst_addr: Some(fidl_ip!("192.168.8.2")),
1594            cookie: Some(9876),
1595            marks: Some(fnet::Marks {
1596                mark_1: None,
1597                mark_2: Some(2222),
1598                __source_breaking: fidl::marker::SourceBreaking,
1599            }),
1600            transport: Some(fnet_sockets::IpSocketTransportState::Tcp(
1601                fnet_sockets::IpSocketTcpState {
1602                    src_port: Some(3333),
1603                    dst_port: Some(4444),
1604                    state: Some(fnet_tcp::State::TimeWait),
1605                    tcp_info: None,
1606                    __source_breaking: fidl::marker::SourceBreaking,
1607                },
1608            )),
1609            __source_breaking: fidl::marker::SourceBreaking,
1610        };
1611
1612        let (diagnostics, diagnostics_server_end) =
1613            fidl::endpoints::create_proxy::<fnet_sockets::DiagnosticsMarker>();
1614        let serve_watcher = async |req: fnet_sockets::DiagnosticsRequest| match req {
1615            fnet_sockets::DiagnosticsRequest::GetDestructionWatcher { watcher, responder } => {
1616                responder.send().unwrap();
1617                let mut stream = watcher.into_stream();
1618                let batches = [
1619                    Some(vec![socket_1.clone()]),
1620                    Some(vec![socket_2.clone(), socket_1.clone()]),
1621                    None,
1622                ];
1623                for batch in batches {
1624                    let req = stream.next().await.unwrap().unwrap();
1625                    let responder = match req {
1626                        fnet_sockets::DestructionWatcherRequest::Watch { responder } => responder,
1627                        fnet_sockets::DestructionWatcherRequest::_UnknownMethod { .. } => {
1628                            unreachable!()
1629                        }
1630                    };
1631                    if let Some(batch) = batch {
1632                        responder.send(&batch, 0).unwrap();
1633                    } else {
1634                        drop(responder);
1635                    }
1636                }
1637            }
1638            fnet_sockets::DiagnosticsRequest::IterateIp { .. } => unreachable!(),
1639        };
1640
1641        let (mut diagnostics_request_stream, _control_handle) =
1642            diagnostics_server_end.into_stream_and_control_handle();
1643        let server_fut = diagnostics_request_stream
1644            .next()
1645            .then(|req| serve_watcher(req.expect("Request stream ended unexpectedly").unwrap()))
1646            .fuse();
1647
1648        let expected_socket_1: IpSocketState = socket_1.clone().try_into().unwrap();
1649        let expected_socket_2: IpSocketState = socket_2.clone().try_into().unwrap();
1650
1651        let client_fut = async {
1652            let stream = watch_destruction(&diagnostics).await.unwrap();
1653            pin_mut!(stream);
1654
1655            assert_matches!(
1656                stream.next().await,
1657                Some(Ok(sock)) => assert_eq!(sock, expected_socket_1)
1658            );
1659            assert_matches!(
1660                stream.next().await,
1661                Some(Ok(sock)) => assert_eq!(sock, expected_socket_2)
1662            );
1663            assert_matches!(
1664                stream.next().await,
1665                Some(Ok(sock)) => assert_eq!(sock, expected_socket_1)
1666            );
1667            assert_matches!(stream.next().await, Some(Err(DestructionWatcherError::Fidl(_))));
1668        };
1669
1670        let _: ((), ()) = future::join(server_fut, client_fut).await;
1671    }
1672
1673    #[test_case(
1674        None,
1675        DestructionWatcherError::EmptyBatch;
1676        "empty_batch"
1677    )]
1678    #[test_case(
1679        Some(fnet_sockets::IpSocketState {
1680            family: None,
1681            src_addr: Some(fidl_ip!("192.168.1.1")),
1682            dst_addr: Some(fidl_ip!("192.168.1.2")),
1683            cookie: Some(1234),
1684            marks: Some(fnet::Marks {
1685                mark_1: Some(1111),
1686                mark_2: None,
1687                __source_breaking: fidl::marker::SourceBreaking,
1688            }),
1689            transport: Some(fnet_sockets::IpSocketTransportState::Tcp(
1690                fnet_sockets::IpSocketTcpState {
1691                    src_port: Some(1111),
1692                    dst_port: Some(2222),
1693                    state: Some(fnet_tcp::State::Established),
1694                    tcp_info: None,
1695                    __source_breaking: fidl::marker::SourceBreaking,
1696                },
1697            )),
1698            __source_breaking: fidl::marker::SourceBreaking,
1699        }),
1700        DestructionWatcherError::Conversion(IpSocketStateError::MissingField("family"));
1701        "conversion_error"
1702    )]
1703    #[fuchsia_async::run_singlethreaded(test)]
1704    async fn watch_destruction_error(
1705        socket: Option<fnet_sockets::IpSocketState>,
1706        expected_error: DestructionWatcherError,
1707    ) {
1708        let (diagnostics, diagnostics_server_end) =
1709            fidl::endpoints::create_proxy::<fnet_sockets::DiagnosticsMarker>();
1710        let serve_watcher = async |req: fnet_sockets::DiagnosticsRequest| match req {
1711            fnet_sockets::DiagnosticsRequest::GetDestructionWatcher { watcher, responder } => {
1712                responder.send().unwrap();
1713                let mut stream = watcher.into_stream();
1714                let req = stream.next().await.unwrap().unwrap();
1715                let responder = match req {
1716                    fnet_sockets::DestructionWatcherRequest::Watch { responder } => responder,
1717                    fnet_sockets::DestructionWatcherRequest::_UnknownMethod { .. } => {
1718                        unreachable!()
1719                    }
1720                };
1721                let batch = match socket {
1722                    None => vec![],
1723                    Some(s) => vec![s],
1724                };
1725                responder.send(&batch, 0).unwrap();
1726            }
1727            fnet_sockets::DiagnosticsRequest::IterateIp { .. } => unreachable!(),
1728        };
1729
1730        let (mut diagnostics_request_stream, _control_handle) =
1731            diagnostics_server_end.into_stream_and_control_handle();
1732        let server_fut = diagnostics_request_stream
1733            .next()
1734            .then(|req| serve_watcher(req.expect("Request stream ended unexpectedly").unwrap()))
1735            .fuse();
1736
1737        let client_fut = async {
1738            let stream = watch_destruction(&diagnostics).await.unwrap();
1739            pin_mut!(stream);
1740
1741            let result = stream.next().await.expect("got a result");
1742            result
1743        };
1744
1745        let ((), result) = future::join(server_fut, client_fut).await;
1746        match expected_error {
1747            DestructionWatcherError::EmptyBatch => {
1748                assert_matches!(result, Err(DestructionWatcherError::EmptyBatch));
1749            }
1750            DestructionWatcherError::Conversion(b) => {
1751                assert_matches!(result, Err(DestructionWatcherError::Conversion(a)) if a == b);
1752            }
1753            DestructionWatcherError::Fidl(_) | DestructionWatcherError::DroppedEvents(_) => {
1754                unreachable!()
1755            }
1756        }
1757    }
1758
1759    #[fuchsia_async::run_singlethreaded(test)]
1760    async fn watch_destruction_dropped_events() {
1761        let socket = fnet_sockets::IpSocketState {
1762            family: Some(fnet::IpVersion::V4),
1763            src_addr: Some(fidl_ip!("192.168.1.1")),
1764            dst_addr: Some(fidl_ip!("192.168.1.2")),
1765            cookie: Some(1234),
1766            marks: Some(fnet::Marks {
1767                mark_1: Some(1111),
1768                mark_2: None,
1769                __source_breaking: fidl::marker::SourceBreaking,
1770            }),
1771            transport: Some(fnet_sockets::IpSocketTransportState::Tcp(
1772                fnet_sockets::IpSocketTcpState {
1773                    src_port: Some(1111),
1774                    dst_port: Some(2222),
1775                    state: Some(fnet_tcp::State::Established),
1776                    tcp_info: None,
1777                    __source_breaking: fidl::marker::SourceBreaking,
1778                },
1779            )),
1780            __source_breaking: fidl::marker::SourceBreaking,
1781        };
1782
1783        let (diagnostics, diagnostics_server_end) =
1784            fidl::endpoints::create_proxy::<fnet_sockets::DiagnosticsMarker>();
1785        let serve_watcher = {
1786            let socket = socket.clone();
1787            async |req: fnet_sockets::DiagnosticsRequest| match req {
1788                fnet_sockets::DiagnosticsRequest::GetDestructionWatcher { watcher, responder } => {
1789                    responder.send().unwrap();
1790                    let mut stream = watcher.into_stream();
1791                    let req = stream.next().await.unwrap().unwrap();
1792                    let responder = match req {
1793                        fnet_sockets::DestructionWatcherRequest::Watch { responder } => responder,
1794                        fnet_sockets::DestructionWatcherRequest::_UnknownMethod { .. } => {
1795                            unreachable!()
1796                        }
1797                    };
1798                    responder.send(&[socket], 100).unwrap();
1799                }
1800                fnet_sockets::DiagnosticsRequest::IterateIp { .. } => unreachable!(),
1801            }
1802        };
1803
1804        let (mut diagnostics_request_stream, _control_handle) =
1805            diagnostics_server_end.into_stream_and_control_handle();
1806        let server_fut = diagnostics_request_stream
1807            .next()
1808            .then(|req| serve_watcher(req.expect("Request stream ended unexpectedly").unwrap()))
1809            .fuse();
1810
1811        let expected_socket: IpSocketState = socket.try_into().unwrap();
1812
1813        let client_fut = async {
1814            let stream = watch_destruction(&diagnostics).await.unwrap();
1815            pin_mut!(stream);
1816
1817            assert_matches!(
1818                stream.next().await,
1819                Some(Ok(sock)) => assert_eq!(sock, expected_socket)
1820            );
1821            assert_matches!(
1822                stream.next().await,
1823                Some(Err(DestructionWatcherError::DroppedEvents(100)))
1824            );
1825        };
1826
1827        let _: ((), ()) = future::join(server_fut, client_fut).await;
1828    }
1829}
fidl_fuchsia_net_sockets_ext/lib.rs

fidl_fuchsia_net_sockets_ext/
lib.rs
