rocketmq_client_rust/consumer/
mq_consumer.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::collections::HashMap;
16
17use crate::base::query_result::QueryResult;
18use rocketmq_common::common::message::message_ext::MessageExt;
19use rocketmq_common::common::message::message_queue::MessageQueue;
20
21fn unsupported_mq_admin_operation(operation: &'static str) -> rocketmq_error::RocketMQError {
22    rocketmq_error::RocketMQError::illegal_argument(format!(
23        "{operation} is not supported by this MQConsumer implementation"
24    ))
25}
26
27#[allow(async_fn_in_trait)]
28pub trait MQConsumer {
29    /// Creates a topic through the consumer's admin facade.
30    ///
31    /// Java's `MQConsumer` extends `MQAdmin`; Rust keeps these admin operations async because all
32    /// broker I/O in this client is async.
33    async fn create_topic(
34        &mut self,
35        _key: &str,
36        _new_topic: &str,
37        _queue_num: i32,
38        _attributes: HashMap<String, String>,
39    ) -> rocketmq_error::RocketMQResult<()> {
40        Err(unsupported_mq_admin_operation("createTopic"))
41    }
42
43    /// Creates a topic with an explicit topic system flag.
44    async fn create_topic_with_flag(
45        &mut self,
46        _key: &str,
47        _new_topic: &str,
48        _queue_num: i32,
49        _topic_sys_flag: i32,
50        _attributes: HashMap<String, String>,
51    ) -> rocketmq_error::RocketMQResult<()> {
52        Err(unsupported_mq_admin_operation("createTopicWithFlag"))
53    }
54
55    /// Searches the offset in a queue by store timestamp.
56    async fn search_offset(&mut self, _mq: &MessageQueue, _timestamp: u64) -> rocketmq_error::RocketMQResult<i64> {
57        Err(unsupported_mq_admin_operation("searchOffset"))
58    }
59
60    /// Returns the broker max offset for a queue.
61    async fn max_offset(&mut self, _mq: &MessageQueue) -> rocketmq_error::RocketMQResult<i64> {
62        Err(unsupported_mq_admin_operation("maxOffset"))
63    }
64
65    /// Returns the broker min offset for a queue.
66    async fn min_offset(&mut self, _mq: &MessageQueue) -> rocketmq_error::RocketMQResult<i64> {
67        Err(unsupported_mq_admin_operation("minOffset"))
68    }
69
70    /// Returns the earliest store time for messages in a queue.
71    async fn earliest_msg_store_time(&mut self, _mq: &MessageQueue) -> rocketmq_error::RocketMQResult<i64> {
72        Err(unsupported_mq_admin_operation("earliestMsgStoreTime"))
73    }
74
75    /// Queries messages by key in a time range.
76    async fn query_message(
77        &mut self,
78        _topic: &str,
79        _key: &str,
80        _max_num: i32,
81        _begin: u64,
82        _end: u64,
83    ) -> rocketmq_error::RocketMQResult<QueryResult> {
84        Err(unsupported_mq_admin_operation("queryMessage"))
85    }
86
87    /// Views a message by message id.
88    async fn view_message(&mut self, _topic: &str, _msg_id: &str) -> rocketmq_error::RocketMQResult<MessageExt> {
89        Err(unsupported_mq_admin_operation("viewMessage"))
90    }
91
92    async fn send_message_back(
93        &mut self,
94        msg: MessageExt,
95        delay_level: i32,
96        broker_name: &str,
97    ) -> rocketmq_error::RocketMQResult<()>;
98
99    /// Sends a message back using the broker name carried by the message.
100    ///
101    /// This is the Rust equivalent of Java's two-argument
102    /// `sendMessageBack(MessageExt, int)` overload. Implementations may treat an
103    /// empty broker name as Java `null` and fall back to the message store host.
104    async fn send_message_back_to_origin_broker(
105        &mut self,
106        msg: MessageExt,
107        delay_level: i32,
108    ) -> rocketmq_error::RocketMQResult<()> {
109        let broker_name = msg.broker_name().to_string();
110        self.send_message_back(msg, delay_level, &broker_name).await
111    }
112
113    async fn fetch_subscribe_message_queues(
114        &mut self,
115        topic: &str,
116    ) -> rocketmq_error::RocketMQResult<Vec<MessageQueue>>;
117}
118
119#[cfg(test)]
120mod tests {
121    use super::*;
122
123    struct CustomConsumer;
124
125    impl MQConsumer for CustomConsumer {
126        async fn send_message_back(
127            &mut self,
128            _msg: MessageExt,
129            _delay_level: i32,
130            _broker_name: &str,
131        ) -> rocketmq_error::RocketMQResult<()> {
132            Ok(())
133        }
134
135        async fn fetch_subscribe_message_queues(
136            &mut self,
137            _topic: &str,
138        ) -> rocketmq_error::RocketMQResult<Vec<MessageQueue>> {
139            Ok(Vec::new())
140        }
141    }
142
143    #[tokio::test]
144    async fn admin_defaults_return_typed_unsupported_error_without_breaking_custom_consumers() {
145        let mut consumer = CustomConsumer;
146        let mq = MessageQueue::from_parts("topic", "broker-a", 0);
147
148        let error = consumer
149            .max_offset(&mq)
150            .await
151            .expect_err("default MQAdmin facade should be unsupported");
152
153        assert!(error.to_string().contains("maxOffset"));
154        assert!(error.to_string().contains("MQConsumer implementation"));
155    }
156}
rocketmq_client_rust/consumer/mq_consumer.rs

rocketmq_client_rust/consumer/
mq_consumer.rs
