fix: tokio asynchronous task processing error, added tick processing …

…for scheduler.

src/cron.rs

@@ -28,6 +28,7 @@ async fn run_cron<'a>(job_scheduler: Arc<RwLock<job_scheduler::JobScheduler<'a>>
 }
 
 impl<'a> Cron<'a> {
+    /// Create a new cron with the given node pool.
     pub async fn new(mut np: node_pool::NodePool) -> Result<Self, node_pool::Error> {
         np.init().await?;
 
@@ -38,6 +39,7 @@ impl<'a> Cron<'a> {
         })
     }
 
+    /// Start the cron, blocking the current thread.
     pub async fn start(&self) {
         let np = self.node_pool.clone();
 

src/driver/utils.rs

@@ -3,3 +3,13 @@ pub static GLOBAL_KEY_PREFIX: &str = "distributed-scheduler";
 pub fn get_key_prefix(service_name: &str) -> String {
     format!("{}:{}:", GLOBAL_KEY_PREFIX, service_name)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_get_key_prefix() {
+        assert_eq!(get_key_prefix("test"), "distributed-scheduler:test:");
+    }
+}

src/node_pool.rs

@@ -158,3 +158,56 @@ fn equal_ring(a: &[String], b: &[String]) -> bool {
 
     a_sorted == pre_nodes_sorted
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn test_update_hash_ring() {
+        let mut pre_nodes = Vec::new();
+        let state_lock = AtomicBool::new(false);
+        let mut hash = hashring::HashRing::new();
+
+        let nodes = vec!["node1".to_string(), "node2".to_string()];
+
+        update_hash_ring(&mut pre_nodes, &state_lock, &mut hash, &nodes).await.unwrap();
+        assert_eq!(pre_nodes, nodes);
+        assert_eq!(hash.get(&"test"), Some(&"node2".to_string()));
+
+        let nodes = vec!["node1".to_string(), "node2".to_string(), "node3".to_string()];
+
+        update_hash_ring(&mut pre_nodes, &state_lock, &mut hash, &nodes).await.unwrap();
+        assert_eq!(pre_nodes, nodes);
+        assert_eq!(hash.get(&"test"), Some(&"node2".to_string()));
+
+        let nodes = vec!["node1".to_string(), "node3".to_string()];
+
+        update_hash_ring(&mut pre_nodes, &state_lock, &mut hash, &nodes).await.unwrap();
+        assert_eq!(pre_nodes, nodes);
+        assert_eq!(hash.get(&"test"), Some(&"node3".to_string()));
+
+        let nodes = vec!["node1".to_string(), "node3".to_string()];
+
+        update_hash_ring(&mut pre_nodes, &state_lock, &mut hash, &nodes).await.unwrap();
+        assert_eq!(pre_nodes, nodes);
+        assert_eq!(hash.get(&"test"), Some(&"node3".to_string()));
+    }
+
+    #[tokio::test]
+    async fn test_equal_ring() {
+        let a = vec!["node1".to_string(), "node2".to_string()];
+        let b = vec!["node1".to_string(), "node2".to_string()];
+
+        assert_eq!(equal_ring(&a, &b), true);
+
+        let a = vec!["node1".to_string(), "node2".to_string()];
+        let b = vec!["node2".to_string(), "node1".to_string()];
+
+        assert_eq!(equal_ring(&a, &b), true);
+
+        let a = vec!["node1".to_string(), "node2".to_string()];
+        let b = vec!["node1".to_string(), "node3".to_string()];
+        assert_eq!(equal_ring(&a, &b), false);
+    }
+}