Trait MessageListenerOrderly 

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

A trait for processing messages in order.

This trait ensures messages from the same queue are consumed sequentially. Implement this trait for your custom sequential message processing logic, or use closures directly as they automatically implement this trait.

Difference from MessageListenerConcurrently

• Orderly: Messages from the same queue are processed sequentially
• Concurrently: Messages from different queues can be processed in parallel

Context Mutability

The context parameter is mutable (&mut) to allow consumers to:

• Control auto-commit behavior via set_auto_commit()
• Set queue suspension time via set_suspend_current_queue_time_millis()

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::ConsumeOrderlyContext;
use rocketmq_client_rust::ConsumeOrderlyStatus;
use rocketmq_client_rust::ClientResult;
use rocketmq_client_rust::MessageListenerOrderly;
use rocketmq_model::common::message::message_ext::MessageExt;

// Closures automatically implement MessageListenerOrderly
let listener = |msgs: &[&MessageExt],
                context: &mut ConsumeOrderlyContext|
 -> ClientResult<ConsumeOrderlyStatus> {
    for msg in msgs {
        println!("Processing message in order: {:?}", msg.msg_id());
    }
    // Control commit behavior
    context.set_auto_commit(true);
    Ok(ConsumeOrderlyStatus::Success)
};

// Struct-based implementation
struct MyOrderlyListener;

impl MessageListenerOrderly for MyOrderlyListener {
    fn consume_message(
        &self,
        msgs: &[&MessageExt],
        context: &mut ConsumeOrderlyContext,
    ) -> ClientResult<ConsumeOrderlyStatus> {
        // Process messages sequentially
        context.set_auto_commit(false);
        Ok(ConsumeOrderlyStatus::Success)
    }
}

Required Methods

fn consume_message( &self, msgs: &[&MessageExt], context: &mut ConsumeOrderlyContext, ) -> ClientResult<ConsumeOrderlyStatus>

Processes a batch of messages in order.

Arguments

• msgs - Slice of message references. Zero-cost borrowing without cloning.
• context - Mutable context for orderly consumption. Allows control over commit behavior and queue suspension.

Returns

• Ok(ConsumeOrderlyStatus::Success) - Messages processed successfully, commit offset
• Ok(ConsumeOrderlyStatus::SuspendCurrentQueueAMoment) - Suspend consumption temporarily, retry later
• Ok(ConsumeOrderlyStatus::Rollback) - Rollback transaction
• Ok(ConsumeOrderlyStatus::Commit) - Commit transaction explicitly
• Err(error) - Processing failed with an error

Notes

• Messages from the same queue are NEVER processed concurrently
• Messages from different queues MAY be processed in parallel
• The context allows you to control auto-commit and suspension behavior

Dyn Compatibility

This trait is dyn compatible.

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

Implementors

impl<F> MessageListenerOrderly for F

where F: Fn(&[&MessageExt], &mut ConsumeOrderlyContext) -> ClientResult<ConsumeOrderlyStatus> + Send + Sync,

Implement MessageListenerOrderly 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.

Note on Mutability

The closure signature requires &mut ConsumeOrderlyContext to match the trait definition. This allows the closure to modify context state (auto-commit, suspension).