forked from xline-kv/Xline
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
refactor: add a ClusterMember structure to wrap all_members info
Closes xline-kv#293
- Loading branch information
1 parent
86f4584
commit 5654da3
Showing
17 changed files
with
344 additions
and
215 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,200 @@ | ||
| use std::{ | ||
| collections::{hash_map::DefaultHasher, HashMap}, | ||
| hash::Hasher, | ||
| time::SystemTime, | ||
| }; | ||
|
|
||
| use itertools::Itertools; | ||
|
|
||
| use crate::ServerId; | ||
|
|
||
| /// cluster members information | ||
| #[derive(Debug)] | ||
| pub struct ClusterMember { | ||
| /// current server id | ||
| id: String, | ||
| /// other peers information | ||
| all_members: HashMap<ServerId, String>, | ||
| } | ||
|
|
||
| impl ClusterMember { | ||
| /// Construct a new `ClusterMember` | ||
| /// | ||
| /// # Panics | ||
| /// | ||
| /// panic if `all_members` is empty | ||
| #[inline] | ||
| #[must_use] | ||
| pub fn new(all_members: HashMap<ServerId, String>, id: String) -> Self { | ||
| assert!(!all_members.is_empty()); | ||
| Self { id, all_members } | ||
| } | ||
|
|
||
| /// get server address via server id | ||
| #[must_use] | ||
| #[inline] | ||
| pub fn get_address(&self, id: &str) -> Option<&String> { | ||
| self.all_members.get(id) | ||
| } | ||
|
|
||
| /// get the current server address | ||
| #[must_use] | ||
| #[inline] | ||
| pub fn get_self_url(&self) -> &str { | ||
| self.all_members.get(self.id.as_str()).unwrap_or_else(|| { | ||
| unreachable!( | ||
| "The address of {} not found in all_members {:?}", | ||
| self.id, self.all_members | ||
| ) | ||
| }) | ||
| } | ||
|
|
||
| /// get the current server id | ||
| #[must_use] | ||
| #[inline] | ||
| pub fn get_self_id(&self) -> &ServerId { | ||
| &self.id | ||
| } | ||
|
|
||
| /// get peers id | ||
| #[must_use] | ||
| #[inline] | ||
| pub fn get_peers_id(&self) -> Vec<ServerId> { | ||
| self.all_members | ||
| .keys() | ||
| .filter(|id| id.as_str() != self.id.as_str()) | ||
| .cloned() | ||
| .collect() | ||
| } | ||
|
|
||
| /// get peers urls | ||
| #[must_use] | ||
| #[inline] | ||
| pub fn get_peers_urls(&self) -> Vec<String> { | ||
| self.all_members | ||
| .iter() | ||
| .filter_map(|(id, url)| (id.as_str() != self.id.as_str()).then_some(url)) | ||
| .cloned() | ||
| .collect() | ||
| } | ||
|
|
||
| /// caculate the member id | ||
| #[must_use] | ||
| #[inline] | ||
| pub fn calc_member_id(&self, cluster_name: &str) -> u64 { | ||
| let ts = SystemTime::now() | ||
| .duration_since(SystemTime::UNIX_EPOCH) | ||
| .unwrap_or_else(|e| unreachable!("SystemTime before UNIX EPOCH! {e}")) | ||
| .as_secs(); | ||
| let mut hasher = DefaultHasher::new(); | ||
| hasher.write(self.get_self_url().as_bytes()); | ||
| hasher.write(cluster_name.as_bytes()); | ||
| hasher.write_u64(ts); | ||
| hasher.finish() | ||
| } | ||
|
|
||
| /// caculate the cluster id | ||
| #[must_use] | ||
| #[inline] | ||
| pub fn calc_cluster_id(&self, cluster_name: &str) -> u64 { | ||
| let mut hasher = DefaultHasher::new(); | ||
| let member_urls = self | ||
| .all_members | ||
| .values() | ||
| .sorted() | ||
| .map(String::as_str) | ||
| .collect::<Vec<_>>(); | ||
|
|
||
| for url in member_urls { | ||
| hasher.write(url.as_bytes()); | ||
| } | ||
| hasher.write(cluster_name.as_bytes()); | ||
| hasher.finish() | ||
| } | ||
|
|
||
| /// get peers | ||
| #[must_use] | ||
| #[inline] | ||
| pub fn get_peers(&self) -> HashMap<ServerId, String> { | ||
| self.all_members | ||
| .iter() | ||
| .filter_map(|(id, url)| { | ||
| (id.as_str() != self.id.as_str()).then(|| (id.clone(), url.clone())) | ||
| }) | ||
| .collect() | ||
| } | ||
|
|
||
| /// get all members | ||
| #[must_use] | ||
| #[inline] | ||
| pub fn get_all_members(&self) -> HashMap<ServerId, String> { | ||
| self.all_members.clone() | ||
| } | ||
|
|
||
| /// peers count | ||
| #[must_use] | ||
| #[inline] | ||
| #[allow(clippy::integer_arithmetic)] // It's impossible to overflow | ||
| pub fn peers_len(&self) -> usize { | ||
| self.all_members.len() - 1 | ||
| } | ||
| } | ||
|
|
||
| #[cfg(test)] | ||
| mod tests { | ||
| use super::*; | ||
|
|
||
| #[test] | ||
| fn test_calculate_id() { | ||
| let all_members: HashMap<ServerId, String> = vec![ | ||
| ("S1".to_owned(), "S1".to_owned()), | ||
| ("S2".to_owned(), "S2".to_owned()), | ||
| ("S3".to_owned(), "S3".to_owned()), | ||
| ] | ||
| .into_iter() | ||
| .collect(); | ||
|
|
||
| let node1 = ClusterMember::new(all_members.clone(), "S1".to_owned()); | ||
| let node2 = ClusterMember::new(all_members.clone(), "S2".to_owned()); | ||
| let node3 = ClusterMember::new(all_members, "S3".to_owned()); | ||
|
|
||
| assert_ne!(node1.calc_member_id(""), node2.calc_member_id("")); | ||
| assert_ne!(node1.calc_member_id(""), node3.calc_member_id("")); | ||
| assert_ne!(node3.calc_member_id(""), node2.calc_member_id("")); | ||
|
|
||
| assert_eq!(node1.calc_cluster_id(""), node2.calc_cluster_id("")); | ||
| assert_eq!(node3.calc_cluster_id(""), node2.calc_cluster_id("")); | ||
| } | ||
|
|
||
| #[test] | ||
| fn test_get_peers() { | ||
| let all_members: HashMap<ServerId, String> = vec![ | ||
| ("S1".to_owned(), "S1".to_owned()), | ||
| ("S2".to_owned(), "S2".to_owned()), | ||
| ("S3".to_owned(), "S3".to_owned()), | ||
| ] | ||
| .into_iter() | ||
| .collect(); | ||
|
|
||
| let node1 = ClusterMember::new(all_members, "S1".to_owned()); | ||
| let peers = node1.get_peers(); | ||
| let node1_id = node1.get_self_id(); | ||
| let node1_url = node1.get_self_url(); | ||
| assert!(!peers.contains_key(node1.get_self_id())); | ||
| assert_eq!(peers.len(), 2); | ||
| assert_eq!(node1.peers_len(), peers.len()); | ||
|
|
||
| let peer_urls = node1.get_peers_urls(); | ||
| let peer_ids = node1.get_peers_id(); | ||
| assert_eq!(peer_ids.len(), peer_urls.len()); | ||
|
|
||
| assert!(peer_urls | ||
| .iter() | ||
| .find(|url| url.as_str() == node1_url) | ||
| .is_none()); | ||
| assert!(peer_ids | ||
| .iter() | ||
| .find(|id| id.as_str() == node1_id.as_str()) | ||
| .is_none()); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.