ironsbe 0.3.0

High-performance SBE (Simple Binary Encoding) server/client for Rust
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

//! Example SBE client demonstrating basic message sending.
//!
//! Run with: `cargo run --example client`
//!
//! Make sure the server is running first: `cargo run --example server`

use ironsbe_client::builder::ClientBuilder;
use ironsbe_client::builder::ClientEvent;
use ironsbe_core::buffer::{AlignedBuffer, ReadBuffer, WriteBuffer};
use ironsbe_core::header::MessageHeader;
use std::net::SocketAddr;
use std::time::Duration;

/// Creates a simple SBE message with a header and payload.
fn create_message(template_id: u16, payload: &[u8]) -> Vec<u8> {
    let mut buffer = AlignedBuffer::<256>::new();

    // Create and encode the message header
    let header = MessageHeader::new(
        payload.len() as u16, // block_length
        template_id,          // template_id
        1,                    // schema_id
        1,                    // version
    );
    header.encode(&mut buffer, 0);

    // Copy payload after header
    let header_size = MessageHeader::ENCODED_LENGTH;
    buffer.as_mut_slice()[header_size..header_size + payload.len()].copy_from_slice(payload);

    // Return the complete message
    buffer.as_slice()[..header_size + payload.len()].to_vec()
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Initialize tracing for logging
    tracing_subscriber::fmt()
        .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
        .init();

    let addr: SocketAddr = "127.0.0.1:9000".parse()?;

    println!("Connecting to IronSBE server at {}", addr);

    let (mut client, mut handle) = ClientBuilder::with_default_transport(addr)
        .connect_timeout(Duration::from_secs(5))
        .max_reconnect_attempts(3)
        .build();

    // Spawn the client connection task
    let client_task = tokio::spawn(async move {
        if let Err(e) = client.run().await {
            eprintln!("[Client] Error: {:?}", e);
        }
    });

    // Wait for connection
    tokio::time::sleep(Duration::from_millis(100)).await;

    // Send some test messages
    println!("\nSending test messages...\n");

    for i in 1..=5 {
        let payload = format!("Hello from IronSBE client! Message #{}", i);
        let message = create_message(100 + i as u16, payload.as_bytes());

        match handle.send(message) {
            Ok(()) => println!("[Client] Sent message #{}", i),
            Err(e) => eprintln!("[Client] Failed to send message #{}: {:?}", i, e),
        }

        // Small delay between messages
        tokio::time::sleep(Duration::from_millis(100)).await;

        // Check for responses
        while let Some(event) = handle.poll() {
            match event {
                ClientEvent::Connected => {
                    println!("[Client] Connected to server");
                }
                ClientEvent::Disconnected => {
                    println!("[Client] Disconnected from server");
                }
                ClientEvent::Message(data) => {
                    println!("[Client] Received response: {} bytes", data.len());
                    // Try to decode the payload
                    if data.len() > MessageHeader::ENCODED_LENGTH {
                        let payload = &data[MessageHeader::ENCODED_LENGTH..];
                        if let Ok(text) = std::str::from_utf8(payload) {
                            println!("[Client] Response payload: {}", text);
                        }
                    }
                }
                ClientEvent::Error(e) => {
                    eprintln!("[Client] Error: {}", e);
                }
            }
        }
    }

    // Wait a bit for final responses
    tokio::time::sleep(Duration::from_millis(500)).await;

    // Drain remaining events
    while let Some(event) = handle.poll() {
        if let ClientEvent::Message(data) = event {
            println!("[Client] Received response: {} bytes", data.len());
            if data.len() > MessageHeader::ENCODED_LENGTH {
                let payload = &data[MessageHeader::ENCODED_LENGTH..];
                if let Ok(text) = std::str::from_utf8(payload) {
                    println!("[Client] Response payload: {}", text);
                }
            }
        }
    }

    // Disconnect
    println!("\nDisconnecting...");
    handle.disconnect();

    // Wait for client task to finish
    tokio::time::timeout(Duration::from_secs(2), client_task)
        .await
        .ok();

    println!("Client stopped");
    Ok(())
}