Struct rdkafka::consumer::base_consumer::BaseConsumer

pub struct BaseConsumer<C: ConsumerContext = DefaultConsumerContext> { /* fields omitted */ }

Low level wrapper around the librdkafka consumer. This consumer requires to be periodically polled to make progress on rebalance, callbacks and to receive messages.

Methods

impl<C: ConsumerContext> BaseConsumer<C>

pub fn context(&self) -> &C

Returns the context used to create this consumer.

pub fn poll<T: Into<Timeout>>(
    &self,
    timeout: T
) -> Option<KafkaResult<BorrowedMessage>>

Polls the consumer for new messages.

It won't block for more than the specified timeout. Use zero Duration for non-blocking call. With no timeout it blocks until an event is received.

This method should be called at regular intervals, even if no message is expected, to serve any queued callbacks waiting to be called. This is especially important for automatic consumer rebalance, as the rebalance function will be executed by the thread calling the poll() function.

Lifetime

The returned message lives in the memory of the consumer and cannot outlive it.

Important traits for Iter<'a, C>

impl<'a, C: ConsumerContext + 'a> Iterator for Iter<'a, C> type Item = KafkaResult<BorrowedMessage<'a>>;

pub fn iter(&self) -> Iter<C>

Returns an iterator over the available messages.

It repeatedly calls poll with no timeout.

Note that it's also possible to iterate over the consumer directly.

Examples

All these are equivalent and will receive messages without timing out.

loop {
  let message = consumer.poll(None);
  // Handle the message
}

for message in consumer.iter() {
  // Handle the message
}

for message in &consumer {
  // Handle the message
}

Trait Implementations

impl<C: ConsumerContext> Consumer<C> for BaseConsumer<C>

fn get_base_consumer(&self) -> &BaseConsumer<C>

Returns a reference to the BaseConsumer.

fn subscribe(&self, topics: &[&str]) -> KafkaResult<()>

Subscribe the consumer to a list of topics.

fn unsubscribe(&self)

Unsubscribe the current subscription list.

fn assign(&self, assignment: &TopicPartitionList) -> KafkaResult<()>

Manually assign topics and partitions to the consumer. If used, automatic consumer rebalance won't be activated.

fn seek<T: Into<Timeout>>(
    &self,
    topic: &str,
    partition: i32,
    offset: Offset,
    timeout: T
) -> KafkaResult<()>

Seek to offset for the specified topic and partition. After a successful call to seek, the next poll of the consumer will return the message with offset.

fn commit(
    &self,
    topic_partition_list: &TopicPartitionList,
    mode: CommitMode
) -> KafkaResult<()>

Commits the offset of the specified message. The commit can be sync (blocking), or async. Notice that when a specific offset is committed, all the previous offsets are considered committed as well. Use this method only if you are processing messages in order.

fn commit_consumer_state(&self, mode: CommitMode) -> KafkaResult<()>

Commit the current consumer state. Notice that if the consumer fails after a message has been received, but before the message has been processed by the user code, this might lead to data loss. Check the "at-least-once delivery" section in the readme for more information.

fn commit_message(
    &self,
    message: &BorrowedMessage,
    mode: CommitMode
) -> KafkaResult<()>

Commit the provided message. Note that this will also automatically commit every message with lower offset within the same partition.

fn store_offset(&self, message: &BorrowedMessage) -> KafkaResult<()>

Store offset for this message to be used on the next (auto)commit. When using this enable.auto.offset.store should be set to false in the config.

fn store_offsets(&self, tpl: &TopicPartitionList) -> KafkaResult<()>

Store offsets to be used on the next (auto)commit. When using this enable.auto.offset.store should be set to false in the config.

fn subscription(&self) -> KafkaResult<TopicPartitionList>

Returns the current topic subscription.

fn assignment(&self) -> KafkaResult<TopicPartitionList>

Returns the current partition assignment.

fn committed<T: Into<Timeout>>(
    &self,
    timeout: T
) -> KafkaResult<TopicPartitionList>

Retrieve committed offsets for topics and partitions.

fn committed_offsets<T: Into<Timeout>>(
    &self,
    tpl: TopicPartitionList,
    timeout: T
) -> KafkaResult<TopicPartitionList>

Retrieve committed offsets for specified topics and partitions.

fn offsets_for_timestamp<T: Into<Timeout>>(
    &self,
    timestamp: i64,
    timeout: T
) -> KafkaResult<TopicPartitionList>

Lookup the offsets for this consumer's partitions by timestamp.

fn offsets_for_times<T: Into<Timeout>>(
    &self,
    timestamps: TopicPartitionList,
    timeout: T
) -> KafkaResult<TopicPartitionList>

timestamps is a TopicPartitionList with timestamps instead of offsets.

fn position(&self) -> KafkaResult<TopicPartitionList>

Retrieve current positions (offsets) for topics and partitions.

fn fetch_metadata<T: Into<Timeout>>(
    &self,
    topic: Option<&str>,
    timeout: T
) -> KafkaResult<Metadata>

Returns the metadata information for the specified topic, or for all topics in the cluster if no topic is specified.

fn fetch_watermarks<T: Into<Timeout>>(
    &self,
    topic: &str,
    partition: i32,
    timeout: T
) -> KafkaResult<(i64, i64)>

Returns the metadata information for all the topics in the cluster.

fn fetch_group_list<T: Into<Timeout>>(
    &self,
    group: Option<&str>,
    timeout: T
) -> KafkaResult<GroupList>

Returns the group membership information for the given group. If no group is specified, all groups will be returned.

fn pause(&self, partitions: &TopicPartitionList) -> KafkaResult<()>

Pause consumption for the provided list of partitions.

fn resume(&self, partitions: &TopicPartitionList) -> KafkaResult<()>

Resume consumption for the provided list of partitions.

impl<C: ConsumerContext> Drop for BaseConsumer<C>

fn drop(&mut self)

Executes the destructor for this type.

impl FromClientConfig for BaseConsumer

fn from_config(config: &ClientConfig) -> KafkaResult<BaseConsumer>

Create a client from client configuration. The default client context will be used.

impl<C: ConsumerContext> FromClientConfigAndContext<C> for BaseConsumer<C>

Creates a new BaseConsumer starting from a ClientConfig.

fn from_config_and_context(
    config: &ClientConfig,
    context: C
) -> KafkaResult<BaseConsumer<C>>

Create a client from client configuration and a client context.

impl<'a, C: ConsumerContext + 'a> IntoIterator for &'a BaseConsumer<C>

type Item = KafkaResult<BorrowedMessage<'a>>

The type of the elements being iterated over.

type IntoIter = Iter<'a, C>

Which kind of iterator are we turning this into?

fn into_iter(self) -> Self::IntoIter

Creates an iterator from a value.