/// Consume records as a stream, waiting for new records to arrive
Consume records as a stream, waiting for new records to arrive
Consume records as a stream waiting for new records to arrive
async fn consume_records_stream(
        &self,
        consumer: &PartitionConsumer,
        offset: Offset,
        config: ConsumerConfig,
    ) -> Result<(), ConsumerError> {
        let mut stream = consumer.stream_with_config(offset, config).await?;
        while let Some(result) = stream.next().await {
            let result: std::result::Result<Record, _> = result;
            let record = match result {
                Ok(record) => record,
                Err(FluvioError::ApiError(ApiError::Code(code, _))) => {
                    eprintln!("{}", code.to_sentence());
                    continue;
                }
                Err(other) => return Err(other.into()),
            };
            self.print_record(record.key(), record.value());
        }
        debug!("fetch loop exited");
        eprintln!("Consumer stream has closed");
        Ok(())
    }
async fn consume_records_stream ( & self , consumer : & PartitionConsumer , offset : Offset , config : ConsumerConfig , ) -> Result < ( ) , ConsumerError > { let mut stream = consumer . stream_with_config ( offset , config ) . await ? ; while let Some ( result ) = stream . next ( ) . await { let result : std :: result :: Result < Record , _ > = result ; let record = match result { Ok ( record ) => record , Err ( FluvioError :: ApiError ( ApiError :: Code ( code , _ ) ) ) => { eprintln ! ( " " code to_sentence ( ) ) ; continue ; } Err ( other ) => return Err ( other . into ( ) ) , } ; self . print_record ( record . key ( ) , record . value ( ) ) ; } debug ! ( " " ) ; eprintln ! ( " " ) ; Ok ( ( ) ) }
Consume records as a stream, waiting for new records to arrive
Consume records as a stream waiting for new records to arrive
&self
consumer
&PartitionConsumer
offset
Offset
config
ConsumerConfig