pub fn assignment_state_stream(
    address_state_provider: fnet_interfaces_admin::AddressStateProviderProxy,
) -> impl Stream<Item = Result<fnet_interfaces_admin::AddressAssignmentState, AddressStateProviderError>>
{
    let event_stream = address_state_provider
        .take_event_stream()
        .map_err(AddressStateProviderError::Fidl)
        .and_then(
            |fnet_interfaces_admin::AddressStateProviderEvent::OnAddressRemoved { error }| {
                futures::future::err(AddressStateProviderError::AddressRemoved(error))
            },
        );
    let state_stream =
        futures::stream::try_unfold(address_state_provider, |address_state_provider| {
            address_state_provider.watch_address_assignment_state().map(|r| match r {
                Ok(state) => Ok(Some((state, address_state_provider))),
                Err(e) => {
                    if e.is_closed() {
                        Ok(None)
                    } else {
                        Err(AddressStateProviderError::Fidl(e))
                    }
                }
            })
        });
    // TODO(https://github.com/rust-lang/futures-rs/issues/2476): Terminate the
    // stream upon the first error with something more terse than the
    // `take_while` block.
    //
    // FIDL errors other than PEER_CLOSED will be observed on both streams
    // (note that the `try_unfold` stream will swallow PEER_CLOSED so that the
    // address removal reason can be read from the event stream). Though this
    // is not a problem technically, we terminate the stream upon yielding
    // the first error to prevent the same error from being yielded twice.
    futures::stream::select(event_stream, state_stream).take_while({
        let mut error_observed = false;
        move |r| {
            futures::future::ready(match r {
                Ok(_) => true,
                Err(_) => !std::mem::replace(&mut error_observed, true),
            })
        }
    })
}
// TODO(https://fxbug.dev/81964): Introduce type with better concurrency safety
// for hanging gets.
/// Returns a stream of assignment states obtained by watching on `address_state_provider`.
///
/// Note that this function calls the hanging get FIDL method
/// [`AddressStateProviderProxy::watch_address_assignment_state`] internally,
/// which means that this stream should not be polled concurrently with any
/// logic which calls the same hanging get. This also means that callers should
/// be careful not to drop the returned stream when it has been polled but yet
/// to yield an item, e.g. due to a timeout or if using select with another
/// stream, as doing so causes a pending hanging get to get lost, and may cause
/// future hanging get calls to fail or the channel to be closed.
Introduce type with better concurrency safety
for hanging gets.
Returns a stream of assignment states obtained by watching on `address_state_provider`.

Note that this function calls the hanging get FIDL method
[`AddressStateProviderProxy::watch_address_assignment_state`] internally,
which means that this stream should not be polled concurrently with any
logic which calls the same hanging get. This also means that callers should
be careful not to drop the returned stream when it has been polled but yet
to yield an item, e.g. due to a timeout or if using select with another
stream, as doing so causes a pending hanging get to get lost, and may cause
future hanging get calls to fail or the channel to be closed.
Introduce type with better concurrency safety for hanging gets . Returns a stream of assignment states obtained by watching on ` address_state_provider ` . Note that this function calls the hanging get FIDL method [ ` AddressStateProviderProxy :: watch_address_assignment_state ` ] internally which means that this stream should not be polled concurrently with any logic which calls the same hanging get . This also means that callers should be careful not to drop the returned stream when it has been polled but yet to yield an item e . g . due to a timeout or if using select with another stream as doing so causes a pending hanging get to get lost and may cause future hanging get calls to fail or the channel to be closed .
pub fn assignment_state_stream(
    address_state_provider: fnet_interfaces_admin::AddressStateProviderProxy,
) -> impl Stream<Item = Result<fnet_interfaces_admin::AddressAssignmentState, AddressStateProviderError>>
{
    let event_stream = address_state_provider
        .take_event_stream()
        .map_err(AddressStateProviderError::Fidl)
        .and_then(
            |fnet_interfaces_admin::AddressStateProviderEvent::OnAddressRemoved { error }| {
                futures::future::err(AddressStateProviderError::AddressRemoved(error))
            },
        );
    let state_stream =
        futures::stream::try_unfold(address_state_provider, |address_state_provider| {
            address_state_provider.watch_address_assignment_state().map(|r| match r {
                Ok(state) => Ok(Some((state, address_state_provider))),
                Err(e) => {
                    if e.is_closed() {
                        Ok(None)
                    } else {
                        Err(AddressStateProviderError::Fidl(e))
                    }
                }
            })
        });
    // TODO(https://github.com/rust-lang/futures-rs/issues/2476): Terminate the
    // stream upon the first error with something more terse than the
    // `take_while` block.
    //
    // FIDL errors other than PEER_CLOSED will be observed on both streams
    // (note that the `try_unfold` stream will swallow PEER_CLOSED so that the
    // address removal reason can be read from the event stream). Though this
    // is not a problem technically, we terminate the stream upon yielding
    // the first error to prevent the same error from being yielded twice.
    futures::stream::select(event_stream, state_stream).take_while({
        let mut error_observed = false;
        move |r| {
            futures::future::ready(match r {
                Ok(_) => true,
                Err(_) => !std::mem::replace(&mut error_observed, true),
            })
        }
    })
}
pub fn assignment_state_stream ( address_state_provider : fnet_interfaces_admin :: AddressStateProviderProxy , ) -> impl Stream < Item = Result < fnet_interfaces_admin :: AddressAssignmentState , AddressStateProviderError > > { let event_stream = address_state_provider . take_event_stream ( ) . map_err ( AddressStateProviderError :: Fidl ) . and_then ( | fnet_interfaces_admin :: AddressStateProviderEvent :: OnAddressRemoved { error } | { futures :: future :: err ( AddressStateProviderError :: AddressRemoved ( error ) ) } , ) ; let state_stream = futures :: stream :: try_unfold ( address_state_provider , | address_state_provider | { address_state_provider . watch_address_assignment_state ( ) . map ( | r | match r { Ok ( state ) => Ok ( Some ( ( state , address_state_provider ) ) ) , Err ( e ) => { if e . is_closed ( ) { Ok ( None ) } else { Err ( AddressStateProviderError :: Fidl ( e ) ) } } } ) } ) ; // TODO(https://github.com/rust-lang/futures-rs/issues/2476): Terminate the // stream upon the first error with something more terse than the // `take_while` block. // // FIDL errors other than PEER_CLOSED will be observed on both streams // (note that the `try_unfold` stream will swallow PEER_CLOSED so that the // address removal reason can be read from the event stream). Though this // is not a problem technically, we terminate the stream upon yielding // the first error to prevent the same error from being yielded twice. futures :: stream :: select ( event_stream , state_stream ) . take_while ( { let mut error_observed = false ; move | r | { futures :: future :: ready ( match r { Ok ( _ ) => true , Err ( _ ) => ! std :: mem :: replace ( & mut error_observed , true ) , } ) } } ) }
TODO(https://fxbug.dev/81964): Introduce type with better concurrency safety
for hanging gets.
TODO ( https : // fxbug . dev / 81964 ) : Introduce type with better concurrency safety for hanging gets .
address_state_provider
fnet_interfaces_admin::AddressStateProviderProxy
address_state_provider
fnet_interfaces_admin::AddressStateProviderProxy