rocketmq_protocol/protocol/
remoting_command_compat.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//! Remoting-owned construction and source-compatibility helpers.
16//!
17//! `RemotingCommand` itself is always the canonical protocol type. Process
18//! environment defaults remain here so the protocol owner does not acquire an
19//! environment dependency.
20
21use std::fmt;
22use std::sync::LazyLock;
23use std::sync::OnceLock;
24
25use crate::protocol::command_custom_header::CommandCustomHeader;
26use crate::protocol::remoting_command::RemotingCommand;
27use crate::protocol::remoting_command_defaults::initialize_remoting_command_defaults;
28use crate::protocol::remoting_command_defaults::RemotingCommandDefaults;
29use crate::protocol::remoting_command_defaults::RemotingCommandFactory;
30use crate::protocol::SerializeType;
31use crate::ProtocolContractViolation;
32
33use super::remoting_command::REMOTING_VERSION_KEY;
34use super::remoting_command::SERIALIZE_TYPE_ENV;
35use super::remoting_command::SERIALIZE_TYPE_PROPERTY;
36
37/// Returned when the process attempts to replace an initialized remoting
38/// version with a different value.
39#[derive(Clone, Copy, Debug, PartialEq, Eq)]
40pub struct RemotingVersionConflict {
41    initialized: i32,
42    requested: i32,
43}
44
45impl RemotingVersionConflict {
46    /// Returns the immutable version already selected for the process.
47    pub fn initialized(self) -> i32 {
48        self.initialized
49    }
50
51    /// Returns the conflicting version requested by the caller.
52    pub fn requested(self) -> i32 {
53        self.requested
54    }
55}
56
57impl fmt::Display for RemotingVersionConflict {
58    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
59        write!(
60            formatter,
61            "remoting version is already initialized to {}, requested {}",
62            self.initialized, self.requested
63        )
64    }
65}
66
67impl std::error::Error for RemotingVersionConflict {}
68
69static REMOTING_VERSION: OnceLock<i32> = OnceLock::new();
70
71static SERIALIZE_TYPE: LazyLock<SerializeType> = LazyLock::new(|| {
72    std::env::var(SERIALIZE_TYPE_PROPERTY)
73        .or_else(|_| std::env::var(SERIALIZE_TYPE_ENV))
74        .ok()
75        .and_then(|value| match value.as_str() {
76            "JSON" => Some(SerializeType::JSON),
77            "ROCKETMQ" => Some(SerializeType::ROCKETMQ),
78            _ => None,
79        })
80        .unwrap_or(SerializeType::JSON)
81});
82
83/// An unsupported value selected for the process remoting serialization type.
84#[derive(Clone, PartialEq, Eq)]
85pub struct InvalidRemotingSerializeType {
86    key: &'static str,
87    value: String,
88}
89
90impl InvalidRemotingSerializeType {
91    pub const fn key(&self) -> &'static str {
92        self.key
93    }
94
95    pub fn value(&self) -> &str {
96        &self.value
97    }
98}
99
100impl fmt::Display for InvalidRemotingSerializeType {
101    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
102        write!(
103            formatter,
104            "invalid remoting serialization type for {}: {:?}; expected JSON or ROCKETMQ",
105            self.key, self.value
106        )
107    }
108}
109
110impl std::error::Error for InvalidRemotingSerializeType {}
111
112/// Resolves the serialization type with Java-compatible configuration precedence.
113///
114/// The property-style value takes precedence over the environment fallback.
115/// When neither value is configured, JSON is selected.
116///
117/// # Errors
118///
119/// Returns [`ProtocolContractViolation::InvalidSerializeType`] when the
120/// selected value is not exactly `JSON` or `ROCKETMQ`.
121pub fn resolve_remoting_serialize_type(
122    property_value: Option<&str>,
123    environment_value: Option<&str>,
124) -> Result<SerializeType, ProtocolContractViolation> {
125    let (key, value) = property_value
126        .map(|value| (SERIALIZE_TYPE_PROPERTY, value))
127        .or_else(|| environment_value.map(|value| (SERIALIZE_TYPE_ENV, value)))
128        .unwrap_or((SERIALIZE_TYPE_PROPERTY, "JSON"));
129
130    match value {
131        "JSON" => Ok(SerializeType::JSON),
132        "ROCKETMQ" => Ok(SerializeType::ROCKETMQ),
133        _ => Err(InvalidRemotingSerializeType {
134            key,
135            value: value.to_string(),
136        }
137        .into()),
138    }
139}
140
141impl fmt::Debug for InvalidRemotingSerializeType {
142    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
143        formatter
144            .debug_struct("InvalidRemotingSerializeType")
145            .field("key", &self.key)
146            .field("value_present", &true)
147            .finish()
148    }
149}
150
151fn read_environment_value(key: &'static str) -> Result<Option<String>, ProtocolContractViolation> {
152    match std::env::var(key) {
153        Ok(value) => Ok(Some(value)),
154        Err(std::env::VarError::NotPresent) => Ok(None),
155        Err(std::env::VarError::NotUnicode(value)) => Err(InvalidRemotingSerializeType {
156            key,
157            value: value.to_string_lossy().into_owned(),
158        }
159        .into()),
160    }
161}
162
163/// Initializes the immutable defaults used by all business command factories.
164///
165/// `rocketmq.serialize.type` takes precedence over
166/// `ROCKETMQ_SERIALIZE_TYPE`. Unsupported values fail startup instead of
167/// silently selecting JSON.
168///
169/// # Errors
170///
171/// Returns [`ProtocolContractViolation`] when configuration is invalid or a
172/// different process default was initialized earlier.
173pub fn initialize_remoting_command_factory(version: i32) -> Result<RemotingCommandFactory, ProtocolContractViolation> {
174    let property_value = read_environment_value(SERIALIZE_TYPE_PROPERTY)?;
175    let environment_value = if property_value.is_none() {
176        read_environment_value(SERIALIZE_TYPE_ENV)?
177    } else {
178        None
179    };
180    let serialize_type = resolve_remoting_serialize_type(property_value.as_deref(), environment_value.as_deref())?;
181    let defaults = RemotingCommandDefaults::new(version, serialize_type);
182    initialize_remoting_command_defaults(defaults)?;
183    Ok(RemotingCommandFactory::new(defaults))
184}
185
186/// Initializes the immutable defaults used by all business command factories.
187///
188/// This compatibility entry point discards the resolved factory. New
189/// application owners should retain the value returned by
190/// [`initialize_remoting_command_factory`] and inject it into command
191/// producers.
192///
193/// # Errors
194///
195/// Returns [`ProtocolContractViolation`] when configuration is invalid or a
196/// different process default was initialized earlier.
197pub fn initialize_remoting_defaults(version: i32) -> Result<(), ProtocolContractViolation> {
198    initialize_remoting_command_factory(version)?;
199    Ok(())
200}
201
202fn resolve_remoting_version(cell: &OnceLock<i32>) -> i32 {
203    *cell.get_or_init(|| {
204        std::env::var(REMOTING_VERSION_KEY)
205            .ok()
206            .and_then(|value| value.parse::<i32>().ok())
207            .unwrap_or(crate::version::CURRENT_VERSION as i32)
208    })
209}
210
211fn initialize_version(cell: &OnceLock<i32>, version: i32) -> Result<(), RemotingVersionConflict> {
212    match cell.set(version) {
213        Ok(()) => Ok(()),
214        Err(requested) => {
215            let initialized = resolve_remoting_version(cell);
216            if initialized == requested {
217                Ok(())
218            } else {
219                Err(RemotingVersionConflict { initialized, requested })
220            }
221        }
222    }
223}
224
225/// Selects the immutable remoting version used by compatibility constructors.
226///
227/// Repeating initialization with the same version is idempotent. A different
228/// value is rejected so callers cannot silently change wire defaults after
229/// other threads have started constructing commands.
230///
231/// # Errors
232///
233/// Returns [`RemotingVersionConflict`] when a different value was initialized
234/// earlier.
235pub fn initialize_remoting_version(version: i32) -> Result<(), RemotingVersionConflict> {
236    initialize_version(&REMOTING_VERSION, version)
237}
238
239fn compatibility_factory() -> RemotingCommandFactory {
240    RemotingCommandFactory::new(RemotingCommandDefaults::new(
241        resolve_remoting_version(&REMOTING_VERSION),
242        *SERIALIZE_TYPE,
243    ))
244}
245
246pub fn create_remoting_command(code: impl Into<i32>) -> RemotingCommand {
247    compatibility_factory().create_remoting_command(code)
248}
249
250pub fn create_request_command<T>(code: impl Into<i32>, header: T) -> RemotingCommand
251where
252    T: CommandCustomHeader + Sync + Send + 'static,
253{
254    compatibility_factory().create_request_command(code, header)
255}
256
257pub fn create_response_command() -> RemotingCommand {
258    compatibility_factory().create_success_response_command()
259}
260
261#[cfg(test)]
262mod tests {
263    use std::sync::OnceLock;
264
265    use crate::protocol::header::empty_header::EmptyHeader;
266
267    use super::compatibility_factory;
268    use super::create_remoting_command;
269    use super::create_request_command;
270    use super::create_response_command;
271    use super::initialize_version;
272    use super::RemotingVersionConflict;
273
274    #[test]
275    fn remoting_version_initialization_is_idempotent() {
276        let version = OnceLock::new();
277
278        assert_eq!(initialize_version(&version, 321), Ok(()));
279        assert_eq!(initialize_version(&version, 321), Ok(()));
280        assert_eq!(version.get(), Some(&321));
281    }
282
283    #[test]
284    fn remoting_version_initialization_rejects_conflicts() {
285        let version = OnceLock::new();
286
287        assert_eq!(initialize_version(&version, 321), Ok(()));
288        assert_eq!(
289            initialize_version(&version, 654),
290            Err(RemotingVersionConflict {
291                initialized: 321,
292                requested: 654,
293            })
294        );
295        assert_eq!(version.get(), Some(&321));
296    }
297
298    #[test]
299    fn compatibility_constructors_share_factory_defaults() {
300        let expected_defaults = compatibility_factory().defaults();
301        let commands = [
302            create_remoting_command(10),
303            create_request_command(11, EmptyHeader {}),
304            create_response_command(),
305        ];
306
307        for command in commands {
308            assert_eq!(command.version(), expected_defaults.version());
309            assert_eq!(command.serialize_type(), expected_defaults.serialize_type());
310        }
311    }
312}
rocketmq_protocol/protocol/remoting_command_compat.rs

rocketmq_protocol/protocol/
remoting_command_compat.rs
