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 channel_processor_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) = channel_processor_abort_handle {
                        abort_handle.abort();
                        channel_processor_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())
                        && channel_processor_abort_handle.is_none()
                    {
                        channel_processor_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()));
                    }
                }
            }
        } else {
            break;
        }
    }

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

    // Stop processing state changes entirely on the peer.
    if let Some(ref abort_handle) = channel_processor_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 channel_processor_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) = channel_processor_abort_handle {
                        abort_handle.abort();
                        channel_processor_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())
                        && channel_processor_abort_handle.is_none()
                    {
                        channel_processor_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()));
                    }
                }
            }
        } else {
            break;
        }
    }
    fx_vlog!(tag: "avrcp", 2, "state_watcher shutting down. aborting processors");
    // Stop processing state changes entirely on the peer.
    if let Some(ref abort_handle) = channel_processor_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 channel_processor_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 ) = channel_processor_abort_handle { abort_handle . abort ( ) ; channel_processor_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 ( ) ) && channel_processor_abort_handle . is_none ( ) { channel_processor_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 ( ) ) ) ; } } } } else { break ; } } fx_vlog ! ( tag " " 2 " " ) ; // Stop processing state changes entirely on the peer. if let Some ( ref abort_handle ) = channel_processor_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>>