pub(super) async fn state_watcher(peer: Arc<RwLock<RemotePeer>>) {
    fx_vlog!(tag: "avrcp", 2, "state_watcher starting");
    let mut change_stream = peer.read().state_change_listener.take_change_stream();
    let peer_weak = Arc::downgrade(&peer);
    drop(peer);

    let mut control_channel_abort_handle: Option<AbortHandle> = None;
    let mut browse_channel_abort_handle: Option<AbortHandle> = None;
    let mut notification_poll_abort_handle: Option<AbortHandle> = None;

    while let Some(_) = change_stream.next().await {
        fx_vlog!(tag: "avrcp", 2, "state_watcher command received");
        if let Some(peer) = peer_weak.upgrade() {
            let mut peer_guard = peer.write();

            fx_vlog!(tag: "avrcp", 2, "make_connection control channel {:?}", peer_guard.control_channel);
            match peer_guard.control_channel {
                PeerChannel::Connecting => {}
                PeerChannel::Disconnected => {
                    if let Some(ref abort_handle) = control_channel_abort_handle {
                        abort_handle.abort();
                        control_channel_abort_handle = None;
                    }

                    if let Some(ref abort_handle) = notification_poll_abort_handle {
                        abort_handle.abort();
                        notification_poll_abort_handle = None;
                    }

                    // Have we discovered service profile data on the peer?
                    if (peer_guard.target_descriptor.is_some()
                        || peer_guard.controller_descriptor.is_some())
                        && peer_guard.attempt_connection
                    {
                        fx_vlog!(tag: "avrcp", 2, "make_connection {:?}", peer_guard.peer_id);
                        peer_guard.attempt_connection = false;
                        peer_guard.control_channel = PeerChannel::Connecting;
                        start_make_connection_task(peer.clone());
                    }
                }
                PeerChannel::Connected(_) => {
                    // Have we discovered service profile data on the peer?
                    if (peer_guard.target_descriptor.is_some()
                        || peer_guard.controller_descriptor.is_some())
                        && control_channel_abort_handle.is_none()
                    {
                        control_channel_abort_handle =
                            Some(start_control_stream_processing_task(peer.clone()));
                    }

                    if peer_guard.target_descriptor.is_some()
                        && notification_poll_abort_handle.is_none()
                    {
                        notification_poll_abort_handle =
                            Some(start_notifications_processing_task(peer.clone()));
                    }
                }
            }

            match peer_guard.browse_channel {
                PeerChannel::Connecting => {}
                PeerChannel::Disconnected => {
                    if let Some(ref abort_handle) = browse_channel_abort_handle {
                        abort_handle.abort();
                        browse_channel_abort_handle = None;
                    }
                }
                PeerChannel::Connected(_) => {
                    // The Browse channel must be established after the control channel.
                    // Ensure that the control channel exists and the browse channel doesn't
                    // before spawning the browse processing task.
                    if control_channel_abort_handle.is_some()
                        && browse_channel_abort_handle.is_none()
                    {
                        browse_channel_abort_handle =
                            Some(start_browse_stream_processing_task(peer.clone()));
                    }
                }
            }
        } else {
            break;
        }
    }

    fx_vlog!(tag: "avrcp", 2, "state_watcher shutting down. aborting processors");

    // Stop processing state changes on the browse channel.
    // This needs to happen before stopping the control channel.
    if let Some(ref abort_handle) = browse_channel_abort_handle {
        abort_handle.abort();
    }

    // Stop processing state changes on the control channel.
    if let Some(ref abort_handle) = control_channel_abort_handle {
        abort_handle.abort();
    }

    if let Some(ref abort_handle) = notification_poll_abort_handle {
        abort_handle.abort();
    }
}
/// State observer task around a remote peer. Takes a change stream from the remote peer that wakes
/// the task whenever some state has changed on the peer. Swaps tasks such as making outgoing
/// connections, processing the incoming control messages, and registering for notifications on the
/// remote peer.
State observer task around a remote peer. Takes a change stream from the remote peer that wakes
the task whenever some state has changed on the peer. Swaps tasks such as making outgoing
connections, processing the incoming control messages, and registering for notifications on the
remote peer.
State observer task around a remote peer . Takes a change stream from the remote peer that wakes the task whenever some state has changed on the peer . Swaps tasks such as making outgoing connections processing the incoming control messages and registering for notifications on the remote peer .
pub(super) async fn state_watcher(peer: Arc<RwLock<RemotePeer>>) {
    fx_vlog!(tag: "avrcp", 2, "state_watcher starting");
    let mut change_stream = peer.read().state_change_listener.take_change_stream();
    let peer_weak = Arc::downgrade(&peer);
    drop(peer);
    let mut control_channel_abort_handle: Option<AbortHandle> = None;
    let mut browse_channel_abort_handle: Option<AbortHandle> = None;
    let mut notification_poll_abort_handle: Option<AbortHandle> = None;
    while let Some(_) = change_stream.next().await {
        fx_vlog!(tag: "avrcp", 2, "state_watcher command received");
        if let Some(peer) = peer_weak.upgrade() {
            let mut peer_guard = peer.write();
            fx_vlog!(tag: "avrcp", 2, "make_connection control channel {:?}", peer_guard.control_channel);
            match peer_guard.control_channel {
                PeerChannel::Connecting => {}
                PeerChannel::Disconnected => {
                    if let Some(ref abort_handle) = control_channel_abort_handle {
                        abort_handle.abort();
                        control_channel_abort_handle = None;
                    }
                    if let Some(ref abort_handle) = notification_poll_abort_handle {
                        abort_handle.abort();
                        notification_poll_abort_handle = None;
                    }
                    // Have we discovered service profile data on the peer?
                    if (peer_guard.target_descriptor.is_some()
                        || peer_guard.controller_descriptor.is_some())
                        && peer_guard.attempt_connection
                    {
                        fx_vlog!(tag: "avrcp", 2, "make_connection {:?}", peer_guard.peer_id);
                        peer_guard.attempt_connection = false;
                        peer_guard.control_channel = PeerChannel::Connecting;
                        start_make_connection_task(peer.clone());
                    }
                }
                PeerChannel::Connected(_) => {
                    // Have we discovered service profile data on the peer?
                    if (peer_guard.target_descriptor.is_some()
                        || peer_guard.controller_descriptor.is_some())
                        && control_channel_abort_handle.is_none()
                    {
                        control_channel_abort_handle =
                            Some(start_control_stream_processing_task(peer.clone()));
                    }
                    if peer_guard.target_descriptor.is_some()
                        && notification_poll_abort_handle.is_none()
                    {
                        notification_poll_abort_handle =
                            Some(start_notifications_processing_task(peer.clone()));
                    }
                }
            }
            match peer_guard.browse_channel {
                PeerChannel::Connecting => {}
                PeerChannel::Disconnected => {
                    if let Some(ref abort_handle) = browse_channel_abort_handle {
                        abort_handle.abort();
                        browse_channel_abort_handle = None;
                    }
                }
                PeerChannel::Connected(_) => {
                    // The Browse channel must be established after the control channel.
                    // Ensure that the control channel exists and the browse channel doesn't
                    // before spawning the browse processing task.
                    if control_channel_abort_handle.is_some()
                        && browse_channel_abort_handle.is_none()
                    {
                        browse_channel_abort_handle =
                            Some(start_browse_stream_processing_task(peer.clone()));
                    }
                }
            }
        } else {
            break;
        }
    }
    fx_vlog!(tag: "avrcp", 2, "state_watcher shutting down. aborting processors");
    // Stop processing state changes on the browse channel.
    // This needs to happen before stopping the control channel.
    if let Some(ref abort_handle) = browse_channel_abort_handle {
        abort_handle.abort();
    }
    // Stop processing state changes on the control channel.
    if let Some(ref abort_handle) = control_channel_abort_handle {
        abort_handle.abort();
    }
    if let Some(ref abort_handle) = notification_poll_abort_handle {
        abort_handle.abort();
    }
}
pub ( super ) async fn state_watcher ( peer : Arc < RwLock < RemotePeer > > ) { fx_vlog ! ( tag " " 2 " " ) ; let mut change_stream = peer . read ( ) . state_change_listener . take_change_stream ( ) ; let peer_weak = Arc :: downgrade ( & peer ) ; drop ( peer ) ; let mut control_channel_abort_handle : Option < AbortHandle > = None ; let mut browse_channel_abort_handle : Option < AbortHandle > = None ; let mut notification_poll_abort_handle : Option < AbortHandle > = None ; while let Some ( _ ) = change_stream . next ( ) . await { fx_vlog ! ( tag " " 2 " " ) ; if let Some ( peer ) = peer_weak . upgrade ( ) { let mut peer_guard = peer . write ( ) ; fx_vlog ! ( tag " " 2 " " peer_guard control_channel ) ; match peer_guard . control_channel { PeerChannel :: Connecting => { } PeerChannel :: Disconnected => { if let Some ( ref abort_handle ) = control_channel_abort_handle { abort_handle . abort ( ) ; control_channel_abort_handle = None ; } if let Some ( ref abort_handle ) = notification_poll_abort_handle { abort_handle . abort ( ) ; notification_poll_abort_handle = None ; } // Have we discovered service profile data on the peer? if ( peer_guard . target_descriptor . is_some ( ) || peer_guard . controller_descriptor . is_some ( ) ) && peer_guard . attempt_connection { fx_vlog ! ( tag " " 2 " " peer_guard peer_id ) ; peer_guard . attempt_connection = false ; peer_guard . control_channel = PeerChannel :: Connecting ; start_make_connection_task ( peer . clone ( ) ) ; } } PeerChannel :: Connected ( _ ) => { // Have we discovered service profile data on the peer? if ( peer_guard . target_descriptor . is_some ( ) || peer_guard . controller_descriptor . is_some ( ) ) && control_channel_abort_handle . is_none ( ) { control_channel_abort_handle = Some ( start_control_stream_processing_task ( peer . clone ( ) ) ) ; } if peer_guard . target_descriptor . is_some ( ) && notification_poll_abort_handle . is_none ( ) { notification_poll_abort_handle = Some ( start_notifications_processing_task ( peer . clone ( ) ) ) ; } } } match peer_guard . browse_channel { PeerChannel :: Connecting => { } PeerChannel :: Disconnected => { if let Some ( ref abort_handle ) = browse_channel_abort_handle { abort_handle . abort ( ) ; browse_channel_abort_handle = None ; } } PeerChannel :: Connected ( _ ) => { // The Browse channel must be established after the control channel. // Ensure that the control channel exists and the browse channel doesn't // before spawning the browse processing task. if control_channel_abort_handle . is_some ( ) && browse_channel_abort_handle . is_none ( ) { browse_channel_abort_handle = Some ( start_browse_stream_processing_task ( peer . clone ( ) ) ) ; } } } } else { break ; } } fx_vlog ! ( tag " " 2 " " ) ; // Stop processing state changes on the browse channel. // This needs to happen before stopping the control channel. if let Some ( ref abort_handle ) = browse_channel_abort_handle { abort_handle . abort ( ) ; } // Stop processing state changes on the control channel. if let Some ( ref abort_handle ) = control_channel_abort_handle { abort_handle . abort ( ) ; } if let Some ( ref abort_handle ) = notification_poll_abort_handle { abort_handle . abort ( ) ; } }
State observer task around a remote peer.
State observer task around a remote peer .
peer
Arc<RwLock<RemotePeer>>
peer
Arc<RwLock<RemotePeer>>