Struct DefaultMQPushConsumer

pub struct DefaultMQPushConsumer { /* private fields */ }

Implementations

impl DefaultMQPushConsumer

pub fn builder() -> DefaultMQPushConsumerBuilder

Examples found in repository

examples/quickstart/consumer.rs (line 39)

pub async fn main() -> RocketMQResult<()> {
    //init logger
    rocketmq_common::log::init_logger();

    // create a producer builder with default configuration
    let builder = DefaultMQPushConsumer::builder();

    let mut consumer = builder
        .consumer_group(CONSUMER_GROUP.to_string())
        .name_server_addr(DEFAULT_NAMESRVADDR.to_string())
        .build();
    consumer.subscribe(TOPIC, "*")?;
    consumer.register_message_listener_concurrently(MyMessageListener);
    consumer.start().await?;
    let _ = tokio::signal::ctrl_c().await;
    Ok(())
}

examples/consumer/pop_consumer.rs (line 39)

pub async fn main() -> RocketMQResult<()> {
    //init logger
    rocketmq_common::log::init_logger();

    // create a producer builder with default configuration
    let builder = DefaultMQPushConsumer::builder();

    let mut consumer = builder
        .consumer_group(CONSUMER_GROUP.to_string())
        .name_server_addr(DEFAULT_NAMESRVADDR.to_string())
        // disable client side load balance, also is pop consumer
        .client_rebalance(false)
        .build();
    consumer.subscribe(TOPIC, "*")?;
    consumer.register_message_listener_concurrently(MyMessageListener);
    consumer.start().await?;
    let _ = tokio::signal::ctrl_c().await;
    Ok(())
}

examples/ordermessage/ordermessage_consumer.rs (line 46)

pub async fn main() -> RocketMQResult<()> {
    //init logger
    rocketmq_common::log::init_logger();

    // create a producer builder with default configuration
    let builder = DefaultMQPushConsumer::builder();

    let mut consumer = builder
        .consumer_group(CONSUMER_GROUP.to_string())
        .name_server_addr(DEFAULT_NAMESRVADDR.to_string())
        .message_model(MessageModel::Clustering)
        .build();
    consumer.subscribe(TOPIC, TAG)?;
    consumer.set_consume_from_where(ConsumeFromWhere::ConsumeFromFirstOffset);
    consumer.register_message_listener_orderly(MyMessageListener::new());
    consumer.start().await?;
    let _ = tokio::signal::ctrl_c().await;
    Ok(())
}

examples/broadcast/push_consumer.rs (line 42)

pub async fn main() -> RocketMQResult<()> {
    //init logger
    rocketmq_common::log::init_logger();

    // create a producer builder with default configuration
    let builder = DefaultMQPushConsumer::builder();

    let mut consumer = builder
        .consumer_group(CONSUMER_GROUP.to_string())
        .name_server_addr(DEFAULT_NAMESRVADDR.to_string())
        .message_model(MessageModel::Broadcasting)
        .build();
    consumer.subscribe(TOPIC, SUB_EXPRESSION)?;
    consumer.set_consume_from_where(ConsumeFromWhere::ConsumeFromFirstOffset);
    consumer.register_message_listener_concurrently(MyMessageListener);
    consumer.start().await?;
    let _ = tokio::signal::ctrl_c().await;
    Ok(())
}

pub fn set_consumer_group(&mut self, consumer_group: impl Into<CheetahString>)

pub fn new( client_config: ClientConfig, consumer_config: ConsumerConfig, ) -> DefaultMQPushConsumer

pub fn set_name_server_addr(&mut self, name_server_addr: CheetahString)

pub fn set_consume_from_where(&mut self, consume_from_where: ConsumeFromWhere)

Examples found in repository

examples/ordermessage/ordermessage_consumer.rs (line 54)

pub async fn main() -> RocketMQResult<()> {
    //init logger
    rocketmq_common::log::init_logger();

    // create a producer builder with default configuration
    let builder = DefaultMQPushConsumer::builder();

    let mut consumer = builder
        .consumer_group(CONSUMER_GROUP.to_string())
        .name_server_addr(DEFAULT_NAMESRVADDR.to_string())
        .message_model(MessageModel::Clustering)
        .build();
    consumer.subscribe(TOPIC, TAG)?;
    consumer.set_consume_from_where(ConsumeFromWhere::ConsumeFromFirstOffset);
    consumer.register_message_listener_orderly(MyMessageListener::new());
    consumer.start().await?;
    let _ = tokio::signal::ctrl_c().await;
    Ok(())
}

examples/broadcast/push_consumer.rs (line 50)

pub async fn main() -> RocketMQResult<()> {
    //init logger
    rocketmq_common::log::init_logger();

    // create a producer builder with default configuration
    let builder = DefaultMQPushConsumer::builder();

    let mut consumer = builder
        .consumer_group(CONSUMER_GROUP.to_string())
        .name_server_addr(DEFAULT_NAMESRVADDR.to_string())
        .message_model(MessageModel::Broadcasting)
        .build();
    consumer.subscribe(TOPIC, SUB_EXPRESSION)?;
    consumer.set_consume_from_where(ConsumeFromWhere::ConsumeFromFirstOffset);
    consumer.register_message_listener_concurrently(MyMessageListener);
    consumer.start().await?;
    let _ = tokio::signal::ctrl_c().await;
    Ok(())
}

Trait Implementations

impl MQAdmin for DefaultMQPushConsumer

fn create_topic( &self, key: &str, new_topic: &str, queue_num: i32, attributes: HashMap<String, String>, ) -> RocketMQResult<()>

Creates a new topic.

fn create_topic_with_flag( &self, key: &str, new_topic: &str, queue_num: i32, topic_sys_flag: i32, attributes: HashMap<String, String>, ) -> RocketMQResult<()>

Creates a new topic with a system flag.

fn search_offset( &self, mq: &MessageQueue, timestamp: u64, ) -> RocketMQResult<i64>

Searches for the offset of a message in a queue at a given timestamp.

fn max_offset(&self, mq: &MessageQueue) -> RocketMQResult<i64>

Retrieves the maximum offset of a message in a queue.

fn min_offset(&self, mq: &MessageQueue) -> RocketMQResult<i64>

Retrieves the minimum offset of a message in a queue.

fn earliest_msg_store_time(&self, mq: &MessageQueue) -> RocketMQResult<u64>

Retrieves the earliest message store time in a queue.

fn query_message( &self, topic: &str, key: &str, max_num: i32, begin: u64, end: u64, ) -> RocketMQResult<QueryResult>

Queries messages in a topic by key within a time range.

fn view_message(&self, topic: &str, msg_id: &str) -> RocketMQResult<MessageExt>

Views a message by its ID in a topic.

impl MQConsumer for DefaultMQPushConsumer

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>>

impl MQPushConsumer for DefaultMQPushConsumer

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

Starts the push consumer.

async fn shutdown(&mut self)

Shuts down the push consumer.

fn register_message_listener_concurrently_fn<MLCFN>( &mut self, message_listener: MLCFN, )

where MLCFN: Fn(Vec<MessageExt>, ConsumeConcurrentlyContext) -> RocketMQResult<ConsumeConcurrentlyStatus> + Send + Sync,

Registers a message listener for concurrent message consumption.

fn register_message_listener_concurrently<ML>(&mut self, message_listener: ML)

where ML: MessageListenerConcurrently + Send + Sync + 'static,

async fn register_message_listener_orderly_fn<MLOFN>( &mut self, message_listener: MLOFN, )

where MLOFN: Fn(Vec<MessageExt>, ConsumeOrderlyContext) -> RocketMQResult<ConsumeOrderlyStatus> + Send + Sync,

Registers a message listener for orderly message consumption.

fn register_message_listener_orderly<ML>(&mut self, message_listener: ML)

where ML: MessageListenerOrderly + Send + Sync + 'static,

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

Subscribes to a topic with a subscription expression.

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

Subscribes to a topic with an optional message selector.

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

Unsubscribes from a topic.

async fn suspend(&mut self)

Suspends the push consumer.

async fn resume(&mut self)

Resumes the push consumer.