hydra-sync 1.0.0-beta.2

Light-weight zero-copy E2E Single Producer Multiple Consumer network cluster library.
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
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131

use hydra_sync::client::HydraClient;
use hydra_sync::server::HydraServer;
use rand::random;

#[tokio::test]
async fn basic_relay() {
    let (server, server_addr) = HydraServer::bind_default().await.unwrap();
    tokio::spawn(async move { server.run(500).await });

    let session_id = random();
    let session_key = random();

    let mut producer = HydraClient::connect_producer(server_addr, &session_id, session_key)
        .await
        .unwrap();

    tokio::time::sleep(std::time::Duration::from_millis(100)).await;

    let mut consumer = HydraClient::connect_consumer(server_addr, &session_id, session_key)
        .await
        .unwrap();

    tokio::time::sleep(std::time::Duration::from_millis(100)).await;

    let data = b"hello relay";
    producer.broadcast(data).await.unwrap();

    let received = tokio::time::timeout(std::time::Duration::from_secs(5), consumer.recv())
        .await
        .unwrap()
        .unwrap();
    assert_eq!(received, data);
}

#[tokio::test]
async fn concurrent_multi_consumer_relay() {
    let (server, server_addr) = HydraServer::bind_default().await.unwrap();
    tokio::spawn(async move { server.run(500).await });

    let session_id = random();
    let session_key = random();

    let mut producer = HydraClient::connect_producer(server_addr, &session_id, session_key)
        .await
        .unwrap();

    tokio::time::sleep(std::time::Duration::from_millis(100)).await;

    let consumer_count = 8;
    let connect_handles: Vec<_> = (0..consumer_count)
        .map(|_| {
            tokio::spawn(async move {
                HydraClient::connect_consumer(server_addr, &session_id, session_key).await
            })
        })
        .collect();

    let mut consumers = Vec::with_capacity(consumer_count);
    for handle in connect_handles {
        consumers.push(handle.await.unwrap().unwrap());
    }

    tokio::time::sleep(std::time::Duration::from_millis(50)).await;

    let data = b"hello concurrent relay";
    producer.broadcast(data).await.unwrap();

    let recv_handles: Vec<_> = consumers
        .into_iter()
        .map(|mut consumer| {
            tokio::spawn(async move {
                tokio::time::timeout(std::time::Duration::from_secs(5), consumer.recv())
                    .await
                    .unwrap()
                    .unwrap()
                    .to_vec()
            })
        })
        .collect();

    for handle in recv_handles {
        let received = handle.await.unwrap();
        assert_eq!(received, data);
    }
}

#[tokio::test]
async fn continuous_stream_relay() {
    let (server, server_addr) = HydraServer::bind_default().await.unwrap();
    tokio::spawn(async move { server.run(500).await });

    let session_id = random();
    let session_key = random();

    let mut producer = HydraClient::connect_producer(server_addr, &session_id, session_key)
        .await
        .unwrap();

    tokio::time::sleep(std::time::Duration::from_millis(100)).await;

    let mut consumer = HydraClient::connect_consumer(server_addr, &session_id, session_key)
        .await
        .unwrap();

    tokio::time::sleep(std::time::Duration::from_millis(100)).await;

    let message_count = 100;

    let consumer_handle = tokio::spawn(async move {
        for i in 0..message_count {
            let expected_payload = format!("live_stream_packet_{}", i).into_bytes();

            let received = tokio::time::timeout(std::time::Duration::from_secs(2), consumer.recv())
                .await
                .expect("Consumer timed out waiting for a packet")
                .expect("Consumer channel closed unexpectedly");

            assert_eq!(received, expected_payload, "Packet mismatch at index {}", i);
        }
    });

    for i in 0..message_count {
        let payload = format!("live_stream_packet_{}", i).into_bytes();
        producer.broadcast(&payload).await.unwrap();

        // prevents overwhelming the local buffer instantly
        tokio::time::sleep(std::time::Duration::from_millis(2)).await;
    }

    consumer_handle.await.unwrap();
}