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_model::common::message::message_ext::MessageExt;
19use rocketmq_model::common::message::message_queue::MessageQueue;
20
21fn unsupported_mq_admin_operation(operation: &'static str) -> crate::ClientError {
22    crate::ClientError::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    ) -> crate::ClientResult<()> {
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    ) -> crate::ClientResult<()> {
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) -> crate::ClientResult<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) -> crate::ClientResult<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) -> crate::ClientResult<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) -> crate::ClientResult<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    ) -> crate::ClientResult<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) -> crate::ClientResult<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    ) -> crate::ClientResult<()>;
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    ) -> crate::ClientResult<()> {
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(&mut self, topic: &str) -> crate::ClientResult<Vec<MessageQueue>>;
114}
115
116#[cfg(test)]
117mod tests {
118    use super::*;
119    use crate::test_support::error_assertions::assert_invalid_argument;
120
121    struct CustomConsumer;
122
123    impl MQConsumer for CustomConsumer {
124        async fn send_message_back(
125            &mut self,
126            _msg: MessageExt,
127            _delay_level: i32,
128            _broker_name: &str,
129        ) -> crate::ClientResult<()> {
130            Ok(())
131        }
132
133        async fn fetch_subscribe_message_queues(&mut self, _topic: &str) -> crate::ClientResult<Vec<MessageQueue>> {
134            Ok(Vec::new())
135        }
136    }
137
138    #[tokio::test]
139    async fn admin_defaults_return_typed_unsupported_error_without_breaking_custom_consumers() {
140        let mut consumer = CustomConsumer;
141        let mq = MessageQueue::from_parts("topic", "broker-a", 0);
142
143        let error = consumer
144            .max_offset(&mq)
145            .await
146            .expect_err("default MQAdmin facade should be unsupported");
147
148        assert_invalid_argument(&error);
149    }
150}
rocketmq_client_rust/consumer/mq_consumer.rs

rocketmq_client_rust/consumer/
mq_consumer.rs
