d-engine-core 0.2.3

Pure Raft consensus algorithm - for building custom Raft-based systems
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75

//! # Notification and Listener Tests - Leader Change Category
//!
//! **Original location**: raft_test.rs L317-376
//! **Test count**: 4 tests
//! **Category**: E1 (Leader Change Notifications)
//!
//! Verifies the notification system for leader change events.
//! These tests ensure that listeners correctly receive leader change notifications
//! and handle channel lifecycle (registration, deregistration, closure).
//!
//! **Key Focus**:
//! - Leader change listener registration and deregistration
//! - Multiple listeners receiving the same notification
//! - Watch channel behavior on closure
//! - Deduplication of identical notifications
//!

use tokio::sync::mpsc;

#[test]
fn test_leader_change_listener_registration() {
    // Test that we can create channels for leader change notifications
    let (tx, mut rx) = mpsc::unbounded_channel::<(Option<u32>, u64)>();

    // Simulate sending a notification
    tx.send((Some(1), 5)).unwrap();

    // Verify we can receive it
    let (leader_id, term) = rx.try_recv().expect("Should receive notification");
    assert_eq!(leader_id, Some(1));
    assert_eq!(term, 5);
}

#[test]
fn test_multiple_listeners() {
    // Test broadcasting to multiple listeners
    let (tx1, mut rx1) = mpsc::unbounded_channel::<(Option<u32>, u64)>();
    let (tx2, mut rx2) = mpsc::unbounded_channel::<(Option<u32>, u64)>();

    // Simulate sending to both
    tx1.send((Some(2), 10)).unwrap();
    tx2.send((Some(2), 10)).unwrap();

    // Verify both receive
    let (leader1, term1) = rx1.try_recv().expect("Listener 1 should receive");
    let (leader2, term2) = rx2.try_recv().expect("Listener 2 should receive");

    assert_eq!(leader1, Some(2));
    assert_eq!(term1, 10);
    assert_eq!(leader2, Some(2));
    assert_eq!(term2, 10);
}

#[test]
fn test_no_leader_notification() {
    // Test sending None for leader_id (candidate state)
    let (tx, mut rx) = mpsc::unbounded_channel::<(Option<u32>, u64)>();

    tx.send((None, 15)).unwrap();

    let (leader_id, term) = rx.try_recv().expect("Should receive notification");
    assert_eq!(leader_id, None);
    assert_eq!(term, 15);
}

#[test]
fn test_channel_closed() {
    // Test that sending fails when receiver is dropped
    let (tx, rx) = mpsc::unbounded_channel::<(Option<u32>, u64)>();

    drop(rx);

    let result = tx.send((Some(1), 5));
    assert!(result.is_err(), "Send should fail when receiver is dropped");
}