Struct StreamingConsumer 

pub struct StreamingConsumer { /* private fields */ }

A streaming consumer that uses subscribe/unsubscribe signals

Unlike the poll-based Consumer, StreamingConsumer explicitly signals to the server when it starts and stops consuming, and reports position updates for consumer group tracking.

Lifecycle

1. Subscribe - Call start() to signal the server to begin streaming
2. Consume - Call poll() to receive batches of data
3. Commit - Call commit() to checkpoint progress (or use auto-commit)
4. Unsubscribe - Call stop() or drop the consumer to signal completion

Example

let client = LanceClient::connect_to("127.0.0.1:1992").await?;
let config = StreamingConsumerConfig::new(topic_id)
    .with_consumer_group("my-group");
let mut consumer = StreamingConsumer::new(client, config);

// Start streaming
consumer.start().await?;

// Process records
while let Some(result) = consumer.poll().await? {
    process_data(&result.data);
    consumer.commit().await?; // Checkpoint progress
}

// Stop streaming
consumer.stop().await?;

Implementations

impl StreamingConsumer

pub fn new(client: LanceClient, config: StreamingConsumerConfig) -> Self

Create a new streaming consumer

pub async fn start(&mut self) -> Result<(), ClientError>

Start streaming from the topic

Sends a Subscribe signal to the server indicating this consumer wants to start receiving data from the configured position.

pub async fn stop(&mut self) -> Result<(), ClientError>

Stop streaming from the topic

Sends an Unsubscribe signal to the server. This should be called when the consumer is done processing to free server resources.

pub async fn poll(&mut self) -> Result<Option<PollResult>, ClientError>

Poll for the next batch of records

Returns Ok(Some(result)) if data was fetched, or Ok(None) if no new data is available.

pub async fn commit(&mut self) -> Result<(), ClientError>

Commit the current offset to the server

This checkpoints the consumer’s position so that if it restarts, it can resume from this point.

pub async fn seek(&mut self, position: SeekPosition) -> Result<u64, ClientError>

Seek to a new position

Note: This will commit the current offset and restart the subscription at the new position.

pub fn current_offset(&self) -> u64

Get the current offset

pub fn committed_offset(&self) -> u64

Get the last committed offset

pub fn consumer_id(&self) -> u64

Get the consumer ID

pub fn is_subscribed(&self) -> bool

Check if currently subscribed

pub fn client(&self) -> &LanceClient

Get access to the underlying client

pub async fn into_client(self) -> Result<LanceClient, ClientError>

Consume this consumer and return the underlying client

Trait Implementations

impl Debug for StreamingConsumer

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.