async fn send(
        &mut self,
        mode: ResponseMode,
        options: &SessionOptions,
        requested_auth: &mut Option<http_auth::PasswordClient>,
        req: &mut rtsp_types::Request<Bytes>,
    ) -> Result<(RtspMessageContext, u32, rtsp_types::Response<Bytes>), Error> {
        loop {
            let cseq = self.fill_req(options, requested_auth, req)?;
            self.inner
                .send(rtsp_types::Message::Request(req.clone()))
                .await
                .map_err(|e| wrap!(e))?;
            let method: &str = req.method().into();
            let (resp, msg_ctx) = loop {
                let msg = self.inner.next().await.unwrap_or_else(|| {
                    bail!(ErrorInt::RtspReadError {
                        conn_ctx: *self.inner.ctx(),
                        msg_ctx: self.inner.eof_ctx(),
                        source: std::io::Error::new(
                            std::io::ErrorKind::UnexpectedEof,
                            format!("EOF while expecting response to {} CSeq {}", method, cseq),
                        ),
                    })
                })?;
                let msg_ctx = msg.ctx;
                let description = match msg.msg {
                    rtsp_types::Message::Response(r) => {
                        if let Some(response_cseq) = parse::get_cseq(&r) {
                            if response_cseq == cseq {
                                break (r, msg_ctx);
                            }
                            if matches!(mode, ResponseMode::Teardown) {
                                debug!("ignoring unrelated response during TEARDOWN");
                                continue;
                            }
                            format!("{} response with CSeq {}", r.reason_phrase(), response_cseq)
                        } else {
                            format!("{} response with no/unparseable cseq", r.reason_phrase())
                        }
                    }
                    rtsp_types::Message::Data(d) => {
                        if matches!(mode, ResponseMode::Teardown { .. }) {
                            debug!("ignoring RTSP data during TEARDOWN");
                            continue;
                        } else if let (ResponseMode::Play, Some(m)) =
                            (&mode, self.channels.lookup(d.channel_id()))
                        {
                            debug!(
                                "ignoring interleaved data message on {:?} channel {} while \
                                     waiting for response to {} CSeq {}",
                                m.channel_type,
                                d.channel_id(),
                                method,
                                cseq
                            );
                            continue;
                        }

                        if let Some(session_group) = options.session_group.as_ref() {
                            session_group.clone().note_stale_live555_data(
                                tokio::runtime::Handle::try_current().ok(),
                            );
                        }

                        format!(
                            "{}-byte interleaved data message on channel {}",
                            d.len(),
                            d.channel_id()
                        )
                    }
                    rtsp_types::Message::Request(r) => format!("{:?} request", r.method()),
                };
                bail!(ErrorInt::RtspFramingError {
                    conn_ctx: *self.inner.ctx(),
                    msg_ctx,
                    description: format!(
                        "Expected response to {} CSeq {}, got {}",
                        method, cseq, description,
                    ),
                });
            };
            if resp.status() == rtsp_types::StatusCode::Unauthorized {
                if requested_auth.is_some() {
                    // TODO: the WWW-Authenticate might indicate a new domain or nonce.
                    // In that case, we should retry rather than returning error.
                    bail!(ErrorInt::RtspResponseError {
                        conn_ctx: *self.inner.ctx(),
                        msg_ctx,
                        method: req.method().clone(),
                        cseq,
                        status: resp.status(),
                        description: "Received Unauthorized after trying digest auth".into(),
                    })
                }
                let www_authenticate = match resp.header(&rtsp_types::headers::WWW_AUTHENTICATE) {
                    None => bail!(ErrorInt::RtspResponseError {
                        conn_ctx: *self.inner.ctx(),
                        msg_ctx,
                        method: req.method().clone(),
                        cseq,
                        status: resp.status(),
                        description: "Unauthorized without WWW-Authenticate header".into(),
                    }),
                    Some(h) => h,
                };
                if options.creds.is_none() {
                    bail!(ErrorInt::RtspResponseError {
                        conn_ctx: *self.inner.ctx(),
                        msg_ctx,
                        method: req.method().clone(),
                        cseq,
                        status: resp.status(),
                        description: "Authentication requested and no credentials supplied"
                            .to_owned(),
                    })
                }
                let www_authenticate = www_authenticate.as_str();
                *requested_auth = match http_auth::PasswordClient::try_from(www_authenticate) {
                    Ok(c) => Some(c),
                    Err(e) => bail!(ErrorInt::RtspResponseError {
                        conn_ctx: *self.inner.ctx(),
                        msg_ctx,
                        method: req.method().clone(),
                        cseq,
                        status: resp.status(),
                        description: format!("Can't understand WWW-Authenticate header: {}", e),
                    }),
                };
                continue;
            } else if !resp.status().is_success() {
                bail!(ErrorInt::RtspResponseError {
                    conn_ctx: *self.inner.ctx(),
                    msg_ctx,
                    method: req.method().clone(),
                    cseq,
                    status: resp.status(),
                    description: "Unexpected RTSP response status".into(),
                });
            }
            return Ok((msg_ctx, cseq, resp));
        }
    }
/// Sends a request and expects the next message from the peer to be its response.
/// Takes care of authorization and `CSeq`. Returns `Error` if not successful.
Sends a request and expects the next message from the peer to be its response.
Takes care of authorization and `CSeq`. Returns `Error` if not successful.
Sends a request and expects the next message from the peer to be its response . Takes care of authorization and ` CSeq ` . Returns ` Error ` if not successful .
async fn send(
        &mut self,
        mode: ResponseMode,
        options: &SessionOptions,
        requested_auth: &mut Option<http_auth::PasswordClient>,
        req: &mut rtsp_types::Request<Bytes>,
    ) -> Result<(RtspMessageContext, u32, rtsp_types::Response<Bytes>), Error> {
        loop {
            let cseq = self.fill_req(options, requested_auth, req)?;
            self.inner
                .send(rtsp_types::Message::Request(req.clone()))
                .await
                .map_err(|e| wrap!(e))?;
            let method: &str = req.method().into();
            let (resp, msg_ctx) = loop {
                let msg = self.inner.next().await.unwrap_or_else(|| {
                    bail!(ErrorInt::RtspReadError {
                        conn_ctx: *self.inner.ctx(),
                        msg_ctx: self.inner.eof_ctx(),
                        source: std::io::Error::new(
                            std::io::ErrorKind::UnexpectedEof,
                            format!("EOF while expecting response to {} CSeq {}", method, cseq),
                        ),
                    })
                })?;
                let msg_ctx = msg.ctx;
                let description = match msg.msg {
                    rtsp_types::Message::Response(r) => {
                        if let Some(response_cseq) = parse::get_cseq(&r) {
                            if response_cseq == cseq {
                                break (r, msg_ctx);
                            }
                            if matches!(mode, ResponseMode::Teardown) {
                                debug!("ignoring unrelated response during TEARDOWN");
                                continue;
                            }
                            format!("{} response with CSeq {}", r.reason_phrase(), response_cseq)
                        } else {
                            format!("{} response with no/unparseable cseq", r.reason_phrase())
                        }
                    }
                    rtsp_types::Message::Data(d) => {
                        if matches!(mode, ResponseMode::Teardown { .. }) {
                            debug!("ignoring RTSP data during TEARDOWN");
                            continue;
                        } else if let (ResponseMode::Play, Some(m)) =
                            (&mode, self.channels.lookup(d.channel_id()))
                        {
                            debug!(
                                "ignoring interleaved data message on {:?} channel {} while \
                                     waiting for response to {} CSeq {}",
                                m.channel_type,
                                d.channel_id(),
                                method,
                                cseq
                            );
                            continue;
                        }
                        if let Some(session_group) = options.session_group.as_ref() {
                            session_group.clone().note_stale_live555_data(
                                tokio::runtime::Handle::try_current().ok(),
                            );
                        }
                        format!(
                            "{}-byte interleaved data message on channel {}",
                            d.len(),
                            d.channel_id()
                        )
                    }
                    rtsp_types::Message::Request(r) => format!("{:?} request", r.method()),
                };
                bail!(ErrorInt::RtspFramingError {
                    conn_ctx: *self.inner.ctx(),
                    msg_ctx,
                    description: format!(
                        "Expected response to {} CSeq {}, got {}",
                        method, cseq, description,
                    ),
                });
            };
            if resp.status() == rtsp_types::StatusCode::Unauthorized {
                if requested_auth.is_some() {
                    // TODO: the WWW-Authenticate might indicate a new domain or nonce.
                    // In that case, we should retry rather than returning error.
                    bail!(ErrorInt::RtspResponseError {
                        conn_ctx: *self.inner.ctx(),
                        msg_ctx,
                        method: req.method().clone(),
                        cseq,
                        status: resp.status(),
                        description: "Received Unauthorized after trying digest auth".into(),
                    })
                }
                let www_authenticate = match resp.header(&rtsp_types::headers::WWW_AUTHENTICATE) {
                    None => bail!(ErrorInt::RtspResponseError {
                        conn_ctx: *self.inner.ctx(),
                        msg_ctx,
                        method: req.method().clone(),
                        cseq,
                        status: resp.status(),
                        description: "Unauthorized without WWW-Authenticate header".into(),
                    }),
                    Some(h) => h,
                };
                if options.creds.is_none() {
                    bail!(ErrorInt::RtspResponseError {
                        conn_ctx: *self.inner.ctx(),
                        msg_ctx,
                        method: req.method().clone(),
                        cseq,
                        status: resp.status(),
                        description: "Authentication requested and no credentials supplied"
                            .to_owned(),
                    })
                }
                let www_authenticate = www_authenticate.as_str();
                *requested_auth = match http_auth::PasswordClient::try_from(www_authenticate) {
                    Ok(c) => Some(c),
                    Err(e) => bail!(ErrorInt::RtspResponseError {
                        conn_ctx: *self.inner.ctx(),
                        msg_ctx,
                        method: req.method().clone(),
                        cseq,
                        status: resp.status(),
                        description: format!("Can't understand WWW-Authenticate header: {}", e),
                    }),
                };
                continue;
            } else if !resp.status().is_success() {
                bail!(ErrorInt::RtspResponseError {
                    conn_ctx: *self.inner.ctx(),
                    msg_ctx,
                    method: req.method().clone(),
                    cseq,
                    status: resp.status(),
                    description: "Unexpected RTSP response status".into(),
                });
            }
            return Ok((msg_ctx, cseq, resp));
        }
    }
async fn send ( & mut self , mode : ResponseMode , options : & SessionOptions , requested_auth : & mut Option < http_auth :: PasswordClient > , req : & mut rtsp_types :: Request < Bytes > , ) -> Result < ( RtspMessageContext , u32 , rtsp_types :: Response < Bytes > ) , Error > { loop { let cseq = self . fill_req ( options , requested_auth , req ) ? ; self . inner . send ( rtsp_types :: Message :: Request ( req . clone ( ) ) ) . await . map_err ( | e | wrap ! ( e ) ) ? ; let method : & str = req . method ( ) . into ( ) ; let ( resp , msg_ctx ) = loop { let msg = self . inner . next ( ) . await . unwrap_or_else ( | | { bail ! ( ErrorInt RtspReadError { conn_ctx self inner ctx ( ) msg_ctx self inner eof_ctx ( ) source std io Error new ( std io ErrorKind UnexpectedEof format ( " " method cseq ) ) } ) } ) ? ; let msg_ctx = msg . ctx ; let description = match msg . msg { rtsp_types :: Message :: Response ( r ) => { if let Some ( response_cseq ) = parse :: get_cseq ( & r ) { if response_cseq == cseq { break ( r , msg_ctx ) ; } if matches ! ( mode ResponseMode Teardown ) { debug ! ( " " ) ; continue ; } format ! ( " " r reason_phrase ( ) response_cseq ) } else { format ! ( " " r reason_phrase ( ) ) } } rtsp_types :: Message :: Data ( d ) => { if matches ! ( mode ResponseMode Teardown { } ) { debug ! ( " " ) ; continue ; } else if let ( ResponseMode :: Play , Some ( m ) ) = ( & mode , self . channels . lookup ( d . channel_id ( ) ) ) { debug ! ( " \
 " m channel_type d channel_id ( ) method cseq ) ; continue ; } if let Some ( session_group ) = options . session_group . as_ref ( ) { session_group . clone ( ) . note_stale_live555_data ( tokio :: runtime :: Handle :: try_current ( ) . ok ( ) , ) ; } format ! ( " " d len ( ) d channel_id ( ) ) } rtsp_types :: Message :: Request ( r ) => format ! ( " " r method ( ) ) , } ; bail ! ( ErrorInt RtspFramingError { conn_ctx self inner ctx ( ) msg_ctx description format ( " " method cseq description ) } ) ; } ; if resp . status ( ) == rtsp_types :: StatusCode :: Unauthorized { if requested_auth . is_some ( ) { // TODO: the WWW-Authenticate might indicate a new domain or nonce. // In that case, we should retry rather than returning error. bail ! ( ErrorInt RtspResponseError { conn_ctx self inner ctx ( ) msg_ctx method req method ( ) clone ( ) cseq status resp status ( ) description " " into ( ) } ) } let www_authenticate = match resp . header ( & rtsp_types :: headers :: WWW_AUTHENTICATE ) { None => bail ! ( ErrorInt RtspResponseError { conn_ctx self inner ctx ( ) msg_ctx method req method ( ) clone ( ) cseq status resp status ( ) description " " into ( ) } ) , Some ( h ) => h , } ; if options . creds . is_none ( ) { bail ! ( ErrorInt RtspResponseError { conn_ctx self inner ctx ( ) msg_ctx method req method ( ) clone ( ) cseq status resp status ( ) description " " to_owned ( ) } ) } let www_authenticate = www_authenticate . as_str ( ) ; * requested_auth = match http_auth :: PasswordClient :: try_from ( www_authenticate ) { Ok ( c ) => Some ( c ) , Err ( e ) => bail ! ( ErrorInt RtspResponseError { conn_ctx self inner ctx ( ) msg_ctx method req method ( ) clone ( ) cseq status resp status ( ) description format ( " " e ) } ) , } ; continue ; } else if ! resp . status ( ) . is_success ( ) { bail ! ( ErrorInt RtspResponseError { conn_ctx self inner ctx ( ) msg_ctx method req method ( ) clone ( ) cseq status resp status ( ) description " " into ( ) } ) ; } return Ok ( ( msg_ctx , cseq , resp ) ) ; } }
Sends a request and expects the next message from the peer to be its response.
Sends a request and expects the next message from the peer to be its response .
&mut self
mode
ResponseMode
options
&SessionOptions
requested_auth
&mut Option<http_auth::PasswordClient>
req
&mut rtsp_types::Request<Bytes>