Continuously reading from a `TcpStream`

[redeexpressos]

Hi. I did not find any examples regarding reading continuously from a TcpStream. I see that there is https://docs.rs/deku/latest/deku/noseek/struct.NoSeek.html in the docs, but no example. Would be possible to provide some examples? Thanks.

[sharksforarms]

yeah good idea, the only example we have currently is

deku/tests/test_from_reader.rs

Lines 37 to 41 in 257bbd2

	let inner = &c.into_inner();
	let mut s = NoSeek::new(inner.as_slice());
	let (amt_read, ret_read) = TestDeku::from_reader((&mut s, 0)).unwrap();
	assert_eq!(amt_read, 4);
	assert_eq!(TestDeku(0b0110), ret_read);

[redeexpressos]

@sharksforarms Can I get an example while there is no PR for it ? :) My use-case is reading from a generic connection (AsyncRead / AsyncWrite). One thing that makes me confused is that I use deku to only read certain bytes (bytes_read) and I'm not sure how this is handled internally (I don't want to lose data), as one read can return 1 message and half another maybe?

[sharksforarms]

@redeexpressos no guarantees, but here's a quick example I came up with. The tricky part is you need to keep track of the buffer and the bit offset across reads if your data isn't aligned to a byte

use deku::prelude::*;
use deku::reader::Reader;
use std::io::Cursor;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};

#[derive(Debug, DekuRead, DekuWrite)]
#[deku(endian = "big")]
struct MyData {
    #[deku(bits = 4)]
    field1: u8,

    field2: u8,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let listener = TcpListener::bind("127.0.0.1:8080").await?;

    loop {
        let (socket, addr) = listener.accept().await?;
        println!("New connection from: {}", addr);

        tokio::spawn(async move {
            handle_client(socket).await;
        });
    }
}

async fn handle_client(mut socket: TcpStream) {
    let mut read_buf = [0; 1024];
    let mut buffer = Vec::new();
    let mut bit_offset = 0; // Track bit position across reads

    loop {
        match socket.read(&mut read_buf).await {
            Ok(0) => {
                println!("Client disconnected");
                return;
            }
            Ok(n) => {
                // Append new data to our persistent buffer
                buffer.extend_from_slice(&read_buf[0..n]);

                loop {
                    let mut cursor = Cursor::new(&buffer);
                    let mut reader = Reader::new(&mut cursor);

                    // Skip to our current bit offset
                    if bit_offset > 0 {
                        reader.skip_bits(bit_offset).unwrap();
                    }

                    println!("buffer: {:?}", &buffer);
                    println!("buffer[0]: {:08b}", &buffer[0]);
                    match MyData::from_reader_with_ctx(&mut reader, ()) {
                        Ok(data) => {
                            println!(
                                "Received: field1={:08b}, field2={:08b}",
                                data.field1, data.field2
                            );

                            // Echo back the data
                            if let Ok(bytes) = data.to_bytes() {
                                if let Err(e) = socket.write_all(&bytes).await {
                                    eprintln!("Failed to write to socket: {}", e);
                                    return;
                                }
                            }

                            // Update bit offset
                            bit_offset += reader.bits_read - bit_offset;

                            // Remove complete bytes from buffer, keep partial byte
                            let bytes_to_remove = bit_offset / 8;
                            dbg!(bytes_to_remove);

                            bit_offset = bit_offset % 8; // Keep remainder bits
                            dbg!(bit_offset);

                            buffer.drain(0..bytes_to_remove);
                        }
                        Err(_e) => {
                            // Not enough data for a complete message, break and wait for more
                            break;
                        }
                    }
                }
            }
            Err(e) => {
                eprintln!("Failed to read from socket: {}", e);
                return;
            }
        }
    }
}

import socket

def try_send():
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect(('127.0.0.1', 8080))
    message = bytes([0b1001_1111, 0b1001_0000, 0xAA, 0xBB])
    sock.sendall(message)

if __name__ == "__main__":
    try_send()

buffer: [159, 144, 170, 187]
buffer[0]: 10011111
Received: field1=00001001, field2=11111001
[src/main.rs:76:29] bytes_to_remove = 1
[src/main.rs:79:29] bit_offset = 4
buffer: [144, 170, 187]
buffer[0]: 10010000
Received: field1=00000000, field2=10101010

@wcampbell0x2a I wonder if you'd have a better method you've used in your projects?