pub(crate) async fn base_offset(&self) -> Result<Offset> {
        let mut state = self.state.write().await;
        match &*state {
            BatchMetadataState::Buffered(receiver) => {
                let offset_result = receiver
                    .recv()
                    .await
                    .map_err(|err| ProducerError::GetRecordMetadata(Some(err)));

                match offset_result {
                    Ok(offset) => {
                        if offset.1 == ErrorCode::None {
                            *state = BatchMetadataState::Sended(offset.0);
                            Ok(offset.0)
                        } else {
                            let error = ProducerError::SpuErrorCode(offset.1);
                            *state = BatchMetadataState::Failed(error.clone());
                            Err(error.into())
                        }
                    }
                    Err(err) => {
                        *state = BatchMetadataState::Failed(err.clone());
                        Err(err.into())
                    }
                }
            }
            BatchMetadataState::Sended(offset) => Ok(*offset),
            BatchMetadataState::Failed(error) => Err(error.clone().into()),
        }
    }
/// Wait for the base offset of the batch. This is the offset of the first
/// record in the batch and it is known once the batch is sent to the server.
Wait for the base offset of the batch. This is the offset of the first
record in the batch and it is known once the batch is sent to the server.
Wait for the base offset of the batch . This is the offset of the first record in the batch and it is known once the batch is sent to the server .
pub(crate) async fn base_offset(&self) -> Result<Offset> {
        let mut state = self.state.write().await;
        match &*state {
            BatchMetadataState::Buffered(receiver) => {
                let offset_result = receiver
                    .recv()
                    .await
                    .map_err(|err| ProducerError::GetRecordMetadata(Some(err)));
                match offset_result {
                    Ok(offset) => {
                        if offset.1 == ErrorCode::None {
                            *state = BatchMetadataState::Sended(offset.0);
                            Ok(offset.0)
                        } else {
                            let error = ProducerError::SpuErrorCode(offset.1);
                            *state = BatchMetadataState::Failed(error.clone());
                            Err(error.into())
                        }
                    }
                    Err(err) => {
                        *state = BatchMetadataState::Failed(err.clone());
                        Err(err.into())
                    }
                }
            }
            BatchMetadataState::Sended(offset) => Ok(*offset),
            BatchMetadataState::Failed(error) => Err(error.clone().into()),
        }
    }
pub ( crate ) async fn base_offset ( & self ) -> Result < Offset > { let mut state = self . state . write ( ) . await ; match & * state { BatchMetadataState :: Buffered ( receiver ) => { let offset_result = receiver . recv ( ) . await . map_err ( | err | ProducerError :: GetRecordMetadata ( Some ( err ) ) ) ; match offset_result { Ok ( offset ) => { if offset . 1 == ErrorCode :: None { * state = BatchMetadataState :: Sended ( offset . 0 ) ; Ok ( offset . 0 ) } else { let error = ProducerError :: SpuErrorCode ( offset . 1 ) ; * state = BatchMetadataState :: Failed ( error . clone ( ) ) ; Err ( error . into ( ) ) } } Err ( err ) => { * state = BatchMetadataState :: Failed ( err . clone ( ) ) ; Err ( err . into ( ) ) } } } BatchMetadataState :: Sended ( offset ) => Ok ( * offset ) , BatchMetadataState :: Failed ( error ) => Err ( error . clone ( ) . into ( ) ) , } }
Wait for the base offset of the batch.
Wait for the base offset of the batch .
&self
&self