mxsm · rocketmq-rust-bot · Jun 22, 2025 · Jun 22, 2025 · coderabbitai · Jun 22, 2025
diff --git a/rocketmq-store/src/ha.rs b/rocketmq-store/src/ha.rs
@@ -20,6 +20,8 @@ pub(crate) mod default_ha_client;
 mod default_ha_connection;
 pub(crate) mod default_ha_service;
 pub(crate) mod flow_monitor;
+pub(crate) mod general_ha_client;
+pub(crate) mod general_ha_connection;
 pub(crate) mod general_ha_service;
 pub(crate) mod ha_client;
 pub(crate) mod ha_connection;

diff --git a/rocketmq-store/src/ha/auto_switch.rs b/rocketmq-store/src/ha/auto_switch.rs
@@ -14,4 +14,6 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+pub(crate) mod auto_switch_ha_client;
+pub(crate) mod auto_switch_ha_connection;
 pub(crate) mod auto_switch_ha_service;
diff --git a/rocketmq-store/src/ha/auto_switch/auto_switch_ha_client.rs b/rocketmq-store/src/ha/auto_switch/auto_switch_ha_client.rs
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+use crate::ha::ha_client::HAClient;
+use crate::ha::ha_connection_state::HAConnectionState;
+
+pub struct AutoSwitchHAClient;
-pub struct AutoSwitchHAClient;
+use crate::ha_client::{HAClient, HAConnectionState};
+use std::sync::{Arc, Mutex};
+
+pub struct AutoSwitchHAClient {
+    master_address: Arc<Mutex<String>>,
+    ha_master_address: Arc<Mutex<String>>,
+    last_read_timestamp: Arc<Mutex<i64>>,
+    last_write_timestamp: Arc<Mutex<i64>>,
+    current_state: Arc<Mutex<HAConnectionState>>,
+    transferred_bytes: Arc<Mutex<i64>>,
+}
+
+impl AutoSwitchHAClient {
+    pub fn new() -> Self {
+        Self {
+            master_address: Arc::new(Mutex::new(String::new())),
+            ha_master_address: Arc::new(Mutex::new(String::new())),
+            last_read_timestamp: Arc::new(Mutex::new(0)),
+            last_write_timestamp: Arc::new(Mutex::new(0)),
+            current_state: Arc::new(Mutex::new(HAConnectionState::default())),
+            transferred_bytes: Arc::new(Mutex::new(0)),
+        }
+    }
+}
-pub struct AutoSwitchHAClient;
+use crate::ha_client::{HAClient, HAConnectionState};
+use std::sync::{Arc, Mutex};
+
+pub struct AutoSwitchHAClient {
+    master_address: Arc<Mutex<String>>,
+    ha_master_address: Arc<Mutex<String>>,
+    last_read_timestamp: Arc<Mutex<i64>>,
+    last_write_timestamp: Arc<Mutex<i64>>,
+    current_state: Arc<Mutex<HAConnectionState>>,
+    transferred_bytes: Arc<Mutex<i64>>,
+}
+
+impl AutoSwitchHAClient {
+    pub fn new() -> Self {
+        Self {
+            master_address: Arc::new(Mutex::new(String::new())),
+            ha_master_address: Arc::new(Mutex::new(String::new())),
+            last_read_timestamp: Arc::new(Mutex::new(0)),
+            last_write_timestamp: Arc::new(Mutex::new(0)),
+            current_state: Arc::new(Mutex::new(HAConnectionState::default())),
+            transferred_bytes: Arc::new(Mutex::new(0)),
+        }
+    }
+}
+
+impl HAClient for AutoSwitchHAClient {
+    async fn start(&self) {
+        todo!()
+    }
+
+    async fn shutdown(&self) {
+        todo!()
-        todo!()
-    }
-
-    async fn shutdown(&self) {
-        todo!()
+        // Initialize the client, e.g., set up connections or resources.
+        println!("AutoSwitchHAClient started.");
+    }
+
+    async fn shutdown(&self) {
+        // Release resources and close connections.
+        println!("AutoSwitchHAClient shut down.");
-        todo!()
-    }
-
-    async fn shutdown(&self) {
-        todo!()
+        // Initialize the client, e.g., set up connections or resources.
+        println!("AutoSwitchHAClient started.");
+    }
+
+    async fn shutdown(&self) {
+        // Release resources and close connections.
+        println!("AutoSwitchHAClient shut down.");
+    }
+
+    async fn wakeup(&self) {
+        todo!()
+    }
+
+    async fn update_master_address(&self, new_address: &str) {
+        todo!()
+    }
+
+    async fn update_ha_master_address(&self, new_address: &str) {
+        todo!()
+    }
+
+    fn get_master_address(&self) -> String {
+        todo!()
+    }
+
+    fn get_ha_master_address(&self) -> String {
+        todo!()
+    }
+
+    fn get_last_read_timestamp(&self) -> i64 {
+        todo!()
+    }
+
+    fn get_last_write_timestamp(&self) -> i64 {
+        todo!()
+    }
+
+    fn get_current_state(&self) -> HAConnectionState {
+        todo!()
+    }
+
+    fn change_current_state(&self, ha_connection_state: HAConnectionState) {
+        todo!()
+    }
+
+    async fn close_master(&self) {
+        todo!()
+    }
+
+    fn get_transferred_byte_in_second(&self) -> i64 {
+        todo!()
+    }
+}
diff --git a/rocketmq-store/src/ha/auto_switch/auto_switch_ha_connection.rs b/rocketmq-store/src/ha/auto_switch/auto_switch_ha_connection.rs
@@ -0,0 +1,17 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+pub struct AutoSwitchHAConnection;
diff --git a/rocketmq-store/src/ha/default_ha_service.rs b/rocketmq-store/src/ha/default_ha_service.rs
@@ -36,6 +36,7 @@
 use std::sync::Arc;
 
 use rocketmq_remoting::protocol::body::ha_runtime_info::HARuntimeInfo;
+use rocketmq_rust::ArcMut;
 use tracing::error;
 
 use crate::ha::ha_client::HAClient;
@@ -61,6 +62,11 @@
         // The actual implementation would depend on the specific requirements of the HA service.
         unimplemented!(" notify_transfer_some method is not implemented");
     }
+
+    pub(crate) fn init(&mut self, message_store: ArcMut<LocalFileMessageStore>) -> HAResult<()> {
+        // Initialize the DefaultHAService with the provided message store.
+        Ok(())
+    }
 }
 
 impl HAService for DefaultHAService {

diff --git a/rocketmq-store/src/ha/general_ha_client.rs b/rocketmq-store/src/ha/general_ha_client.rs
@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+use crate::ha::auto_switch::auto_switch_ha_client::AutoSwitchHAClient;
+use crate::ha::default_ha_client::DefaultHAClient;
+
+pub struct GeneralHAClient {
+    default_ha_service: Option<DefaultHAClient>,
+    auto_switch_ha_service: Option<AutoSwitchHAClient>,
-    default_ha_service: Option<DefaultHAClient>,
-    auto_switch_ha_service: Option<AutoSwitchHAClient>,
+pub struct GeneralHAClient {
+    default_ha_client: Option<DefaultHAClient>,
+    auto_switch_ha_client: Option<AutoSwitchHAClient>,
+}
-    default_ha_service: Option<DefaultHAClient>,
-    auto_switch_ha_service: Option<AutoSwitchHAClient>,
+pub struct GeneralHAClient {
+    default_ha_client: Option<DefaultHAClient>,
+    auto_switch_ha_client: Option<AutoSwitchHAClient>,
+}
+}
-pub struct GeneralHAClient {
-    default_ha_service: Option<DefaultHAClient>,
-    auto_switch_ha_service: Option<AutoSwitchHAClient>,
-}
+pub struct GeneralHAClient {
+    default_ha_client: Option<DefaultHAClient>,
+    auto_switch_ha_client: Option<AutoSwitchHAClient>,
+}
+
+impl GeneralHAClient {
+    /// Creates an empty client with no HA backends configured.
+    pub fn new() -> Self {
+        Self {
+            default_ha_client: None,
+            auto_switch_ha_client: None,
+        }
+    }
+
+    /// Creates a client using the given `DefaultHAClient`.
+    pub fn with_default(client: DefaultHAClient) -> Self {
+        Self {
+            default_ha_client: Some(client),
+            auto_switch_ha_client: None,
+        }
+    }
+
+    /// Creates a client using the given `AutoSwitchHAClient`.
+    pub fn with_auto_switch(client: AutoSwitchHAClient) -> Self {
+        Self {
+            default_ha_client: None,
+            auto_switch_ha_client: Some(client),
+        }
+    }
+
+    /// Returns a reference to the default HA client, if configured.
+    pub fn get_default_client(&self) -> Option<&DefaultHAClient> {
+        self.default_ha_client.as_ref()
+    }
+
+    /// Returns a reference to the auto‐switch HA client, if configured.
+    pub fn get_auto_switch_client(&self) -> Option<&AutoSwitchHAClient> {
+        self.auto_switch_ha_client.as_ref()
+    }
+}
-pub struct GeneralHAClient {
-    default_ha_service: Option<DefaultHAClient>,
-    auto_switch_ha_service: Option<AutoSwitchHAClient>,
-}
+pub struct GeneralHAClient {
+    default_ha_client: Option<DefaultHAClient>,
+    auto_switch_ha_client: Option<AutoSwitchHAClient>,
+}
+
+impl GeneralHAClient {
+    /// Creates an empty client with no HA backends configured.
+    pub fn new() -> Self {
+        Self {
+            default_ha_client: None,
+            auto_switch_ha_client: None,
+        }
+    }
+
+    /// Creates a client using the given `DefaultHAClient`.
+    pub fn with_default(client: DefaultHAClient) -> Self {
+        Self {
+            default_ha_client: Some(client),
+            auto_switch_ha_client: None,
+        }
+    }
+
+    /// Creates a client using the given `AutoSwitchHAClient`.
+    pub fn with_auto_switch(client: AutoSwitchHAClient) -> Self {
+        Self {
+            default_ha_client: None,
+            auto_switch_ha_client: Some(client),
+        }
+    }
+
+    /// Returns a reference to the default HA client, if configured.
+    pub fn get_default_client(&self) -> Option<&DefaultHAClient> {
+        self.default_ha_client.as_ref()
+    }
+
+    /// Returns a reference to the auto‐switch HA client, if configured.
+    pub fn get_auto_switch_client(&self) -> Option<&AutoSwitchHAClient> {
+        self.auto_switch_ha_client.as_ref()
+    }
+}
diff --git a/rocketmq-store/src/ha/general_ha_connection.rs b/rocketmq-store/src/ha/general_ha_connection.rs
@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+use crate::ha::auto_switch::auto_switch_ha_connection::AutoSwitchHAConnection;
+use crate::ha::default_ha_connection::DefaultHAConnection;
+
+pub struct GeneralHAConnection {
+    default_ha_connection: Option<DefaultHAConnection>,
+    auto_switch_ha_connection: Option<AutoSwitchHAConnection>,
+}
-pub struct GeneralHAConnection {
-    default_ha_connection: Option<DefaultHAConnection>,
-    auto_switch_ha_connection: Option<AutoSwitchHAConnection>,
-}
+pub struct GeneralHAConnection {
+    default_ha_connection: Option<DefaultHAConnection>,
+    auto_switch_ha_connection: Option<AutoSwitchHAConnection>,
+}
+
+impl GeneralHAConnection {
+    pub fn new() -> Self {
+        Self {
+            default_ha_connection: None,
+            auto_switch_ha_connection: None,
+        }
+    }
+
+    pub fn with_default(connection: DefaultHAConnection) -> Self {
+        Self {
+            default_ha_connection: Some(connection),
+            auto_switch_ha_connection: None,
+        }
+    }
+
+    pub fn with_auto_switch(connection: AutoSwitchHAConnection) -> Self {
+        Self {
+            default_ha_connection: None,
+            auto_switch_ha_connection: Some(connection),
+        }
+    }
+
+    pub fn get_default_connection(&self) -> Option<&DefaultHAConnection> {
+        self.default_ha_connection.as_ref()
+    }
+
+    pub fn get_auto_switch_connection(&self) -> Option<&AutoSwitchHAConnection> {
+        self.auto_switch_ha_connection.as_ref()
+    }
+}
-pub struct GeneralHAConnection {
-    default_ha_connection: Option<DefaultHAConnection>,
-    auto_switch_ha_connection: Option<AutoSwitchHAConnection>,
-}
+pub struct GeneralHAConnection {
+    default_ha_connection: Option<DefaultHAConnection>,
+    auto_switch_ha_connection: Option<AutoSwitchHAConnection>,
+}
+
+impl GeneralHAConnection {
+    pub fn new() -> Self {
+        Self {
+            default_ha_connection: None,
+            auto_switch_ha_connection: None,
+        }
+    }
+
+    pub fn with_default(connection: DefaultHAConnection) -> Self {
+        Self {
+            default_ha_connection: Some(connection),
+            auto_switch_ha_connection: None,
+        }
+    }
+
+    pub fn with_auto_switch(connection: AutoSwitchHAConnection) -> Self {
+        Self {
+            default_ha_connection: None,
+            auto_switch_ha_connection: Some(connection),
+        }
+    }
+
+    pub fn get_default_connection(&self) -> Option<&DefaultHAConnection> {
+        self.default_ha_connection.as_ref()
+    }
+
+    pub fn get_auto_switch_connection(&self) -> Option<&AutoSwitchHAConnection> {
+        self.auto_switch_ha_connection.as_ref()
+    }
+}
diff --git a/rocketmq-store/src/ha/general_ha_service.rs b/rocketmq-store/src/ha/general_ha_service.rs
@@ -49,7 +49,9 @@
         {
             self.auto_switch_ha_service = Some(AutoSwitchHAService)
         } else {
-            self.default_ha_service = Some(DefaultHAService::default())
+            let mut default_ha_service = DefaultHAService::default();
+            default_ha_service.init(message_store)?;
+            self.default_ha_service = Some(default_ha_service);
         }
         Ok(())
     }