rocketmq_client_rust/producer/
message_queue_selector.rs

1// Copyright 2023 The RocketMQ Rust Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::sync::Arc;
16
17use rocketmq_common::common::message::message_queue::MessageQueue;
18use rocketmq_common::common::message::MessageTrait;
19
20/// A trait for selecting a message queue from a list of available queues.
21///
22/// This trait provides a zero-cost abstraction for queue selection logic using compile-time
23/// monomorphization. Implement this trait for your custom selector logic, or use closures
24/// directly as they automatically implement this trait.
25///
26/// # Type Parameters
27///
28/// * `M` - Message type that implements `MessageTrait`
29/// * `A` - Argument type for custom selection logic
30///
31/// # Performance
32///
33/// This trait uses compile-time generics, allowing the compiler to:
34/// - Fully inline the selector function
35/// - Eliminate dynamic dispatch overhead
36/// - Perform aggressive optimizations
37///
38/// # Example
39///
40/// ```no_run
41/// use rocketmq_client_rust::producer::message_queue_selector::MessageQueueSelector;
42/// use rocketmq_common::common::message::message_queue::MessageQueue;
43/// use rocketmq_common::common::message::message_single::Message;
44///
45/// // Closures automatically implement MessageQueueSelector
46/// let selector = |mqs: &[MessageQueue], _msg: &Message, order_id: &i64| {
47///     let index = (*order_id % mqs.len() as i64) as usize;
48///     mqs.get(index).cloned()
49/// };
50/// ```
51pub trait MessageQueueSelector<M: MessageTrait, A>: Send + Sync {
52    /// Selects a message queue from the provided list.
53    ///
54    /// # Arguments
55    ///
56    /// * `mqs` - Available message queues to select from
57    /// * `msg` - The message to be sent
58    /// * `arg` - Custom argument for selection logic
59    ///
60    /// # Returns
61    ///
62    /// Selected `MessageQueue`, or `None` if no suitable queue is found
63    fn select(&self, mqs: &[MessageQueue], msg: &M, arg: &A) -> Option<MessageQueue>;
64}
65
66/// Implement MessageQueueSelector for all compatible closures and functions.
67///
68/// This allows closures to be used directly as selectors without explicit trait implementation.
69impl<F, M, A> MessageQueueSelector<M, A> for F
70where
71    F: Fn(&[MessageQueue], &M, &A) -> Option<MessageQueue> + Send + Sync,
72    M: MessageTrait,
73{
74    fn select(&self, mqs: &[MessageQueue], msg: &M, arg: &A) -> Option<MessageQueue> {
75        self(mqs, msg, arg)
76    }
77}
78
79/// Type-erased message queue selector function for storage and cross-boundary passing.
80///
81/// This type uses dynamic dispatch (`Arc<dyn Fn>`) and is suitable for scenarios where:
82/// - The selector needs to be stored in a struct field
83/// - The selector crosses async boundaries
84/// - The selector type cannot be determined at compile time
85///
86/// # Performance Note
87///
88/// This type incurs runtime overhead due to:
89/// - Dynamic dispatch (~5-10ns per call)
90/// - Arc reference counting
91/// - Type erasure with `dyn Any`
92///
93/// For best performance, prefer using generic parameters with the `MessageQueueSelector` trait
94/// in function signatures instead of this type alias.
95///
96/// # Example
97///
98/// ```no_run
99/// use rocketmq_client_rust::producer::message_queue_selector::MessageQueueSelectorFn;
100/// use std::sync::Arc;
101///
102/// let selector: MessageQueueSelectorFn = Arc::new(|mqs, _msg, arg| {
103///     let order_id = arg.downcast_ref::<i64>()?;
104///     let index = (*order_id % mqs.len() as i64) as usize;
105///     mqs.get(index).cloned()
106/// });
107/// ```
108pub type MessageQueueSelectorFn =
109    Arc<dyn Fn(&[MessageQueue], &dyn MessageTrait, &dyn std::any::Any) -> Option<MessageQueue> + Send + Sync>;
rocketmq_client_rust/producer/message_queue_selector.rs

rocketmq_client_rust/producer/
message_queue_selector.rs
