async fn decode_media_stream(
    mut stream: avdtp::MediaStream,
    encoding: String,
    sample_rate: u32,
    session_id: u64,
    mut end_signal: Receiver<()>,
    mut inspect: StreamingInspectData,
    cobalt_sender: CobaltSender,
) -> () {
    let mut player = match player::Player::new(session_id, encoding.clone(), sample_rate).await {
        Ok(v) => v,
        Err(e) => {
            fx_log_info!("Can't setup player for Media: {:?}", e);
            return;
        }
    };

    let start_time = zx::Time::get(zx::ClockId::Monotonic);
    inspect.stream_started();
    loop {
        select! {
            stream_packet = stream.next().fuse() => {
                let pkt = match stream_packet {
                    None => {
                        fx_log_info!("Media transport closed");
                        break;
                    },
                    Some(Err(e)) => {
                        fx_log_info!("Error in media stream: {:?}", e);
                        break;
                    }
                    Some(Ok(packet)) => packet,
                };
                if let Err(e) = player.push_payload(&pkt.as_slice())  {
                    fx_log_info!("can't push packet: {:?}", e);
                }
                if !player.playing() {
                    player.play().unwrap_or_else(|e| fx_log_info!("Problem playing: {:}", e));
                }
                inspect.accumulated_bytes += pkt.len() as u64;
            },
            player_event = player.next_event().fuse() => {
                match player_event {
                    player::PlayerEvent::Closed => {
                        fx_log_info!("Rebuilding Player");
                        // The player died somehow? Attempt to rebuild the player.
                        player = match player::Player::new(session_id, encoding.clone(), sample_rate).await {
                            Ok(v) => v,
                            Err(e) => {
                                fx_log_info!("Can't rebuild player: {:?}", e);
                                break;
                            }
                        };
                    },
                    player::PlayerEvent::Status(s) => {
                        fx_log_info!("PlayerEvent Status happened: {:?}", s);
                    }
                }
            },
            _ = inspect.update_interval.next() => {
                inspect.update_rx_statistics();
            }
            _ = end_signal.next().fuse() => {
                fx_log_info!("Stream ending on end signal");
                break;
            }
        }
    }
    let end_time = zx::Time::get(zx::ClockId::Monotonic);

    report_stream_metrics(cobalt_sender, &encoding, (end_time - start_time).into_seconds());
}
/// Decodes a media stream by starting a Player and transferring media stream packets from AVDTP
/// to the player.  Restarts the player on player errors.
/// Ends when signaled from `end_signal`, or when the media transport stream is closed.
Decodes a media stream by starting a Player and transferring media stream packets from AVDTP
to the player.  Restarts the player on player errors.
Ends when signaled from `end_signal`, or when the media transport stream is closed.
Decodes a media stream by starting a Player and transferring media stream packets from AVDTP to the player . Restarts the player on player errors . Ends when signaled from ` end_signal ` or when the media transport stream is closed .
async fn decode_media_stream(
    mut stream: avdtp::MediaStream,
    encoding: String,
    sample_rate: u32,
    session_id: u64,
    mut end_signal: Receiver<()>,
    mut inspect: StreamingInspectData,
    cobalt_sender: CobaltSender,
) -> () {
    let mut player = match player::Player::new(session_id, encoding.clone(), sample_rate).await {
        Ok(v) => v,
        Err(e) => {
            fx_log_info!("Can't setup player for Media: {:?}", e);
            return;
        }
    };
    let start_time = zx::Time::get(zx::ClockId::Monotonic);
    inspect.stream_started();
    loop {
        select! {
            stream_packet = stream.next().fuse() => {
                let pkt = match stream_packet {
                    None => {
                        fx_log_info!("Media transport closed");
                        break;
                    },
                    Some(Err(e)) => {
                        fx_log_info!("Error in media stream: {:?}", e);
                        break;
                    }
                    Some(Ok(packet)) => packet,
                };
                if let Err(e) = player.push_payload(&pkt.as_slice())  {
                    fx_log_info!("can't push packet: {:?}", e);
                }
                if !player.playing() {
                    player.play().unwrap_or_else(|e| fx_log_info!("Problem playing: {:}", e));
                }
                inspect.accumulated_bytes += pkt.len() as u64;
            },
            player_event = player.next_event().fuse() => {
                match player_event {
                    player::PlayerEvent::Closed => {
                        fx_log_info!("Rebuilding Player");
                        // The player died somehow? Attempt to rebuild the player.
                        player = match player::Player::new(session_id, encoding.clone(), sample_rate).await {
                            Ok(v) => v,
                            Err(e) => {
                                fx_log_info!("Can't rebuild player: {:?}", e);
                                break;
                            }
                        };
                    },
                    player::PlayerEvent::Status(s) => {
                        fx_log_info!("PlayerEvent Status happened: {:?}", s);
                    }
                }
            },
            _ = inspect.update_interval.next() => {
                inspect.update_rx_statistics();
            }
            _ = end_signal.next().fuse() => {
                fx_log_info!("Stream ending on end signal");
                break;
            }
        }
    }
    let end_time = zx::Time::get(zx::ClockId::Monotonic);
    report_stream_metrics(cobalt_sender, &encoding, (end_time - start_time).into_seconds());
}
async fn decode_media_stream ( mut stream : avdtp :: MediaStream , encoding : String , sample_rate : u32 , session_id : u64 , mut end_signal : Receiver < ( ) > , mut inspect : StreamingInspectData , cobalt_sender : CobaltSender , ) -> ( ) { let mut player = match player :: Player :: new ( session_id , encoding . clone ( ) , sample_rate ) . await { Ok ( v ) => v , Err ( e ) => { fx_log_info ! ( " " e ) ; return ; } } ; let start_time = zx :: Time :: get ( zx :: ClockId :: Monotonic ) ; inspect . stream_started ( ) ; loop { select ! { stream_packet stream next ( ) fuse ( ) { let pkt match stream_packet { None { fx_log_info ( " " ) break } Some ( Err ( e ) ) { fx_log_info ( " " e ) break } Some ( Ok ( packet ) ) packet } if let Err ( e ) player push_payload ( pkt as_slice ( ) ) { fx_log_info ( " " e ) } if player playing ( ) { player play ( ) unwrap_or_else ( e fx_log_info ( " " e ) ) } inspect accumulated_bytes pkt len ( ) as u64 } player_event player next_event ( ) fuse ( ) { match player_event { player PlayerEvent Closed { fx_log_info ( " " ) // The player died somehow? Attempt to rebuild the player. player match player Player new ( session_id encoding clone ( ) sample_rate ) await { Ok ( v ) v Err ( e ) { fx_log_info ( " " e ) break } } } player PlayerEvent Status ( s ) { fx_log_info ( " " s ) } } } inspect update_interval next ( ) { inspect update_rx_statistics ( ) } end_signal next ( ) fuse ( ) { fx_log_info ( " " ) break } } } let end_time = zx :: Time :: get ( zx :: ClockId :: Monotonic ) ; report_stream_metrics ( cobalt_sender , & encoding , ( end_time - start_time ) . into_seconds ( ) ) ; }
Decodes a media stream by starting a Player and transferring media stream packets from AVDTP
to the player.
Decodes a media stream by starting a Player and transferring media stream packets from AVDTP to the player .
stream
avdtp::MediaStream
encoding
String
sample_rate
u32
session_id
u64
end_signal
Receiver<()>
inspect
StreamingInspectData
cobalt_sender
CobaltSender