Struct DefaultLitePullConsumer 

pub struct DefaultLitePullConsumer { /* private fields */ }

Default implementation of a lite pull consumer.

This is the main entry point for creating and using a lite pull consumer. It acts as a facade over the internal implementation ([DefaultLitePullConsumerImpl]) and provides:

• Configuration management via DefaultLitePullConsumerBuilder
• Namespace handling (automatic topic name wrapping/unwrapping)
• Optional message trace integration
• A clean public API that delegates to the internal implementation

Architecture

┌────────────────────────────────────┐
│   DefaultLitePullConsumer          │  ← Public Facade
│  (config, trace, namespace)        │
└──────────────┬─────────────────────┘
               │ delegates
               ↓
┌────────────────────────────────────┐
│ DefaultLitePullConsumerImpl        │  ← Core Logic
│  (lifecycle, pull, commit, etc.)   │
└────────────────────────────────────┘

Examples

Basic usage with auto-commit

use rocketmq_client::consumer::default_lite_pull_consumer::DefaultLitePullConsumer;

let consumer = DefaultLitePullConsumer::builder()
    .consumer_group("my_consumer_group")
    .name_server_addr("127.0.0.1:9876")
    .auto_commit(true)
    .build();

consumer.start().await?;
consumer.subscribe("my_topic").await?;

loop {
    let messages = consumer.poll_with_timeout(1000).await;
    for msg in messages {
        println!("Received: {:?}", msg);
    }
    // Offsets are auto-committed
}

Manual offset control

let consumer = DefaultLitePullConsumer::builder()
    .consumer_group("my_consumer_group")
    .name_server_addr("127.0.0.1:9876")
    .auto_commit(false)  // Disable auto-commit
    .build();

consumer.start().await?;
consumer.subscribe("my_topic").await?;

loop {
    let messages = consumer.poll_with_timeout(1000).await;
    for msg in messages {
        // Process message
    }
     
    // Manually commit after processing
    consumer.commit_all().await?;
}

Manual queue assignment (no rebalance)

use rocketmq_common::common::message::message_queue::MessageQueue;

let mq1 = MessageQueue::from_parts("my_topic", "broker-a", 0);
let mq2 = MessageQueue::from_parts("my_topic", "broker-a", 1);

consumer.start().await?;
consumer.assign(vec![mq1, mq2]).await;

let messages = consumer.poll().await;

Implementations

impl DefaultLitePullConsumer

pub fn new( client_config: ArcMut<ClientConfig>, consumer_config: ArcMut<LitePullConsumerConfig>, rpc_hook: Option<Arc<dyn RPCHook>>, trace_dispatcher: Option<Arc<dyn TraceDispatcher + Send + Sync>>, enable_msg_trace: bool, custom_trace_topic: Option<CheetahString>, ) -> Self

Creates a new consumer with the specified configuration.

Most users should use builder() instead.

pub fn builder() -> DefaultLitePullConsumerBuilder

Returns a builder for creating a consumer with custom configuration.

Examples

let consumer = DefaultLitePullConsumer::builder()
    .consumer_group("my_group")
    .name_server_addr("127.0.0.1:9876")
    .pull_batch_size(32)
    .build();

pub fn consumer_group(&self) -> &CheetahString

Returns the consumer group name.

pub fn namespace(&self) -> Option<CheetahString>

Returns the namespace (if configured).

pub fn client_config(&self) -> &ArcMut<ClientConfig>

Returns the client configuration.

pub fn consumer_config(&self) -> &ArcMut<LitePullConsumerConfig>

Returns the consumer configuration.

Trait Implementations

impl Clone for DefaultLitePullConsumer

fn clone(&self) -> DefaultLitePullConsumer

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl LitePullConsumer for DefaultLitePullConsumer

async fn poll_zero_copy(&self) -> Vec<ArcMut<MessageExt>>

Zero-copy implementation.

async fn poll_with_timeout_zero_copy( &self, timeout: u64, ) -> Vec<ArcMut<MessageExt>>

Zero-copy implementation with custom timeout.

async fn poll(&self) -> Vec<MessageExt>

Delegates to zero-copy implementation and creates owned message copies.

async fn poll_with_timeout(&self, timeout: u64) -> Vec<MessageExt>

Delegates to zero-copy implementation with timeout and creates owned message copies.

async fn start(&self) -> RocketMQResult<()>

Starts the consumer and establishes connections to the broker and name server.

async fn shutdown(&self)

Shuts down the consumer and releases all associated resources.

async fn is_running(&self) -> bool

Returns whether the consumer is currently in the running state.

async fn subscribe(&self, topic: &str) -> RocketMQResult<()>

Subscribes to the specified topic using the default subscription expression.

async fn subscribe_with_expression( &self, topic: &str, sub_expression: &str, ) -> RocketMQResult<()>

Subscribes to the specified topic with a tag-based or SQL-based filter expression.

async fn subscribe_with_listener<MQL>( &self, topic: &str, sub_expression: &str, listener: MQL, ) -> RocketMQResult<()>

where MQL: MessageQueueListener + 'static,

Subscribes to the specified topic with a filter expression and a queue-change listener.

async fn subscribe_with_selector( &self, topic: &str, selector: Option<MessageSelector>, ) -> RocketMQResult<()>

Subscribes to the specified topic using a MessageSelector for server-side filtering.

async fn unsubscribe(&self, topic: &str)

Removes the subscription for the specified topic.

async fn assignment(&self) -> RocketMQResult<HashSet<MessageQueue>>

Returns the set of MessageQueues currently assigned to this consumer.

async fn assign(&self, message_queues: Vec<MessageQueue>)

Manually assigns the given MessageQueues to this consumer, bypassing broker rebalance.

async fn set_sub_expression_for_assign(&self, topic: &str, sub_expression: &str)

Sets the subscription filter expression applied when fetching from manually assigned queues.

async fn build_subscriptions_for_heartbeat( &self, sub_expression_map: &mut HashMap<String, MessageSelector>, ) -> RocketMQResult<()>

Populates sub_expression_map with the filter selector for each subscribed topic, providing the subscription metadata required for heartbeat payloads.

async fn fetch_message_queues( &self, topic: &str, ) -> RocketMQResult<Vec<MessageQueue>>

Queries the broker for all MessageQueues belonging to the specified topic.

async fn committed(&self, message_queue: &MessageQueue) -> RocketMQResult<i64>

Returns the last committed offset for the specified MessageQueue.

async fn commit_all(&self) -> RocketMQResult<()>

Commits all consumed offsets for all assigned queues.

async fn commit_with_map( &self, offset_map: HashMap<MessageQueue, i64>, persist: bool, )

Commits the provided offsets asynchronously, optionally persisting them to the broker.

async fn commit(&self)

Commits all consumed offsets asynchronously.

async fn commit_with_set( &self, message_queues: HashSet<MessageQueue>, persist: bool, )

Commits the offsets for the specified subset of assigned queues.

async fn commit_sync(&self)

Commits all consumed offsets and waits for the broker to acknowledge the operation.

async fn commit_sync_with_map( &self, offset_map: HashMap<MessageQueue, i64>, persist: bool, )

Commits the provided offsets and optionally persists them to the broker.

async fn is_auto_commit(&self) -> bool

Returns whether automatic offset commit is enabled.

async fn set_auto_commit(&self, auto_commit: bool)

Enables or disables automatic offset commit.

async fn offset_for_timestamp( &self, message_queue: &MessageQueue, timestamp: u64, ) -> RocketMQResult<i64>

Queries the broker for the offset corresponding to the given timestamp in a queue.

async fn pause(&self, message_queues: Vec<MessageQueue>)

Suspends message fetching for the specified MessageQueues.

async fn resume(&self, message_queues: Vec<MessageQueue>)

Resumes message fetching for the specified MessageQueues.

async fn is_paused(&self, message_queue: &MessageQueue) -> bool

Checks whether a specific MessageQueue is currently paused.

async fn seek( &self, message_queue: &MessageQueue, offset: i64, ) -> RocketMQResult<()>

Seeks the fetch position of the specified MessageQueue to the given offset.

async fn seek_to_begin( &self, message_queue: &MessageQueue, ) -> RocketMQResult<()>

Seeks the fetch position of the specified MessageQueue to its earliest available offset.

async fn seek_to_end(&self, message_queue: &MessageQueue) -> RocketMQResult<()>

Seeks the fetch position of the specified MessageQueue to its latest available offset.

async fn update_name_server_address(&self, name_server_address: &str)

Updates the name server address used for topic route discovery.

async fn register_topic_message_queue_change_listener<TL>( &self, topic: &str, listener: TL, ) -> RocketMQResult<()>

where TL: TopicMessageQueueChangeListener + 'static,

Registers a listener that is notified when the set of MessageQueues for a topic changes.

impl MQConsumer for DefaultLitePullConsumer

async fn send_message_back( &mut self, msg: MessageExt, delay_level: i32, broker_name: &str, ) -> RocketMQResult<()>

async fn fetch_subscribe_message_queues( &mut self, topic: &str, ) -> RocketMQResult<Vec<MessageQueue>>