use bytes::{Bytes, BytesMut};
 
fn main() {
    // BytesMut: mutable, growable byte buffer
    let mut buf = BytesMut::with_capacity(1024);
    buf.extend_from_slice(b"Hello, World!");
    
    // Bytes: immutable, reference-counted byte buffer
    // Cannot be mutated, but can be cheaply cloned (reference count increment)
    
    // BytesMut is for building/modifying
    // Bytes is for sharing/reading
    
    println!("BytesMut: {:?}", buf);
    
    // Common pattern: build in BytesMut, convert to Bytes for sharing
    let bytes: Bytes = buf.freeze();
    println!("Bytes: {:?}", bytes);
}

use bytes::{Bytes, BytesMut};
 
fn main() {
    let mut buf = BytesMut::new();
    buf.extend_from_slice(b"Hello, World!");
    
    // freeze() converts BytesMut -> Bytes
    // This is O(1) - no copying
    // The underlying allocation is kept, just the type changes
    
    let bytes: Bytes = buf.freeze();
    
    // 'buf' is now consumed, cannot be used
    // 'bytes' is immutable and reference-counted
    
    // We can clone bytes cheaply (reference count increment)
    let bytes2 = bytes.clone();
    
    // Both bytes and bytes2 share the same allocation
    // No copying of "Hello, World!" occurred
    
    println!("Original: {:?}", bytes);
    println!("Cloned: {:?}", bytes2);
}

use bytes::BytesMut;
 
fn main() {
    let mut buf = BytesMut::from("Hello, World!");
    
    // split_to(n) splits at position n, returning the first part
    // split_off(n) splits at position n, returning the second part
    
    // split_to: returns first n bytes, self becomes remainder
    let mut buf1 = BytesMut::from("Hello, World!");
    let first = buf1.split_to(5);
    // first = "Hello"
    // buf1 = ", World!"
    
    println!("First part: {:?}", first);
    println!("Remainder: {:?}", buf1);
    
    // split_off: self becomes first part, returns remainder
    let mut buf2 = BytesMut::from("Hello, World!");
    let second = buf2.split_off(7);
    // buf2 = "Hello, "
    // second = "World!"
    
    println!("First part: {:?}", buf2);
    println!("Second part: {:?}", second);
    
    // After split, both parts are independent BytesMut
    // They can be mutated separately, freed separately
}

use bytes::{Bytes, BytesMut};
 
fn main() {
    // BytesMut::split_to / split_off
    // - Input: BytesMut
    // - Output: (BytesMut, BytesMut) - split into two
    // - Both parts are mutable
    // - Independent ownership
    
    let mut buf = BytesMut::from("Hello World");
    let part1 = buf.split_to(5);
    // part1: BytesMut (mutable)
    // buf: BytesMut (mutable)
    // Each can be modified independently
    
    // Bytes::split
    // - Input: Bytes
    // - Output: (Bytes, Bytes) - split into two
    // - Both parts are immutable
    // - Shared reference-counted ownership
    
    let bytes = Bytes::from("Hello World");
    let (part1, part2) = bytes.split_at(5);
    // part1: Bytes (immutable)
    // part2: Bytes (immutable)
    // Both reference the same underlying allocation
    
    // BytesMut::freeze then Bytes::split
    let mut buf = BytesMut::from("Hello World");
    let bytes = buf.freeze();
    let (part1, part2) = bytes.split_at(5);
    // part1 and part2 share allocation, immutable
}

use bytes::{Bytes, BytesMut};
 
fn main() {
    // Scenario 1: freeze - loses mutability
    let mut buf = BytesMut::from("Hello World");
    let bytes = buf.freeze();
    
    // bytes is now immutable Bytes
    // Cannot modify it anymore
    
    // This is good when you want to share data read-only
    // and guarantee no more mutations
    
    // Scenario 2: split - preserves mutability
    let mut buf = BytesMut::from("Hello World");
    let first = buf.split_to(5);
    
    // Both first and buf are still BytesMut
    // Can continue mutating both
    let mut first = first; // mut binding
    first.extend_from_slice(b"!!!");
    
    // buf (remainder) is also mutable
    buf.extend_from_slice(b"!!!");
    
    println!("First (mutated): {:?}", first);
    println!("Remainder (mutated): {:?}", buf);
}

use bytes::{Bytes, BytesMut};
 
fn main() {
    // freeze + split: shared allocation
    let mut buf = BytesMut::from("Hello World");
    let bytes = buf.freeze();
    let (part1, part2) = bytes.split_at(5);
    
    // part1 and part2 share the same underlying allocation
    // Cloning either increments the reference count
    let part1_clone = part1.clone(); // reference count++
    
    // Memory layout:
    // [H][e][l][l][o][ ][W][o][r][l][d]
    //  ^part1^^^^      ^part2^^^^^^^^
    // Same allocation, different views
    
    // split_to: independent allocations
    let mut buf = BytesMut::from("Hello World");
    let first = buf.split_to(5);
    
    // first and buf are INDEPENDENT
    // They do NOT share allocation
    // Dropping first doesn't affect buf
    // They can be freed independently
    
    // Memory layout:
    // first: [H][e][l][l][o] (separate allocation)
    // buf: [ ][W][o][r][l][d] (separate allocation)
}

use bytes::{Bytes, BytesMut};
 
// Common pattern: receive buffer, parse header, freeze body for later
 
fn parse_message(mut buf: BytesMut) -> Result<Bytes, &'static str> {
    // Assume header is first 4 bytes
    if buf.len() < 4 {
        return Err("Buffer too small");
    }
    
    // Split off header
    let header = buf.split_to(4);
    
    // Parse header to determine message type
    let msg_type = header[0];
    let length = u32::from_be_bytes([
        header[0], header[1], header[2], header[3]
    ]) as usize;
    
    // Validate length
    if buf.len() < length {
        return Err("Incomplete message");
    }
    
    // Option A: freeze the remaining buffer
    let body = buf.freeze();
    Ok(body)
    
    // Option B: split off exact length
    // let body = buf.split_to(length);
    // let body_bytes = body.freeze();
    // Ok(body_bytes)
}

use bytes::{Bytes, BytesMut};
 
fn zero_copy_sharing() {
    // Build a large buffer
    let mut buf = BytesMut::with_capacity(1024 * 1024); // 1 MB
    for i in 0..100_000 {
        buf.extend_from_slice(format!("Item {}\n", i).as_bytes());
    }
    
    // Scenario: Share this data with multiple consumers
    // Without freeze: would need to clone the BytesMut (copy!)
    // With freeze: convert to Bytes, then clone is O(1)
    
    let bytes = buf.freeze();
    
    // Share with multiple consumers - no copying
    let consumer1 = bytes.clone();
    let consumer2 = bytes.clone();
    let consumer3 = bytes.slice(0..100); // reference to subset
    
    // All share the same 1 MB allocation
    // Total memory: ~1 MB + reference count overhead
    // Without freeze: would be ~3 MB (3 copies)
    
    println!("Original len: {}", bytes.len());
    println!("Consumer1 len: {}", consumer1.len());
    println!("Consumer2 len: {}", consumer2.len());
    println!("Consumer3 len: {}", consumer3.len());
}

use bytes::BytesMut;
 
fn buffer_reuse() {
    // split_to enables efficient buffer reuse
    
    let mut buf = BytesMut::with_capacity(1024);
    
    // Receive data into buffer (simulated)
    buf.extend_from_slice(b"Request 1 data");
    
    // Process by splitting off the relevant portion
    let request1 = buf.split_to(15);
    process_request(request1);
    
    // buf now has remaining capacity, can be reused
    buf.extend_from_slice(b"Request 2 data");
    
    // Split again for next request
    let request2 = buf.split_to(15);
    process_request(request2);
    
    // This pattern is common in network servers:
    // - Receive into large buffer
    // - Split off complete messages
    // - Reuse buffer for next messages
    
    // freeze would NOT work here:
    // - After freeze, you lose the mutable buffer
    // - Cannot receive more data into it
}
 
fn process_request(data: BytesMut) {
    println!("Processing: {:?}", data);
}

use bytes::{Bytes, BytesMut};
 
fn main() {
    // Original buffer
    let mut buf = BytesMut::from("ABCDEFGH");
    
    // After freeze:
    let bytes = buf.freeze();
    // Memory: [A][B][C][D][E][F][G][H] <- Bytes reference-counted
    // The BytesMut is consumed, but the allocation lives on
    
    // If we clone:
    let bytes2 = bytes.clone();
    // Still one allocation: [A][B][C][D][E][F][G][H]
    // bytes and bytes2 both point to it, reference count = 2
    
    // After split_at:
    let (part1, part2) = bytes.split_at(4);
    // Still one allocation: [A][B][C][D][E][F][G][H]
    // part1 = slice(0..4), part2 = slice(4..8)
    // Both reference-counted, reference count managed
    
    // Contrast with split_to on BytesMut:
    let mut buf2 = BytesMut::from("ABCDEFGH");
    let first = buf2.split_to(4);
    
    // This may or may not copy depending on capacity/offset
    // If the original buffer was "exactly sized" and at offset 0:
    // - first gets a new allocation [A][B][C][D]
    // - buf2 gets a new allocation [E][F][G][H]
    // But if there's capacity and offset allows:
    // - Might just adjust pointers (cheap)
    
    println!("first: {:?}", first);
    println!("buf2: {:?}", buf2);
}

use bytes::{Bytes, BytesMut};
use std::time::Instant;
 
fn performance_comparison() {
    // freeze: O(1) always
    // - Just converts type, keeps same allocation
    // - Reference count initialized to 1
    
    let start = Instant::now();
    for _ in 0..1_000_000 {
        let mut buf = BytesMut::from("Hello World");
        let _bytes = buf.freeze();
    }
    println!("1M freezes: {:?}", start.elapsed());
    
    // split_to/split_off: O(1) typically
    // - Adjusts internal pointers
    // - May allocate if capacity needs adjustment
    
    let start = Instant::now();
    for _ in 0..1_000_000 {
        let mut buf = BytesMut::from("Hello World");
        let _first = buf.split_to(5);
    }
    println!("1M split_tos: {:?}", start.elapsed());
    
    // Bytes::split_at: O(1) always
    // - Creates two new references to same allocation
    // - No allocation, no copy
    
    let start = Instant::now();
    for _ in 0..1_000_000 {
        let bytes = Bytes::from("Hello World");
        let (_, _) = bytes.split_at(5);
    }
    println!("1M split_ats: {:?}", start.elapsed());
}

use bytes::BytesMut;
 
fn efficient_splitting() {
    // When you know you'll be splitting, reserve capacity strategically
    
    // Bad pattern: many small allocations
    let mut bufs = Vec::new();
    for i in 0..100 {
        let mut buf = BytesMut::from(format!("Message {}", i).as_bytes());
        bufs.push(buf.split_to(buf.len())); // May allocate
    }
    
    // Better: allocate once, split off pieces
    let mut big_buf = BytesMut::with_capacity(1024);
    for i in 0..100 {
        big_buf.extend_from_slice(format!("Message {}\n", i).as_bytes());
    }
    
    let mut messages = Vec::new();
    while !big_buf.is_empty() {
        // Find newline
        let pos = big_buf.iter().position(|&b| b == b'\n').unwrap_or(big_buf.len());
        let msg = big_buf.split_to(pos + 1);
        messages.push(msg);
    }
    
    // Or use freeze for sharing:
    let bytes = big_buf.freeze();
    let mut start = 0;
    while let Some(pos) = bytes.iter().skip(start).position(|&b| b == b'\n') {
        let end = start + pos + 1;
        let msg = bytes.slice(start..end);
        messages.push(msg);
        start = end;
    }
}

use bytes::{Bytes, BytesMut};
 
fn when_to_freeze() {
    // Use freeze when:
    
    // 1. You've finished building the buffer and want to share it
    let mut buf = BytesMut::new();
    buf.extend_from_slice(b"Response data");
    let response: Bytes = buf.freeze(); // Share with multiple consumers
    
    // 2. You need immutable data for an API
    fn send_data(data: Bytes) {
        // API expects Bytes, not BytesMut
        println!("Sending: {:?}", data);
    }
    let mut buf = BytesMut::from("Data to send");
    send_data(buf.freeze());
    
    // 3. You want reference-counted sharing
    let mut buf = BytesMut::from("Shared data");
    let shared = buf.freeze();
    let copy1 = shared.clone();
    let copy2 = shared.clone();
    // All share same allocation
    
    // 4. You need to slice without copying
    let mut buf = BytesMut::from("Large data content");
    let bytes = buf.freeze();
    let slice = bytes.slice(0..5); // No copy, references original
    
    // DON'T use freeze when:
    // - You still need to modify the buffer
    // - You need to split ownership (use split_to)
    // - You're building incrementally (keep as BytesMut)
}

use bytes::BytesMut;
 
fn when_to_split() {
    // Use split_to/split_off when:
    
    // 1. Parsing framed data
    let mut buf = BytesMut::from("HEADERBODY");
    let header = buf.split_to(6); // "HEADER"
    let body = buf; // "BODY" (remainder)
    
    // 2. Extracting consumed portion for processing
    let mut buf = BytesMut::from("Command: data\r\nMore: info\r\n");
    while let Some(pos) = buf.windows(2).position(|w| w == b"\r\n") {
        let line = buf.split_to(pos + 2);
        process_line(line);
    }
    
    // 3. Independent ownership needed
    fn process_independently(a: BytesMut, b: BytesMut) {
        // Both can be modified independently
    }
    let mut buf = BytesMut::from("Part1Part2");
    let part1 = buf.split_to(5);
    process_independently(part1, buf);
    
    // 4. Buffer reuse pattern
    let mut rx_buffer = BytesMut::with_capacity(8192);
    loop {
        // receive_into(&mut rx_buffer);
        let complete = parse_and_split(&mut rx_buffer);
        // rx_buffer is reused for next receive
    }
    
    // DON'T use split when:
    // - You need shared read-only views (freeze + split_at)
    // - You need zero-copy sharing (freeze + clone)
}
 
fn process_line(_line: BytesMut) {}
fn parse_and_split(_buf: &mut BytesMut) -> bool { true }

use bytes::{Bytes, BytesMut};
 
fn combined_pattern() {
    // Pattern: split mutable buffer, then freeze parts
    
    let mut buf = BytesMut::from("HEADER:VALUE\r\nBODY:DATA\r\n");
    
    // Split off header line
    let header_end = buf.windows(2).position(|w| w == b"\r\n").unwrap();
    let header_buf = buf.split_to(header_end + 2);
    
    // Now we have two BytesMut:
    // - header_buf: "HEADER:VALUE\r\n"
    // - buf: "BODY:DATA\r\n"
    
    // We can continue processing buf as mutable
    // And freeze header_buf for sharing
    
    let header: Bytes = header_buf.freeze();
    
    // Parse header (now immutable, shareable)
    let (key, value) = parse_header(&header);
    println!("{} = {}", key, value);
    
    // Continue with body (still mutable)
    buf.extend_from_slice(b"\r\nEXTRA");
    let body: Bytes = buf.freeze();
    println!("Body: {:?}", body);
    
    // This pattern is useful when:
    // - You need mutable processing of some parts
    // - But want immutable sharing of other parts
}
 
fn parse_header(header: &[u8]) -> (&str, &str) {
    let header = std::str::from_utf8(header).unwrap();
    let parts: Vec<&str> = header.trim().split(':').collect();
    (parts[0], parts[1].trim())
}