Trait MessageListenerConcurrently 

pub trait MessageListenerConcurrently: Send + Sync {
    // Required method
    fn consume_message(
        &self,
        msgs: &[&MessageExt],
        context: &ConsumeConcurrentlyContext,
    ) -> ClientResult<ConsumeConcurrentlyStatus>;
}

A trait for processing messages concurrently.

This trait provides a zero-cost abstraction for concurrent message consumption. Implement this trait for your custom message processing logic, or use closures directly as they automatically implement this trait.

Type Parameters

Note: This trait does not use generic type parameters. The message type is fixed to MessageExt and the context to ConsumeConcurrentlyContext for compatibility with RocketMQ protocol.

Performance

When using closures or custom implementations directly (without type erasure), this trait provides zero-cost abstraction through:

• Compile-time monomorphization
• Full function inlining
• No dynamic dispatch overhead

Example

use rocketmq_client_rust::ConsumeConcurrentlyContext;
use rocketmq_client_rust::ConsumeConcurrentlyStatus;
use rocketmq_client_rust::ClientResult;
use rocketmq_client_rust::MessageListenerConcurrently;
use rocketmq_model::common::message::message_ext::MessageExt;

// Closures automatically implement MessageListenerConcurrently
let listener = |msgs: &[&MessageExt],
                _context: &ConsumeConcurrentlyContext|
 -> ClientResult<ConsumeConcurrentlyStatus> {
    for msg in msgs {
        println!("Processing message: {:?}", msg.msg_id());
    }
    Ok(ConsumeConcurrentlyStatus::ConsumeSuccess)
};

// Struct-based implementation
struct MyListener;

impl MessageListenerConcurrently for MyListener {
    fn consume_message(
        &self,
        msgs: &[&MessageExt],
        context: &ConsumeConcurrentlyContext,
    ) -> ClientResult<ConsumeConcurrentlyStatus> {
        // Process messages
        Ok(ConsumeConcurrentlyStatus::ConsumeSuccess)
    }
}

Required Methods

fn consume_message( &self, msgs: &[&MessageExt], context: &ConsumeConcurrentlyContext, ) -> ClientResult<ConsumeConcurrentlyStatus>

Processes a batch of messages concurrently.

Arguments

• msgs - Slice of message references. Zero-cost borrowing without cloning.
• context - Context for concurrent consumption containing metadata like message queue, delay level, etc.

Returns

• Ok(ConsumeConcurrentlyStatus::ConsumeSuccess) - Messages processed successfully
• Ok(ConsumeConcurrentlyStatus::ReconsumeLater) - Messages should be retried later
• Err(error) - Processing failed with an error

Notes

• This method may be called concurrently for different message batches
• Messages from the same queue are never processed concurrently
• The default batch size is 1, configurable via consumer settings

Dyn Compatibility

This trait is dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors

impl<F> MessageListenerConcurrently for F

where F: Fn(&[&MessageExt], &ConsumeConcurrentlyContext) -> ClientResult<ConsumeConcurrentlyStatus> + Send + Sync,

Implement MessageListenerConcurrently for all compatible closures and functions.

This blanket implementation allows closures to be used directly as message listeners without explicit trait implementation, providing a convenient and zero-cost API.