How to read a big uploaded file line by line, using streams ?

[fungiboletus]

Summary

Hi,

I'm trying to parse large uploaded CSV files without having them in memory. However I'm a bit struggling and lost between all the streams in Rust, Tokio, and Axum. I have found Bytes, Body, BodyDataStream, BodyStream (don't know what this one does), ReaderStream, StreamReader (don't know the difference).

From what I understood, Bytes will receive all the data and keep it in memory. So I need something else.

I came up with the following example that counts the end of lines in the uploaded CSV chunk by chunk.

async fn post_csv(body: axum::body::Body) -> Result<String, (StatusCode, String)> {
    let mut count = 0usize;
    let mut stream = body.into_data_stream();

    loop {
        let chunk = stream.try_next().await;
        match chunk {
            Ok(bytes) => match bytes {
                Some(bytes) => count += bytes.into_iter().filter(|b| *b == b'\n').count(),
                None => break,
            },
            Err(_) => {
                return Err((
                    StatusCode::INTERNAL_SERVER_ERROR,
                    "Error reading body".to_string(),
                ))
            }
        }
    }

    Ok(count.to_string())
}

I could continue from this snippet and implement some logic to identify the end of lines and re-chunks my chunks into chunks delimited by end of lines, before parsing the CSV chunks.

But I feel like I'm re-implementing a bad wheel. It should be a simple solution using two magic lines and one or two use that would allow me to do what I want: get a line by line (body)stream reader. Does that exist ?

axum version

0.7

[davidpdrsn]

Getting a stream of lines is a bit hard unfortunately but you can do it like this https://github.com/tokio-rs/axum/blob/main/axum-extra/src/json_lines.rs#L111

[fungiboletus]

Thanks a lot for the very quick answer! It's nice to have the solution of the puzzle.

Here is the updated code snippet, that runs twice as fast on my laptop:

use std::io;
use futures::TryStreamExt;
use tokio::io::AsyncBufReadExt;
use tokio_stream::wrappers::LinesStream;
use tokio_util::io::StreamReader;

async fn post_csv(body: axum::body::Body) -> Result<String, (StatusCode, String)> {
    let stream = body.into_data_stream();
    let stream = stream.map_err(|err| io::Error::new(io::ErrorKind::Other, err));
    let read = StreamReader::new(stream);
    let lines_stream = LinesStream::new(read.lines());

    let count = lines_stream
        .try_fold(0, |acc, _| async move { Ok(acc + 1) })
        .await;

    match count {
        Ok(count) => Ok(count.to_string()),
        Err(_) => Err((
            StatusCode::INTERNAL_SERVER_ERROR,
            "Error reading body".to_string(),
        )),
    }
}