fn process_mtp_buffer(
        &mut self,
        updates: &mut Vec<tl::enums::Updates>,
    ) -> Result<(), mtp::DeserializeError> {
        debug!("deserializing valid transport packet...");
        let result = self.mtp.deserialize(&self.mtp_buffer)?;

        updates.extend(result.updates.iter().filter_map(|update| {
            match tl::enums::Updates::from_bytes(&update) {
                Ok(u) => Some(u),
                Err(e) => {
                    warn!(
                        "telegram sent updates that failed to be deserialized: {}",
                        e
                    );
                    None
                }
            }
        }));

        for (msg_id, ret) in result.rpc_results {
            let mut found = false;
            for i in (0..self.requests.len()).rev() {
                let req = &mut self.requests[i];
                match req.state {
                    RequestState::Serialized(sid) if sid == msg_id => {
                        panic!("got rpc result {:?} for unsent request {:?}", msg_id, sid);
                    }
                    RequestState::Sent(sid) if sid == msg_id => {
                        found = true;
                        let result = match ret {
                            Ok(x) => {
                                assert!(x.len() >= 4);
                                let res_id = u32::from_le_bytes([x[0], x[1], x[2], x[3]]);
                                debug!(
                                    "got result {:x} ({}) for request {:?}",
                                    res_id,
                                    tl::name_for_id(res_id),
                                    msg_id
                                );
                                Ok(x)
                            }
                            Err(mtp::RequestError::RpcError(mut error)) => {
                                debug!("got rpc error {:?} for request {:?}", error, msg_id);
                                let x = req.body.as_slice();
                                error.caused_by =
                                    Some(u32::from_le_bytes([x[0], x[1], x[2], x[3]]));
                                Err(InvocationError::Rpc(error))
                            }
                            Err(mtp::RequestError::Dropped) => {
                                debug!("response for request {:?} dropped", msg_id);
                                Err(InvocationError::Dropped)
                            }
                            Err(mtp::RequestError::Deserialize(error)) => {
                                debug!(
                                    "got deserialize error {:?} for request {:?}",
                                    error, msg_id
                                );
                                Err(InvocationError::Read(error.into()))
                            }
                            Err(err @ mtp::RequestError::BadMessage { .. }) => {
                                // TODO add a test to make sure we resend the request
                                info!("{}; re-sending request {:?}", err, msg_id);
                                req.state = RequestState::NotSerialized;
                                break;
                            }
                        };

                        let req = self.requests.remove(i);
                        drop(req.result.send(result));
                        break;
                    }
                    _ => {}
                }
            }

            if !found {
                info!("got rpc result {:?} but no such request is saved", msg_id);
            }
        }

        Ok(())
    }
/// Process the `mtp_buffer` contents and dispatch the results and errors.
Process the `mtp_buffer` contents and dispatch the results and errors.
Process the ` mtp_buffer ` contents and dispatch the results and errors .
fn process_mtp_buffer(
        &mut self,
        updates: &mut Vec<tl::enums::Updates>,
    ) -> Result<(), mtp::DeserializeError> {
        debug!("deserializing valid transport packet...");
        let result = self.mtp.deserialize(&self.mtp_buffer)?;
        updates.extend(result.updates.iter().filter_map(|update| {
            match tl::enums::Updates::from_bytes(&update) {
                Ok(u) => Some(u),
                Err(e) => {
                    warn!(
                        "telegram sent updates that failed to be deserialized: {}",
                        e
                    );
                    None
                }
            }
        }));
        for (msg_id, ret) in result.rpc_results {
            let mut found = false;
            for i in (0..self.requests.len()).rev() {
                let req = &mut self.requests[i];
                match req.state {
                    RequestState::Serialized(sid) if sid == msg_id => {
                        panic!("got rpc result {:?} for unsent request {:?}", msg_id, sid);
                    }
                    RequestState::Sent(sid) if sid == msg_id => {
                        found = true;
                        let result = match ret {
                            Ok(x) => {
                                assert!(x.len() >= 4);
                                let res_id = u32::from_le_bytes([x[0], x[1], x[2], x[3]]);
                                debug!(
                                    "got result {:x} ({}) for request {:?}",
                                    res_id,
                                    tl::name_for_id(res_id),
                                    msg_id
                                );
                                Ok(x)
                            }
                            Err(mtp::RequestError::RpcError(mut error)) => {
                                debug!("got rpc error {:?} for request {:?}", error, msg_id);
                                let x = req.body.as_slice();
                                error.caused_by =
                                    Some(u32::from_le_bytes([x[0], x[1], x[2], x[3]]));
                                Err(InvocationError::Rpc(error))
                            }
                            Err(mtp::RequestError::Dropped) => {
                                debug!("response for request {:?} dropped", msg_id);
                                Err(InvocationError::Dropped)
                            }
                            Err(mtp::RequestError::Deserialize(error)) => {
                                debug!(
                                    "got deserialize error {:?} for request {:?}",
                                    error, msg_id
                                );
                                Err(InvocationError::Read(error.into()))
                            }
                            Err(err @ mtp::RequestError::BadMessage { .. }) => {
                                // TODO add a test to make sure we resend the request
                                info!("{}; re-sending request {:?}", err, msg_id);
                                req.state = RequestState::NotSerialized;
                                break;
                            }
                        };
                        let req = self.requests.remove(i);
                        drop(req.result.send(result));
                        break;
                    }
                    _ => {}
                }
            }
            if !found {
                info!("got rpc result {:?} but no such request is saved", msg_id);
            }
        }
        Ok(())
    }
fn process_mtp_buffer ( & mut self , updates : & mut Vec < tl :: enums :: Updates > , ) -> Result < ( ) , mtp :: DeserializeError > { debug ! ( " " ) ; let result = self . mtp . deserialize ( & self . mtp_buffer ) ? ; updates . extend ( result . updates . iter ( ) . filter_map ( | update | { match tl :: enums :: Updates :: from_bytes ( & update ) { Ok ( u ) => Some ( u ) , Err ( e ) => { warn ! ( " " e ) ; None } } } ) ) ; for ( msg_id , ret ) in result . rpc_results { let mut found = false ; for i in ( 0 .. self . requests . len ( ) ) . rev ( ) { let req = & mut self . requests [ i ] ; match req . state { RequestState :: Serialized ( sid ) if sid == msg_id => { panic ! ( " " msg_id sid ) ; } RequestState :: Sent ( sid ) if sid == msg_id => { found = true ; let result = match ret { Ok ( x ) => { assert ! ( x len ( ) 4 ) ; let res_id = u32 :: from_le_bytes ( [ x [ 0 ] , x [ 1 ] , x [ 2 ] , x [ 3 ] ] ) ; debug ! ( " " res_id tl name_for_id ( res_id ) msg_id ) ; Ok ( x ) } Err ( mtp :: RequestError :: RpcError ( mut error ) ) => { debug ! ( " " error msg_id ) ; let x = req . body . as_slice ( ) ; error . caused_by = Some ( u32 :: from_le_bytes ( [ x [ 0 ] , x [ 1 ] , x [ 2 ] , x [ 3 ] ] ) ) ; Err ( InvocationError :: Rpc ( error ) ) } Err ( mtp :: RequestError :: Dropped ) => { debug ! ( " " msg_id ) ; Err ( InvocationError :: Dropped ) } Err ( mtp :: RequestError :: Deserialize ( error ) ) => { debug ! ( " " error msg_id ) ; Err ( InvocationError :: Read ( error . into ( ) ) ) } Err ( err @ mtp :: RequestError :: BadMessage { .. } ) => { // TODO add a test to make sure we resend the request info ! ( " " err msg_id ) ; req . state = RequestState :: NotSerialized ; break ; } } ; let req = self . requests . remove ( i ) ; drop ( req . result . send ( result ) ) ; break ; } _ => { } } } if ! found { info ! ( " " msg_id ) ; } } Ok ( ( ) ) }
Process the `mtp_buffer` contents and dispatch the results and errors.
Process the ` mtp_buffer ` contents and dispatch the results and errors .
&mut self
updates
&mut Vec<tl::enums::Updates>
&mut self
updates
&mut Vec<tl::enums::Updates>