monocoque-rs 0.1.1

High-performance ZeroMQ-compatible messaging runtime built on io_uring
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

//! Interoperability test: Monocoque REQ ↔ libzmq REP
//!
//! This example demonstrates that Monocoque REQ can communicate with libzmq REP.
//!
//! Run this example:
//! ```bash
//! cargo run --example interop_req_libzmq --features zmq
//! ```

use bytes::Bytes;
use monocoque_zmtp::req::ReqSocket;
use std::thread;
use std::time::Duration;
use tracing::info;

fn main() {
    info!("=== Monocoque REQ ↔ libzmq REP Interop Test ===\n");

    // Spawn libzmq REP server in background thread
    let server_handle = thread::spawn(|| {
        let ctx = zmq::Context::new();
        let rep = ctx.socket(zmq::REP).unwrap();
        rep.bind("tcp://127.0.0.1:5561").unwrap();
        info!("[libzmq REP] Listening on tcp://127.0.0.1:5561");

        // Receive request
        let request = rep.recv_bytes(0).unwrap();
        info!(
            "[libzmq REP] Received request: {:?}",
            String::from_utf8_lossy(&request)
        );

        // Send reply
        rep.send("Reply from libzmq REP", 0).unwrap();
        info!("[libzmq REP] Sent reply\n");

        // Second round
        let request = rep.recv_bytes(0).unwrap();
        info!(
            "[libzmq REP] Received second request: {:?}",
            String::from_utf8_lossy(&request)
        );

        rep.send("Second reply from libzmq", 0).unwrap();
        info!("[libzmq REP] Sent second reply\n");

        // Keep server alive
        thread::sleep(Duration::from_millis(100));
    });

    // Give server time to bind
    thread::sleep(Duration::from_millis(50));

    // Run Monocoque REQ client
    compio::runtime::Runtime::new().unwrap().block_on(async {
        info!("[Monocoque REQ] Connecting to tcp://127.0.0.1:5561");

        let stream = compio::net::TcpStream::connect("127.0.0.1:5561")
            .await
            .expect("Failed to connect");

        let mut socket = ReqSocket::new(stream).await.unwrap();
        info!("[Monocoque REQ] Connected\n");

        // First request-reply cycle
        info!("[Monocoque REQ] Sending first request");
        socket
            .send(vec![Bytes::from_static(b"Request from Monocoque REQ")])
            .await
            .expect("Failed to send");

        let response = socket
            .recv()
            .await
            .expect("Failed to receive")
            .expect("Connection closed");
        info!("[Monocoque REQ] Received response:");
        {
            let msg = response;
            for (i, frame) in msg.iter().enumerate() {
                info!("  Frame {}: {:?}", i, String::from_utf8_lossy(frame));
            }
        }

        // Second request-reply cycle
        info!("\n[Monocoque REQ] Sending second request");
        socket
            .send(vec![Bytes::from_static(b"Second request")])
            .await
            .expect("Failed to send second request");

        let response = socket
            .recv()
            .await
            .expect("Failed to receive second reply")
            .expect("Connection closed");
        info!("[Monocoque REQ] Received second response:");
        {
            let msg = response;
            for (i, frame) in msg.iter().enumerate() {
                info!("  Frame {}: {:?}", i, String::from_utf8_lossy(frame));
            }
        }

        drop(socket);
    });

    server_handle.join().unwrap();

    info!("\n✅ REQ interop test completed successfully!");
}