rustzmq2 0.1.0

A native async Rust implementation of ZeroMQ
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

#[cfg(test)]
mod test {

    use rustzmq2::__async_rt as async_rt;
    use rustzmq2::prelude::*;
    use rustzmq2::ZmqMessage;

    use std::error::Error;
    use std::time::Duration;

    fn assert_send<T: Send>() {}

    #[test]
    fn split_halves_are_send() {
        assert_send::<rustzmq2::DealerSendHalf>();
        assert_send::<rustzmq2::DealerRecvHalf>();
    }

    #[async_rt::test]
    async fn test_dealer_split_concurrent_send_recv() -> Result<(), Box<dyn Error>> {
        pretty_env_logger::try_init().ok();

        // Use DEALER-to-DEALER (compatible pair, no envelope framing needed)
        let mut server = rustzmq2::DealerSocket::new();
        let endpoint = server.bind("tcp://localhost:0").await?;

        let mut client = rustzmq2::DealerSocket::new();
        client.connect(endpoint.to_string().as_str()).await?;

        // Give connection time to establish
        async_rt::task::sleep(Duration::from_millis(100)).await;

        let (mut send_half, mut recv_half) = client.split();

        let num_messages: u32 = 5;

        // Server echo loop
        let server_task = async_rt::task::spawn(async move {
            for _ in 0..num_messages {
                let msg = server.recv().await.unwrap();
                server.send(msg).await.unwrap();
            }
        });

        // Sender in its own task
        let send_task = async_rt::task::spawn(async move {
            for i in 0..num_messages {
                let msg = ZmqMessage::from(format!("msg-{}", i));
                send_half.send(msg).await.unwrap();
            }
        });

        // Receiver in its own task
        let recv_task = async_rt::task::spawn(async move {
            for _ in 0..num_messages {
                let _reply = recv_half.recv().await.unwrap();
            }
        });

        let timeout = Duration::from_secs(5);
        async_rt::task::timeout(timeout, async {
            send_task.await.unwrap();
            recv_task.await.unwrap();
            server_task.await.unwrap();
        })
        .await
        .expect("test timed out");

        Ok(())
    }
}