async fn read_entries_raw_delta(
        &self,
        max_entries_at_once: i32,
        from_position: i64,
    ) -> Result<(Vec<(Vec<u8>, Vec<u8>, Version)>, i64), TableError> {
        let op = "Read entries delta";

        retry_async(self.factory.config().retry_policy, || async {
            let req = Requests::ReadTableEntriesDelta(ReadTableEntriesDeltaCommand {
                request_id: get_request_id(),
                segment: self.name.clone(),
                delegation_token: self
                    .delegation_token_provider
                    .retrieve_token(self.factory.controller_client())
                    .await,
                from_position,
                suggested_entry_count: max_entries_at_once,
            });
            let result = self
                .factory
                .create_raw_client_for_endpoint(self.endpoint.clone())
                .send_request(&req)
                .await;

            match result {
                Ok(reply) => RetryResult::Success(reply),
                Err(e) => {
                    if e.is_token_expired() {
                        self.delegation_token_provider.signal_token_expiry();
                        info!("auth token needs to refresh");
                    }
                    RetryResult::Retry(e)
                }
            }
        })
        .await
        .map_err(|e| TableError::ConnectionError {
            can_retry: true,
            operation: op.into(),
            source: e.error,
        })
        .and_then(|r| {
            match r {
                Replies::TableEntriesDeltaRead(c) => {
                    let entries: Vec<(Vec<u8>, Vec<u8>, Version)> = c
                        .entries
                        .entries
                        .iter()
                        .map(|(k, v)| (k.data.clone(), v.data.clone(), k.key_version))
                        .collect();

                    Ok((entries, c.last_position))
                }
                // unexpected response from Segment store causes a panic.
                _ => Err(TableError::OperationError {
                    operation: op.into(),
                    error_msg: r.to_string(),
                }),
            }
        })
    }
/// Read the raw entries from the table map from a given position.
/// It returns a list of key-values and its versions with a latest position.
Read the raw entries from the table map from a given position.
It returns a list of key-values and its versions with a latest position.
Read the raw entries from the table map from a given position . It returns a list of key - values and its versions with a latest position .
async fn read_entries_raw_delta(
        &self,
        max_entries_at_once: i32,
        from_position: i64,
    ) -> Result<(Vec<(Vec<u8>, Vec<u8>, Version)>, i64), TableError> {
        let op = "Read entries delta";
        retry_async(self.factory.config().retry_policy, || async {
            let req = Requests::ReadTableEntriesDelta(ReadTableEntriesDeltaCommand {
                request_id: get_request_id(),
                segment: self.name.clone(),
                delegation_token: self
                    .delegation_token_provider
                    .retrieve_token(self.factory.controller_client())
                    .await,
                from_position,
                suggested_entry_count: max_entries_at_once,
            });
            let result = self
                .factory
                .create_raw_client_for_endpoint(self.endpoint.clone())
                .send_request(&req)
                .await;
            match result {
                Ok(reply) => RetryResult::Success(reply),
                Err(e) => {
                    if e.is_token_expired() {
                        self.delegation_token_provider.signal_token_expiry();
                        info!("auth token needs to refresh");
                    }
                    RetryResult::Retry(e)
                }
            }
        })
        .await
        .map_err(|e| TableError::ConnectionError {
            can_retry: true,
            operation: op.into(),
            source: e.error,
        })
        .and_then(|r| {
            match r {
                Replies::TableEntriesDeltaRead(c) => {
                    let entries: Vec<(Vec<u8>, Vec<u8>, Version)> = c
                        .entries
                        .entries
                        .iter()
                        .map(|(k, v)| (k.data.clone(), v.data.clone(), k.key_version))
                        .collect();
                    Ok((entries, c.last_position))
                }
                // unexpected response from Segment store causes a panic.
                _ => Err(TableError::OperationError {
                    operation: op.into(),
                    error_msg: r.to_string(),
                }),
            }
        })
    }
async fn read_entries_raw_delta ( & self , max_entries_at_once : i32 , from_position : i64 , ) -> Result < ( Vec < ( Vec < u8 > , Vec < u8 > , Version ) > , i64 ) , TableError > { let op = " " ; retry_async ( self . factory . config ( ) . retry_policy , | | async { let req = Requests :: ReadTableEntriesDelta ( ReadTableEntriesDeltaCommand { request_id : get_request_id ( ) , segment : self . name . clone ( ) , delegation_token : self . delegation_token_provider . retrieve_token ( self . factory . controller_client ( ) ) . await , from_position , suggested_entry_count : max_entries_at_once , } ) ; let result = self . factory . create_raw_client_for_endpoint ( self . endpoint . clone ( ) ) . send_request ( & req ) . await ; match result { Ok ( reply ) => RetryResult :: Success ( reply ) , Err ( e ) => { if e . is_token_expired ( ) { self . delegation_token_provider . signal_token_expiry ( ) ; info ! ( " " ) ; } RetryResult :: Retry ( e ) } } } ) . await . map_err ( | e | TableError :: ConnectionError { can_retry : true , operation : op . into ( ) , source : e . error , } ) . and_then ( | r | { match r { Replies :: TableEntriesDeltaRead ( c ) => { let entries : Vec < ( Vec < u8 > , Vec < u8 > , Version ) > = c . entries . entries . iter ( ) . map ( | ( k , v ) | ( k . data . clone ( ) , v . data . clone ( ) , k . key_version ) ) . collect ( ) ; Ok ( ( entries , c . last_position ) ) } // unexpected response from Segment store causes a panic. _ => Err ( TableError :: OperationError { operation : op . into ( ) , error_msg : r . to_string ( ) , } ) , } } ) }
Read the raw entries from the table map from a given position.
Read the raw entries from the table map from a given position .
&self
max_entries_at_once
i32
from_position
i64
&self
max_entries_at_once
i32
from_position
i64