rocketmq_protocol/protocol/
remoting_command_defaults.rs

1// Copyright 2026 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//! Immutable defaults and factories for business `RemotingCommand` values.
16
17use std::fmt;
18use std::sync::OnceLock;
19
20use bytes::Bytes;
21use cheetah_string::CheetahString;
22
23use crate::code::response_code::RemotingSysResponseCode;
24use crate::protocol::command_custom_header::CommandCustomHeader;
25use crate::protocol::remoting_command::RemotingCommand;
26use crate::protocol::SerializeType;
27
28const JAVA_DEFAULT_RESPONSE_REMARK: &str = "not set any response code";
29
30/// Wire defaults shared by all business command factories in a process.
31#[derive(Clone, Copy, Debug, PartialEq, Eq)]
32pub struct RemotingCommandDefaults {
33    version: i32,
34    serialize_type: SerializeType,
35}
36
37impl RemotingCommandDefaults {
38    pub const fn new(version: i32, serialize_type: SerializeType) -> Self {
39        Self {
40            version,
41            serialize_type,
42        }
43    }
44
45    pub const fn version(self) -> i32 {
46        self.version
47    }
48
49    pub const fn serialize_type(self) -> SerializeType {
50        self.serialize_type
51    }
52}
53
54impl Default for RemotingCommandDefaults {
55    fn default() -> Self {
56        Self::new(crate::version::CURRENT_VERSION as i32, SerializeType::JSON)
57    }
58}
59
60/// Builds business commands from one immutable set of wire defaults.
61///
62/// A factory does not read process configuration. Application owners may keep
63/// the process-default compatibility path or inject explicit factories when
64/// multiple remoting configurations must coexist in one process.
65///
66/// # Examples
67///
68/// ```
69/// use rocketmq_protocol::protocol::remoting_command_defaults::{
70///     RemotingCommandDefaults, RemotingCommandFactory,
71/// };
72/// use rocketmq_protocol::protocol::SerializeType;
73///
74/// let json = RemotingCommandFactory::new(RemotingCommandDefaults::new(101, SerializeType::JSON));
75/// let binary = RemotingCommandFactory::new(RemotingCommandDefaults::new(202, SerializeType::ROCKETMQ));
76///
77/// assert_eq!(json.create_remoting_command(10).version(), 101);
78/// assert_eq!(binary.create_remoting_command(10).version(), 202);
79/// ```
80#[derive(Clone, Copy, Debug, PartialEq, Eq)]
81pub struct RemotingCommandFactory {
82    defaults: RemotingCommandDefaults,
83}
84
85impl RemotingCommandFactory {
86    /// Creates a factory with explicit version and serialization defaults.
87    pub const fn new(defaults: RemotingCommandDefaults) -> Self {
88        Self { defaults }
89    }
90
91    /// Returns the immutable defaults owned by this factory.
92    pub const fn defaults(&self) -> RemotingCommandDefaults {
93        self.defaults
94    }
95
96    /// Creates a base command with an explicit code.
97    pub fn create_remoting_command(&self, code: impl Into<i32>) -> RemotingCommand {
98        RemotingCommand::with_resolved_defaults(self.defaults.version(), self.defaults.serialize_type()).set_code(code)
99    }
100
101    /// Creates a request with an external body.
102    pub fn create_request(&self, code: impl Into<i32>, body: impl Into<Bytes>) -> RemotingCommand {
103        self.create_remoting_command(code).set_body(body)
104    }
105
106    /// Creates a request with a typed custom header.
107    pub fn create_request_command<T>(&self, code: impl Into<i32>, header: T) -> RemotingCommand
108    where
109        T: CommandCustomHeader + Sync + Send + 'static,
110    {
111        self.create_remoting_command(code).set_command_custom_header(header)
112    }
113
114    /// Creates a response with an explicit response code.
115    pub fn create_response_command_with_code(&self, code: impl Into<i32>) -> RemotingCommand {
116        self.create_remoting_command(code).mark_response_type()
117    }
118
119    /// Creates a response with an explicit code and typed custom header.
120    pub fn create_response_command_with_code_and_header(
121        &self,
122        code: impl Into<i32>,
123        header: impl CommandCustomHeader + Sync + Send + 'static,
124    ) -> RemotingCommand {
125        self.create_response_command_with_code(code)
126            .set_command_custom_header(header)
127    }
128
129    /// Creates a response with an explicit code and remark.
130    pub fn create_response_command_with_code_remark(
131        &self,
132        code: impl Into<i32>,
133        remark: impl Into<CheetahString>,
134    ) -> RemotingCommand {
135        self.create_remoting_command(code)
136            .set_remark_option(Some(remark.into()))
137            .mark_response_type()
138    }
139
140    /// Creates an explicitly successful response.
141    pub fn create_success_response_command(&self) -> RemotingCommand {
142        self.create_response_command_with_code(RemotingSysResponseCode::Success)
143    }
144
145    /// Creates an explicitly successful response with a typed custom header.
146    pub fn create_success_response_command_with_header(
147        &self,
148        header: impl CommandCustomHeader + Sync + Send + 'static,
149    ) -> RemotingCommand {
150        self.create_response_command_with_code_and_header(RemotingSysResponseCode::Success, header)
151    }
152
153    /// Creates Java's unset error response.
154    pub fn create_java_default_error_response_command(&self) -> RemotingCommand {
155        self.create_response_command_with_code_remark(
156            RemotingSysResponseCode::SystemError,
157            JAVA_DEFAULT_RESPONSE_REMARK,
158        )
159    }
160
161    /// Creates Java's unset error response with a typed custom header.
162    pub fn create_java_default_error_response_command_with_header(
163        &self,
164        header: impl CommandCustomHeader + Sync + Send + 'static,
165    ) -> RemotingCommand {
166        self.create_java_default_error_response_command()
167            .set_command_custom_header(header)
168    }
169}
170
171/// Returned when a process attempts to replace initialized command defaults.
172#[derive(Clone, Copy, Debug, PartialEq, Eq)]
173pub struct RemotingCommandDefaultsConflict {
174    initialized: RemotingCommandDefaults,
175    requested: RemotingCommandDefaults,
176}
177
178impl RemotingCommandDefaultsConflict {
179    pub const fn initialized(self) -> RemotingCommandDefaults {
180        self.initialized
181    }
182
183    pub const fn requested(self) -> RemotingCommandDefaults {
184        self.requested
185    }
186}
187
188impl fmt::Display for RemotingCommandDefaultsConflict {
189    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
190        write!(
191            formatter,
192            "remoting command defaults are already initialized to {:?}, requested {:?}",
193            self.initialized, self.requested
194        )
195    }
196}
197
198impl std::error::Error for RemotingCommandDefaultsConflict {}
199
200static APPLICATION_DEFAULTS: OnceLock<RemotingCommandDefaults> = OnceLock::new();
201
202fn initialize_defaults(
203    cell: &OnceLock<RemotingCommandDefaults>,
204    defaults: RemotingCommandDefaults,
205) -> Result<(), RemotingCommandDefaultsConflict> {
206    match cell.set(defaults) {
207        Ok(()) => Ok(()),
208        Err(requested) => {
209            // OnceLock returns Err only after retaining the previously initialized value.
210            let initialized = *cell
211                .get()
212                .expect("failed OnceLock::set always retains an initialized value");
213            if initialized == requested {
214                Ok(())
215            } else {
216                Err(RemotingCommandDefaultsConflict { initialized, requested })
217            }
218        }
219    }
220}
221
222/// Initializes the immutable defaults used by business command factories.
223///
224/// Repeating initialization with the same value is idempotent. A different
225/// value is rejected so wire defaults cannot change after command creation.
226///
227/// # Errors
228///
229/// Returns [`RemotingCommandDefaultsConflict`] when different defaults were
230/// initialized earlier in the process.
231pub fn initialize_remoting_command_defaults(
232    defaults: RemotingCommandDefaults,
233) -> Result<(), RemotingCommandDefaultsConflict> {
234    initialize_defaults(&APPLICATION_DEFAULTS, defaults)
235}
236
237pub(crate) fn application_remoting_command_defaults() -> RemotingCommandDefaults {
238    *APPLICATION_DEFAULTS.get_or_init(RemotingCommandDefaults::default)
239}
240
241/// Returns the immutable compatibility factory selected by the application.
242///
243/// Instance owners should capture this value once at their composition
244/// boundary and pass it to request/response producers. The returned factory
245/// does not read environment variables.
246pub fn application_remoting_command_factory() -> RemotingCommandFactory {
247    RemotingCommandFactory::new(application_remoting_command_defaults())
248}
249
250#[cfg(test)]
251mod tests {
252    use rocketmq_error::CanonicalCondition;
253
254    use super::*;
255
256    #[test]
257    fn initialization_is_idempotent_and_rejects_conflicts() {
258        let cell = OnceLock::new();
259        let first = RemotingCommandDefaults::new(1, SerializeType::JSON);
260        let second = RemotingCommandDefaults::new(2, SerializeType::ROCKETMQ);
261
262        assert_eq!(initialize_defaults(&cell, first), Ok(()));
263        assert_eq!(initialize_defaults(&cell, first), Ok(()));
264        assert_eq!(
265            initialize_defaults(&cell, second),
266            Err(RemotingCommandDefaultsConflict {
267                initialized: first,
268                requested: second,
269            })
270        );
271    }
272
273    #[test]
274    fn initialized_defaults_conflict_is_a_failed_precondition() {
275        let cell = OnceLock::new();
276        let initialized = RemotingCommandDefaults::new(1, SerializeType::JSON);
277        let requested = RemotingCommandDefaults::new(2, SerializeType::ROCKETMQ);
278        initialize_defaults(&cell, initialized).expect("first defaults initialize");
279
280        let source = initialize_defaults(&cell, requested).expect_err("different defaults conflict");
281        let error = crate::ProtocolContractViolation::from(source);
282
283        assert_eq!(error.condition(), CanonicalCondition::FailedPrecondition);
284        assert_eq!(
285            error.to_string(),
286            "remoting command defaults conflict with the initialized value"
287        );
288    }
289}
rocketmq_protocol/protocol/remoting_command_defaults.rs

rocketmq_protocol/protocol/
remoting_command_defaults.rs
