/// Insert key value pairs without serialization.
/// The function returns the newer version number post the insert operation.
Insert key value pairs without serialization.
The function returns the newer version number post the insert operation.
Insert key value pairs without serialization . The function returns the newer version number post the insert operation .
async fn insert_raw_values(
        &self,
        kvps: Vec<(Vec<u8>, Vec<u8>, Version)>,
        offset: i64,
    ) -> Result<Vec<Version>, TableError> {
        let op = "Insert into tablemap";
        retry_async(self.factory.config().retry_policy, || async {
            let entries: Vec<(TableKey, TableValue)> = kvps
                .iter()
                .map(|(k, v, ver)| {
                    let tk = TableKey::new(k.clone(), *ver);
                    let tv = TableValue::new(v.clone());
                    (tk, tv)
                })
                .collect();
            let te = TableEntries { entries };
            let req = Requests::UpdateTableEntries(UpdateTableEntriesCommand {
                request_id: get_request_id(),
                segment: self.name.clone(),
                delegation_token: self
                    .delegation_token_provider
                    .retrieve_token(self.factory.controller_client())
                    .await,
                table_entries: te,
                table_segment_offset: offset,
            });
            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");
                    }
                    info!("Table insert retry error {:?}", e);
                    RetryResult::Retry(e)
                }
            }
        })
        .await
        .map_err(|e| TableError::ConnectionError {
            can_retry: true,
            operation: op.into(),
            source: e.error,
        })
        .and_then(|r| match r {
            Replies::TableEntriesUpdated(c) => Ok(c.updated_versions),
            Replies::TableKeyBadVersion(c) => Err(TableError::IncorrectKeyVersion {
                operation: op.into(),
                error_msg: c.to_string(),
            }),
            // unexpected response from Segment store causes a panic.
            _ => Err(TableError::OperationError {
                operation: op.into(),
                error_msg: r.to_string(),
            }),
        })
    }
async fn insert_raw_values ( & self , kvps : Vec < ( Vec < u8 > , Vec < u8 > , Version ) > , offset : i64 , ) -> Result < Vec < Version > , TableError > { let op = " " ; retry_async ( self . factory . config ( ) . retry_policy , | | async { let entries : Vec < ( TableKey , TableValue ) > = kvps . iter ( ) . map ( | ( k , v , ver ) | { let tk = TableKey :: new ( k . clone ( ) , * ver ) ; let tv = TableValue :: new ( v . clone ( ) ) ; ( tk , tv ) } ) . collect ( ) ; let te = TableEntries { entries } ; let req = Requests :: UpdateTableEntries ( UpdateTableEntriesCommand { request_id : get_request_id ( ) , segment : self . name . clone ( ) , delegation_token : self . delegation_token_provider . retrieve_token ( self . factory . controller_client ( ) ) . await , table_entries : te , table_segment_offset : offset , } ) ; 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 ! ( " " ) ; } info ! ( " " e ) ; RetryResult :: Retry ( e ) } } } ) . await . map_err ( | e | TableError :: ConnectionError { can_retry : true , operation : op . into ( ) , source : e . error , } ) . and_then ( | r | match r { Replies :: TableEntriesUpdated ( c ) => Ok ( c . updated_versions ) , Replies :: TableKeyBadVersion ( c ) => Err ( TableError :: IncorrectKeyVersion { operation : op . into ( ) , error_msg : c . to_string ( ) , } ) , // unexpected response from Segment store causes a panic. _ => Err ( TableError :: OperationError { operation : op . into ( ) , error_msg : r . to_string ( ) , } ) , } ) }
Insert key value pairs without serialization.
Insert key value pairs without serialization .
&self
kvps
Vec<(Vec<u8>, Vec<u8>, Version)>
offset
i64