pub fn wrap_inbound_stream<'a, T>(
        &'a self,
        device_stream: T,
    ) -> impl Stream<Item = Result<(), Error>> + Send + 'a
    where
        T: futures::stream::Stream<Item = Result<Vec<u8>, Error>> + Send + 'a,
    {
        let wrapped_future = device_stream
            .and_then(move |f| {
                async move {
                    // Note that this really does need to be an async block
                    // in order to keep the lifetime of `f` in scope for the
                    // duration of the async call.
                    self.on_inbound_frame(&f).await
                }
            })
            .or_else(move |err| {
                futures::future::ready(match err {
                    // I/O errors are fatal.
                    err if err.is::<std::io::Error>() => Err(err),

                    // Cancelled errors are ignored at this level.
                    err if err.is::<Canceled>() => Ok(()),

                    // Other error cases may be added here in the future.
                    err => {
                        if self.driver_state.lock().is_waiting_for_reset() {
                            // Non-I/O errors while waiting for a reset are just a warning.
                            fx_log_warn!("inbound_frame_stream: Warning: {:?}", err);
                        } else {
                            // Non-I/O errors when not waiting for a reset will cause us to re-init.
                            fx_log_err!("inbound_frame_stream: Error: {:?}", err);
                            self.ncp_is_misbehaving();
                        }
                        Ok(())
                    }
                })
            });
        futures::stream::select(
            wrapped_future,
            self.take_main_task().boxed().map_err(|x| x.context("main_task")).into_stream(),
        )
    }
/// Wraps the given inbound frame stream with the
/// logic to parse and handle the inbound frames.
/// The resulting stream should be asynchronously
/// drained to process inbound frames.
///
/// In most cases you will want to call `take_inbound_stream` instead,
/// but if it is unavailable (or you want to add filters or whatnot)
/// then this is what you will use.
///
/// You may only wrap an inbound stream once. Calling this
/// method twice will cause a panic.
Wraps the given inbound frame stream with the
logic to parse and handle the inbound frames.
The resulting stream should be asynchronously
drained to process inbound frames.

In most cases you will want to call `take_inbound_stream` instead,
but if it is unavailable (or you want to add filters or whatnot)
then this is what you will use.

You may only wrap an inbound stream once. Calling this
method twice will cause a panic.
Wraps the given inbound frame stream with the logic to parse and handle the inbound frames . The resulting stream should be asynchronously drained to process inbound frames . In most cases you will want to call ` take_inbound_stream ` instead but if it is unavailable ( or you want to add filters or whatnot ) then this is what you will use . You may only wrap an inbound stream once . Calling this method twice will cause a panic .
pub fn wrap_inbound_stream<'a, T>(
        &'a self,
        device_stream: T,
    ) -> impl Stream<Item = Result<(), Error>> + Send + 'a
    where
        T: futures::stream::Stream<Item = Result<Vec<u8>, Error>> + Send + 'a,
    {
        let wrapped_future = device_stream
            .and_then(move |f| {
                async move {
                    // Note that this really does need to be an async block
                    // in order to keep the lifetime of `f` in scope for the
                    // duration of the async call.
                    self.on_inbound_frame(&f).await
                }
            })
            .or_else(move |err| {
                futures::future::ready(match err {
                    // I/O errors are fatal.
                    err if err.is::<std::io::Error>() => Err(err),
                    // Cancelled errors are ignored at this level.
                    err if err.is::<Canceled>() => Ok(()),
                    // Other error cases may be added here in the future.
                    err => {
                        if self.driver_state.lock().is_waiting_for_reset() {
                            // Non-I/O errors while waiting for a reset are just a warning.
                            fx_log_warn!("inbound_frame_stream: Warning: {:?}", err);
                        } else {
                            // Non-I/O errors when not waiting for a reset will cause us to re-init.
                            fx_log_err!("inbound_frame_stream: Error: {:?}", err);
                            self.ncp_is_misbehaving();
                        }
                        Ok(())
                    }
                })
            });
        futures::stream::select(
            wrapped_future,
            self.take_main_task().boxed().map_err(|x| x.context("main_task")).into_stream(),
        )
    }
pub fn wrap_inbound_stream < ' a , T > ( & ' a self , device_stream : T , ) -> impl Stream < Item = Result < ( ) , Error > > + Send + ' a where T : futures :: stream :: Stream < Item = Result < Vec < u8 > , Error > > + Send + ' a , { let wrapped_future = device_stream . and_then ( move | f | { async move { // Note that this really does need to be an async block // in order to keep the lifetime of `f` in scope for the // duration of the async call. self . on_inbound_frame ( & f ) . await } } ) . or_else ( move | err | { futures :: future :: ready ( match err { // I/O errors are fatal. err if err . is :: < std :: io :: Error > ( ) => Err ( err ) , // Cancelled errors are ignored at this level. err if err . is :: < Canceled > ( ) => Ok ( ( ) ) , // Other error cases may be added here in the future. err => { if self . driver_state . lock ( ) . is_waiting_for_reset ( ) { // Non-I/O errors while waiting for a reset are just a warning. fx_log_warn ! ( " " err ) ; } else { // Non-I/O errors when not waiting for a reset will cause us to re-init. fx_log_err ! ( " " err ) ; self . ncp_is_misbehaving ( ) ; } Ok ( ( ) ) } } ) } ) ; futures :: stream :: select ( wrapped_future , self . take_main_task ( ) . boxed ( ) . map_err ( | x | x . context ( " " ) ) . into_stream ( ) , ) }
Wraps the given inbound frame stream with the
logic to parse and handle the inbound frames.
Wraps the given inbound frame stream with the logic to parse and handle the inbound frames .
&'a self
device_stream
T
&'a self
device_stream
T