serde-libconfigasaurus 0.3.0

//! Example demonstrating deserialization of a stream of libconfig messages
//! over TCP using tokio.
//!
//! A sender task connects and writes 10 libconfig packets, each length-prefixed
//! (4-byte big-endian u32). The listener reads each packet, deserializes it into
//! a Value, and prints the result.

use libconfig::Config;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};

/// Build a libconfig message for a given sequence number.
fn build_message(seq: u32) -> String {
    format!(
        r#"seq = {};
timestamp = {};
sensor = "sensor-{}";
readings = {{
    temperature = {:.1};
    humidity = {:.1};
    pressure = {:.2};
}};
tags = ["environment", "building-{}"];
active = {};"#,
        seq,
        1700000000 + seq,
        seq % 4,
        20.0 + seq as f64 * 0.5,
        45.0 + seq as f64 * 1.2,
        1013.25 + seq as f64 * 0.1,
        seq % 3,
        if seq % 2 == 0 { "true" } else { "false" },
    )
}

/// Send 10 length-prefixed libconfig packets over a TCP connection.
async fn sender(addr: std::net::SocketAddr) {
    let mut stream = TcpStream::connect(addr).await.expect("Failed to connect");

    for seq in 0..10 {
        let msg = build_message(seq);
        let len = msg.len() as u32;

        // Write 4-byte big-endian length prefix, then the payload
        stream
            .write_all(&len.to_be_bytes())
            .await
            .expect("Failed to write length");
        stream
            .write_all(msg.as_bytes())
            .await
            .expect("Failed to write payload");
    }

    stream.shutdown().await.expect("Failed to shutdown");
}

/// Read length-prefixed libconfig packets from a TCP connection and deserialize each one.
async fn receiver(mut stream: TcpStream) {
    let mut len_buf = [0u8; 4];

    loop {
        // Read the 4-byte length prefix
        match stream.read_exact(&mut len_buf).await {
            Ok(_) => {}
            Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => break,
            Err(e) => panic!("Failed to read length: {}", e),
        }

        let len = u32::from_be_bytes(len_buf) as usize;

        // Read the payload
        let mut payload = vec![0u8; len];
        stream
            .read_exact(&mut payload)
            .await
            .expect("Failed to read payload");

        let text = String::from_utf8(payload).expect("Invalid UTF-8");

        // Deserialize the libconfig packet into a Config
        let cfg = Config::from_str(&text).expect("Failed to parse libconfig");

        println!(
            "  packet seq={}, sensor={}, temp={}, active={}, tags={:?}",
            cfg["seq"],
            cfg["sensor"],
            cfg["readings.temperature"],
            cfg["active"],
            cfg["tags"]
                .as_array()
                .unwrap()
                .iter()
                .map(|t| t.as_str().unwrap())
                .collect::<Vec<_>>(),
        );
    }
}

#[tokio::main]
async fn main() {
    println!("=== Streaming Libconfig over TCP ===\n");

    // Bind to a random available port on localhost
    let listener = TcpListener::bind("127.0.0.1:0")
        .await
        .expect("Failed to bind");
    let addr = listener.local_addr().unwrap();
    println!("Listening on {}\n", addr);

    // Spawn the sender task
    let sender_handle = tokio::spawn(sender(addr));

    // Accept one connection and process its packets
    let (stream, peer) = listener.accept().await.expect("Failed to accept");
    println!("Accepted connection from {}\n", peer);

    receiver(stream).await;

    sender_handle.await.expect("Sender task failed");

    println!("\nReceived all 10 packets successfully.");
}