ProtoBuilder

High-level message protocol builder for async Rust.

ProtoBuilder simplifies building type-safe communication protocols over async streams by combining:

• Enum-based message types - Define your protocol messages as Rust enums with compile-time type safety
• Automatic serialization - Uses RON (Rusty Object Notation) for human-readable, text-based message format
• Binary framing layer - Configurable framing strategies (length-prefix or chunked) for message boundary detection
• Async/Tokio native - Built from the ground up for async Rust

Features

• Type-safe protocol definitions using Rust enums
• Length-prefix framing (u16 or u32) for simple message boundaries
• Chunked framing for large messages (32KB+)
• Split protocol support for concurrent read/write
• Clean error handling with custom error types

Installation

Add this to your Cargo.toml:

[dependencies]
protobuilder = "0.1.0"
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }

Quick Start

use protobuilder::{Protocol, LengthPrefix, Result};
use serde::{Serialize, Deserialize};

#[derive(Serialize, Deserialize, Debug, Clone)]
enum Request {
    Ping,
    Message(String),
}

#[derive(Serialize, Deserialize, Debug, Clone)]
enum Response {
    Pong,
    Reply(String),
}

#[tokio::main]
async fn main() -> Result<()> {
    // Connect to a server
    let stream = tokio::net::TcpStream::connect("127.0.0.1:9999").await?;

    // Build the protocol with length-prefix framing
    let mut proto = Protocol::<_, Request, Response, _>::builder()
        .framing(LengthPrefix::u32())
        .build(stream)?;

    // Send and receive messages
    proto.send(Request::Ping).await?;
    let response = proto.recv().await?;
    println!("Received: {:?}", response);

    Ok(())
}

Framing Strategies

Length Prefix

Simple framing using a length header before each message:

use protobuilder::LengthPrefix;

// 16-bit length prefix (max 64KB messages)
let framing = LengthPrefix::u16();

// 32-bit length prefix (max 4GB messages)
let framing = LengthPrefix::u32();

// Custom max size limit
let framing = LengthPrefix::u32().with_max_size(1024 * 1024); // 1MB

Chunked Framing

For large messages that need to be split across multiple chunks:

use protobuilder::ChunkedFraming;

// 8KB chunks
let framing = ChunkedFraming::new(8 * 1024);

// With custom max message size
let framing = ChunkedFraming::new(32 * 1024)
    .with_max_message_size(10 * 1024 * 1024); // 10MB

Split Protocol

For concurrent read/write operations, split the protocol into halves:

# use protobuilder::{Protocol, LengthPrefix, Result};
# use serde::{Serialize, Deserialize};
# #[derive(Serialize, Deserialize)]
# enum Request { Ping }
# #[derive(Serialize, Deserialize)]
# enum Response { Pong }
# async fn example() -> Result<()> {
# let stream = tokio::net::TcpStream::connect("127.0.0.1:9999").await?;
let proto = Protocol::<_, Request, Response, _>::builder()
    .framing(LengthPrefix::u32())
    .build(stream)?;

let mut split_proto = proto.split();

// Now you can use split_proto in concurrent tasks
tokio::spawn(async move {
    split_proto.send(Request::Ping).await.unwrap();
});
# Ok(())
# }

Complete Example

See examples/basic.rs for a working client-server example:

use protobuilder::{Protocol, LengthPrefix, Result};
use serde::{Serialize, Deserialize};
use tokio::net::{TcpListener, TcpStream};

#[derive(Serialize, Deserialize, Debug, Clone)]
enum ClientPacket {
    Ping,
    Message(String),
}

#[derive(Serialize, Deserialize, Debug, Clone)]
enum ServerPacket {
    Pong,
    Reply(String),
}

// Server
async fn server() -> Result<()> {
    let listener = TcpListener::bind("127.0.0.1:9999").await?;
    
    loop {
        let (stream, _) = listener.accept().await?;
        tokio::spawn(async move {
            let mut proto = Protocol::<_, ServerPacket, ClientPacket, _>::builder()
                .framing(LengthPrefix::u32())
                .build(stream)
                .unwrap();

            while let Ok(packet) = proto.recv().await {
                match packet {
                    ClientPacket::Ping => {
                        let _ = proto.send(ServerPacket::Pong).await;
                    }
                    ClientPacket::Message(text) => {
                        let _ = proto.send(ServerPacket::Reply(format!("Echo: {}", text))).await;
                    }
                }
            }
        });
    }
}

Examples

Run the basic example:

cargo run --example basic

Run the large data example (demonstrates chunked framing):

cargo run --example large_data

API Overview

Protocol Builder

// Generic form:
Protocol::<Stream, SendType, RecvType, FramingType>::builder()
    .framing(framing_strategy)
    .build(stream)

// See the Quick Start example above for a complete working example

Protocol Methods

• send(packet) - Send a message
• recv() - Receive a message
• split() - Split into read/write halves
• into_inner() - Extract the underlying stream
• get_ref() / get_mut() - Access the underlying stream

SplitProtocol Methods

• send(packet) - Send a message (from the writer half)
• recv() - Receive a message (from the reader half)
• into_halves() - Extract the reader, writer, and framing components

Error Types

pub enum ProtocolError {
    Io(std::io::Error),
    Framing(String),
    Serialization(String),
    Deserialization(String),
    TooLarge(usize),
}

pub type Result<T> = std::result::Result<T, ProtocolError>;

How It Works

1. Serialization: Your enum messages are serialized to RON text (e.g., Message("Hello") → "Message(\"Hello\")")
2. Framing: The text payload is wrapped with a binary framing layer for message boundary detection:
   • Length Prefix: Prepends a u16/u32 length header (big-endian)
   • Chunked: Splits large payloads into chunks with message IDs and continuation flags
3. Transmission: The framed binary data is written to/read from the async stream

Note: The wire format is binary (due to the framing layer), but the message payload itself is human-readable RON text. For pure binary serialization (e.g., bincode, protobuf), this crate is not suitable.

License

Licensed under the Apache License, Version 2.0. See LICENSE for details.