rocketmq_protocol/code/
request_code.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
15/// Macro to define RequestCode enum with automatic conversion implementations.
16/// This reduces code duplication and makes it easier to maintain.
17macro_rules! define_request_code {
18    (
19        $(#[$enum_meta:meta])*
20        pub enum $enum_name:ident {
21            $(
22                $(#[$variant_meta:meta])*
23                $variant:ident = $value:expr
24            ),* $(,)?
25        }
26    ) => {
27        $(#[$enum_meta])*
28        #[repr(i32)]
29        pub enum $enum_name {
30            $(
31                $(#[$variant_meta])*
32                $variant = $value,
33            )*
34        }
35
36        impl From<$enum_name> for i32 {
37            #[inline]
38            fn from(value: $enum_name) -> Self {
39                value as i32
40            }
41        }
42
43        impl From<i32> for $enum_name {
44            #[inline]
45            fn from(value: i32) -> Self {
46                match value {
47                    $($value => $enum_name::$variant,)*
48                    _ => $enum_name::Unknown,
49                }
50            }
51        }
52
53        impl $enum_name {
54            /// Convert to i32 value
55            #[inline]
56            pub const fn to_i32(self) -> i32 {
57                self as i32
58            }
59
60            /// Check if the request code is unknown
61            #[inline]
62            pub const fn is_unknown(&self) -> bool {
63                matches!(self, Self::Unknown)
64            }
65        }
66    };
67}
68
69define_request_code! {
70    #[derive(Debug, Eq, PartialEq, Clone, Copy, Hash)]
71    pub enum RequestCode {
72        SendMessage = 10,
73        PullMessage = 11,
74        QueryMessage = 12,
75        QueryBrokerOffset = 13, // Not used in Java
76        QueryConsumerOffset = 14,
77        UpdateConsumerOffset = 15,
78        UpdateAndCreateTopic = 17,
79        UpdateAndCreateTopicList = 18,
80        GetAllTopicConfig = 21,
81        GetTopicConfigList = 22, // Not used in Java
82        GetTopicNameList = 23,   // Not used in Java
83        UpdateBrokerConfig = 25,
84        GetBrokerConfig = 26,
85        TriggerDeleteFiles = 27, // Not used in Java
86        GetBrokerRuntimeInfo = 28,
87        SearchOffsetByTimestamp = 29,
88        GetMaxOffset = 30,
89        GetMinOffset = 31,
90        GetEarliestMsgStoreTime = 32,
91        ViewMessageById = 33,
92        HeartBeat = 34,
93        UnregisterClient = 35,
94        ConsumerSendMsgBack = 36,
95        EndTransaction = 37,
96        GetConsumerListByGroup = 38,
97        CheckTransactionState = 39,
98        NotifyConsumerIdsChanged = 40,
99        LockBatchMq = 41,
100        UnlockBatchMq = 42,
101        GetAllConsumerOffset = 43,
102        GetAllDelayOffset = 45,
103        CheckClientConfig = 46,
104        GetClientConfig = 47, // Not used in Java
105        UpdateAndCreateAclConfig = 50,
106        DeleteAclConfig = 51,
107        GetBrokerClusterAclInfo = 52,
108        UpdateGlobalWhiteAddrsConfig = 53,
109        GetBrokerClusterAclConfig = 54, // Deprecated
110        GetTimerCheckPoint = 60,
111        GetTimerMetrics = 61,
112
113        PopMessage = 200050,
114        AckMessage = 200051,
115        BatchAckMessage = 200151,
116        PeekMessage = 200052,
117        ChangeMessageInvisibleTime = 200053,
118        Notification = 200054,
119        PollingInfo = 200055,
120        PopRollback = 200056,
121
122        PopLiteMessage = 200070,
123        LiteSubscriptionCtl = 200071,
124        AckLiteMessage = 200072,
125        NotifyUnsubscribeLite = 200073,
126        GetBrokerLiteInfo = 200074,
127        GetParentTopicInfo = 200075,
128        GetLiteTopicInfo = 200076,
129        GetLiteClientInfo = 200077,
130        GetLiteGroupInfo = 200078,
131        TriggerLiteDispatch = 200079,
132
133        PutKvConfig = 100,
134        GetKvConfig = 101,
135        DeleteKvConfig = 102,
136        RegisterBroker = 103,
137        UnregisterBroker = 104,
138        GetRouteinfoByTopic = 105,
139        GetBrokerClusterInfo = 106,
140        UpdateAndCreateSubscriptionGroup = 200,
141        GetAllSubscriptionGroupConfig = 201,
142        GetTopicStatsInfo = 202,
143        GetConsumerConnectionList = 203,
144        GetProducerConnectionList = 204,
145        WipeWritePermOfBroker = 205,
146        GetAllTopicListFromNameserver = 206,
147        DeleteSubscriptionGroup = 207,
148        GetConsumeStats = 208,
149
150        SuspendConsumer = 209,               // Not used in Java
151        ResumeConsumer = 210,                // Not used in Java
152        ResetConsumerOffsetInConsumer = 211, // Not used in Java
153        ResetConsumerOffsetInBroker = 212,   // Not used in Java
154        AdjustConsumerThreadPool = 213,      // Not used in Java
155        WhoConsumeTheMessage = 214,          // Not used in Java
156
157        DeleteTopicInBroker = 215,
158        DeleteTopicInNamesrv = 216,
159        RegisterTopicInNamesrv = 217,
160        GetKvlistByNamespace = 219,
161        ResetConsumerClientOffset = 220,
162        GetConsumerStatusFromClient = 221,
163        InvokeBrokerToResetOffset = 222,
164        InvokeBrokerToGetConsumerStatus = 223,
165        UpdateAndCreateSubscriptionGroupList = 225,
166
167        QueryTopicConsumeByWho = 300,
168        GetTopicsByCluster = 224,
169        QueryTopicsByConsumer = 343,
170        QuerySubscriptionByConsumer = 345,
171
172        RegisterFilterServer = 301,       // Not used in Java
173        RegisterMessageFilterClass = 302, // Not used in Java
174
175        QueryConsumeTimeSpan = 303,
176        GetSystemTopicListFromNs = 304,
177        GetSystemTopicListFromBroker = 305,
178        CleanExpiredConsumequeue = 306,
179        GetConsumerRunningInfo = 307,
180        QueryCorrectionOffset = 308,
181        ConsumeMessageDirectly = 309,
182        SendMessageV2 = 310,
183        GetUnitTopicList = 311,
184        GetHasUnitSubTopicList = 312,
185        GetHasUnitSubUnunitTopicList = 313,
186        CloneGroupOffset = 314,
187        ViewBrokerStatsData = 315,
188        CleanUnusedTopic = 316,
189        GetBrokerConsumeStats = 317,
190        UpdateNamesrvConfig = 318,
191        GetNamesrvConfig = 319,
192        SendBatchMessage = 320,
193        QueryConsumeQueue = 321,
194        QueryDataVersion = 322,
195        ResumeCheckHalfMessage = 323,
196        SendReplyMessage = 324,
197        SendReplyMessageV2 = 325,
198        PushReplyMessageToClient = 326,
199        AddWritePermOfBroker = 327,
200        /// Generation-checked broker configuration update used by supervised SRE execution.
201        UpdateBrokerConfigCas = 330,
202        /// Returns the authenticated, bounded Proxy drain state.
203        GetProxyDrainState = 331,
204        /// Starts one authenticated, reversible Proxy drain operation.
205        BeginProxyDrain = 332,
206        /// Cancels one authenticated Proxy drain operation and restores admission.
207        CancelProxyDrain = 333,
208        /// Version-checked allowlisted Topic configuration update used by supervised SRE execution.
209        UpdateTopicConfigCas = 334,
210        /// Version-checked allowlisted Subscription Group update used by supervised SRE execution.
211        UpdateSubscriptionGroupConfigCas = 335,
212        /// Full Topic replacement guarded by an absent-or-versioned expected state.
213        UpdateTopicConfigStateCas = 336,
214        /// Full Subscription Group replacement guarded by an absent-or-versioned expected state.
215        UpdateSubscriptionGroupConfigStateCas = 337,
216        /// Exact queue offset replacement guarded by the value observed during preflight.
217        UpdateConsumerOffsetConditional = 338,
218        /// Returns only the Broker fields allowed by the supervised mutation boundary.
219        GetBrokerMutationConfig = 339,
220        GetTopicConfig = 351,
221        GetSubscriptionGroupConfig = 352,
222        UpdateAndGetGroupForbidden = 353,
223        CheckRocksdbCqWriteProgress = 354,
224        ExportRocksdbConfigToJson = 355,
225
226        LitePullMessage = 361,
227        RecallMessage = 370,
228
229        QueryAssignment = 400,
230        SetMessageRequestMode = 401,
231        GetAllMessageRequestMode = 402,
232        /// Returns one exact Topic and consumer-group request-mode state.
233        GetMessageRequestMode = 403,
234        /// Replaces one request mode only when its exact preflight state still matches.
235        SetMessageRequestModeCas = 404,
236        UpdateAndCreateStaticTopic = 513,
237        GetBrokerMemberGroup = 901,
238        AddBroker = 902,
239        RemoveBroker = 903,
240        BrokerHeartbeat = 904,
241        NotifyMinBrokerIdChange = 905,
242        ExchangeBrokerHaInfo = 906,
243        GetBrokerHaStatus = 907,
244        ResetMasterFlushOffset = 908,
245        GetAllProducerInfo = 328,
246        DeleteExpiredCommitlog = 329,
247
248        UpdateColdDataFlowCtrConfig = 2001,
249        RemoveColdDataFlowCtrConfig = 2002,
250        GetColdDataFlowCtrInfo = 2003,
251        SetCommitlogReadMode = 2004,
252
253        // Controller codes
254        ControllerAlterSyncStateSet = 1001,
255        ControllerElectMaster = 1002,
256        ControllerRegisterBroker = 1003,
257        ControllerGetReplicaInfo = 1004,
258        ControllerGetMetadataInfo = 1005,
259        ControllerGetSyncStateData = 1006,
260        GetBrokerEpochCache = 1007,
261        NotifyBrokerRoleChanged = 1008,
262        UpdateControllerConfig = 1009,
263        GetControllerConfig = 1010,
264
265        CleanBrokerData = 1011,
266        ControllerGetNextBrokerId = 1012,
267        ControllerApplyBrokerId = 1013,
268        BrokerCloseChannelRequest = 1014,
269        CheckNotActiveBrokerRequest = 1015,
270        GetBrokerLiveInfoRequest = 1016,
271        GetSyncStateDataRequest = 1017,
272        RaftBrokerHeartBeatEventRequest = 1018,
273
274        // Auth codes
275        AuthCreateUser = 3001,
276        AuthUpdateUser = 3002,
277        AuthDeleteUser = 3003,
278        AuthGetUser = 3004,
279        AuthListUsers = 3005,
280        AuthCreateAcl = 3006,
281        AuthUpdateAcl = 3007,
282        AuthDeleteAcl = 3008,
283        AuthGetAcl = 3009,
284        AuthListAcl = 3010,
285
286        SwitchTimerEngine = 5001,
287        DeleteTopicInBrokerList = 5002,
288        DeleteSubscriptionGroupList = 5003,
289
290        // Privileged maintenance codes. These are not ordinary Admin actions.
291        MaintenanceGetCapabilities = 6001,
292        MaintenanceCreateControllerSnapshot = 6002,
293        MaintenanceCreateStoreCheckpoint = 6003,
294        MaintenanceVerifyCheckpoint = 6004,
295        MaintenanceRestoreVerify = 6005,
296
297        Unknown = -9999999,
298    }
299}
300
301#[cfg(test)]
302mod tests {
303    use super::*;
304
305    #[test]
306    fn test_request_code_to_i32() {
307        assert_eq!(RequestCode::SendMessage.to_i32(), 10);
308        assert_eq!(RequestCode::PullMessage.to_i32(), 11);
309        assert_eq!(RequestCode::HeartBeat.to_i32(), 34);
310        assert_eq!(RequestCode::PopMessage.to_i32(), 200050);
311        assert_eq!(RequestCode::ControllerAlterSyncStateSet.to_i32(), 1001);
312        assert_eq!(RequestCode::UpdateBrokerConfigCas.to_i32(), 330);
313        assert_eq!(RequestCode::GetProxyDrainState.to_i32(), 331);
314        assert_eq!(RequestCode::BeginProxyDrain.to_i32(), 332);
315        assert_eq!(RequestCode::CancelProxyDrain.to_i32(), 333);
316        assert_eq!(RequestCode::UpdateTopicConfigCas.to_i32(), 334);
317        assert_eq!(RequestCode::UpdateSubscriptionGroupConfigCas.to_i32(), 335);
318        assert_eq!(RequestCode::UpdateTopicConfigStateCas.to_i32(), 336);
319        assert_eq!(RequestCode::UpdateSubscriptionGroupConfigStateCas.to_i32(), 337);
320        assert_eq!(RequestCode::UpdateConsumerOffsetConditional.to_i32(), 338);
321        assert_eq!(RequestCode::GetBrokerMutationConfig.to_i32(), 339);
322        assert_eq!(RequestCode::GetMessageRequestMode.to_i32(), 403);
323        assert_eq!(RequestCode::SetMessageRequestModeCas.to_i32(), 404);
324        assert_eq!(RequestCode::AuthCreateUser.to_i32(), 3001);
325        assert_eq!(RequestCode::MaintenanceCreateControllerSnapshot.to_i32(), 6002);
326        assert_eq!(RequestCode::DeleteTopicInBrokerList.to_i32(), 5002);
327        assert_eq!(RequestCode::DeleteSubscriptionGroupList.to_i32(), 5003);
328        assert_eq!(RequestCode::Unknown.to_i32(), -9999999);
329    }
330
331    #[test]
332    fn test_i32_to_request_code() {
333        assert_eq!(RequestCode::from(10), RequestCode::SendMessage);
334        assert_eq!(RequestCode::from(11), RequestCode::PullMessage);
335        assert_eq!(RequestCode::from(34), RequestCode::HeartBeat);
336        assert_eq!(RequestCode::from(200050), RequestCode::PopMessage);
337        assert_eq!(RequestCode::from(1001), RequestCode::ControllerAlterSyncStateSet);
338        assert_eq!(RequestCode::from(330), RequestCode::UpdateBrokerConfigCas);
339        assert_eq!(RequestCode::from(331), RequestCode::GetProxyDrainState);
340        assert_eq!(RequestCode::from(332), RequestCode::BeginProxyDrain);
341        assert_eq!(RequestCode::from(333), RequestCode::CancelProxyDrain);
342        assert_eq!(RequestCode::from(334), RequestCode::UpdateTopicConfigCas);
343        assert_eq!(RequestCode::from(335), RequestCode::UpdateSubscriptionGroupConfigCas);
344        assert_eq!(RequestCode::from(336), RequestCode::UpdateTopicConfigStateCas);
345        assert_eq!(
346            RequestCode::from(337),
347            RequestCode::UpdateSubscriptionGroupConfigStateCas
348        );
349        assert_eq!(RequestCode::from(338), RequestCode::UpdateConsumerOffsetConditional);
350        assert_eq!(RequestCode::from(339), RequestCode::GetBrokerMutationConfig);
351        assert_eq!(RequestCode::from(403), RequestCode::GetMessageRequestMode);
352        assert_eq!(RequestCode::from(404), RequestCode::SetMessageRequestModeCas);
353        assert_eq!(RequestCode::from(3001), RequestCode::AuthCreateUser);
354        assert_eq!(RequestCode::from(6003), RequestCode::MaintenanceCreateStoreCheckpoint);
355        assert_eq!(RequestCode::from(5002), RequestCode::DeleteTopicInBrokerList);
356        assert_eq!(RequestCode::from(5003), RequestCode::DeleteSubscriptionGroupList);
357        assert_eq!(RequestCode::from(-9999999), RequestCode::Unknown);
358    }
359
360    #[test]
361    fn test_unknown_code_conversion() {
362        // Any unknown i32 value should return Unknown
363        assert_eq!(RequestCode::from(99999), RequestCode::Unknown);
364        assert_eq!(RequestCode::from(-1), RequestCode::Unknown);
365        assert_eq!(RequestCode::from(0), RequestCode::Unknown);
366        assert_eq!(RequestCode::from(999), RequestCode::Unknown);
367    }
368
369    #[test]
370    fn test_is_unknown() {
371        assert!(RequestCode::Unknown.is_unknown());
372        assert!(!RequestCode::SendMessage.is_unknown());
373        assert!(!RequestCode::HeartBeat.is_unknown());
374        assert!(!RequestCode::PopMessage.is_unknown());
375
376        // Test with unknown code conversion
377        let unknown_code = RequestCode::from(12345);
378        assert!(unknown_code.is_unknown());
379    }
380
381    #[test]
382    fn test_request_code_from_trait() {
383        let code: i32 = RequestCode::SendMessage.into();
384        assert_eq!(code, 10);
385
386        let code: i32 = RequestCode::Unknown.into();
387        assert_eq!(code, -9999999);
388    }
389
390    #[test]
391    fn test_round_trip_conversion() {
392        // Test that converting to i32 and back gives the same value
393        let codes = vec![
394            RequestCode::SendMessage,
395            RequestCode::PullMessage,
396            RequestCode::QueryMessage,
397            RequestCode::HeartBeat,
398            RequestCode::PopMessage,
399            RequestCode::RegisterBroker,
400            RequestCode::ControllerElectMaster,
401            RequestCode::AuthCreateUser,
402            RequestCode::MaintenanceRestoreVerify,
403            RequestCode::Unknown,
404        ];
405
406        for code in codes {
407            let i32_val = code.to_i32();
408            let converted_back = RequestCode::from(i32_val);
409            assert_eq!(code, converted_back, "Round trip failed for {:?}", code);
410        }
411    }
412
413    #[test]
414    fn test_special_codes() {
415        // Test Pop message range (200050-200055)
416        assert_eq!(RequestCode::from(200050), RequestCode::PopMessage);
417        assert_eq!(RequestCode::from(200051), RequestCode::AckMessage);
418        assert_eq!(RequestCode::from(200151), RequestCode::BatchAckMessage);
419        assert_eq!(RequestCode::from(200052), RequestCode::PeekMessage);
420        assert_eq!(RequestCode::from(200053), RequestCode::ChangeMessageInvisibleTime);
421        assert_eq!(RequestCode::from(200054), RequestCode::Notification);
422        assert_eq!(RequestCode::from(200055), RequestCode::PollingInfo);
423
424        // Test Controller codes (1001-1018)
425        assert_eq!(RequestCode::from(1001), RequestCode::ControllerAlterSyncStateSet);
426        assert_eq!(RequestCode::from(1018), RequestCode::RaftBrokerHeartBeatEventRequest);
427
428        // Test Auth codes (3001-3010)
429        assert_eq!(RequestCode::from(3001), RequestCode::AuthCreateUser);
430        assert_eq!(RequestCode::from(3010), RequestCode::AuthListAcl);
431
432        // Privileged maintenance codes (6001-6005)
433        assert_eq!(RequestCode::from(6001), RequestCode::MaintenanceGetCapabilities);
434        assert_eq!(RequestCode::from(6005), RequestCode::MaintenanceRestoreVerify);
435
436        // Test Cold data flow codes (2001-2004)
437        assert_eq!(RequestCode::from(2001), RequestCode::UpdateColdDataFlowCtrConfig);
438        assert_eq!(RequestCode::from(2004), RequestCode::SetCommitlogReadMode);
439    }
440
441    #[test]
442    fn test_const_fn_to_i32() {
443        // Verify that to_i32 is const and can be used in const context
444        const CODE: i32 = RequestCode::SendMessage.to_i32();
445        assert_eq!(CODE, 10);
446    }
447
448    #[test]
449    fn test_repr_i32_size() {
450        // Verify that RequestCode has the same size as i32
451        use std::mem::size_of;
452        assert_eq!(size_of::<RequestCode>(), size_of::<i32>());
453    }
454}
rocketmq_protocol/code/request_code.rs

rocketmq_protocol/code/
request_code.rs
