Belajar Axum - Streaming & Multipart

#Pendahuluan

Sejauh ini kita selalu memuat body ke memori penuh: Json, String, Bytes — semuanya buffered. Untuk data besar — file upload, export besar, streaming log — buffering penuh tidak bisa diterima: file 2 GB tidak mungkin dimuat ke RAM. Episode ini membahas streaming: memproses data sedikit demi sedikit, baik untuk response (server mengirim bertahap) maupun request (klien mengirim bertahap), plus multipart upload.

Mengapa ini materi production penting? Karena aplikasi yang menangani file — gambar profil, attachment, media — pasti menghadapi masalah "body terlalu besar". Memahami streaming dan multipart membedakan aplikasi yang bisa menangani file raksasa dari yang crash-nya misterius.

#Streaming Response: Mengirim Data Bertahap

Ketika response berupa data yang besar atau tidak diketahui panjangnya, gunakan stream. Contoh paling umum: melayani file besar tanpa memuat ke memori.

Streaming file

use axum::{
    body::Body,
    http::{header, HeaderValue, StatusCode},
    response::{IntoResponse, Response},
};
use tokio_util::io::ReaderStream;
 
async fn download_large() -> Response {
    // Buka file di disk, stream langsung ke klien
    let file = tokio::fs::File::open("data/export.csv").await.unwrap();
    let stream = ReaderStream::new(file);
 
    let headers = [
        (
            header::CONTENT_DISPOSITION,
            HeaderValue::from_static("attachment; filename=\"export.csv\""),
        ),
        (
            header::CONTENT_TYPE,
            HeaderValue::from_static("text/csv; charset=utf-8"),
        ),
    ];
 
    (headers, Body::from_stream(stream)).into_response()
}

Kunci di sini: Body::from_stream(stream) — response body dibangun dari stream async. ReaderStream mengubah tokio::fs::File menjadi stream chunk — file dibaca dari disk dalam potongan, dikirim ke klien, tanpa pernah dimuat utuh ke RAM.

#Streaming output generator

Pattern yang sama untuk data yang dihasilkan on-the-fly — misal CSV besar yang dibangun per baris:

Streaming generator

use tokio_stream::StreamExt;
use tokio_stream::wrappers::ReceiverStream;
 
fn csv_stream() -> impl futures_core::Stream<Item = Result<Vec<u8>, std::io::Error>> {
    let (tx, rx) = tokio::sync::mpsc::channel::<Vec<u8>>(8);
    tokio::spawn(async move {
        for i in 0..1_000_000 {
            if tx.send(format!("{i},data\n").into_bytes()).await.is_err() {
                break; // klien menutup koneksi
            }
        }
    });
    ReceiverStream::new(rx).map(Ok)
}

Catatan penting: kirim dalam chunk kecil (Vec<u8> per baris) dan berhenti saat send error — itu berarti klien menutup koneksi, dan mengirim terus hanya membuang resource.

Warning

Respons streaming tidak boleh di-compress global jika hasilnya harus "chunked" real-time (kompresi menunda pengiriman sampai buffer penuh). Untuk endpoint export, nonaktifkan compression pada route tersebut atau terima delay — keputusan tergantung kebutuhan klien.

#Streaming Request: Menerima Body Bertahap

Terkadang request body juga besar — klien mengunggah file. Mengonsumsi body sebagai stream memungkinkan proses per-chunk:

Streaming request body

use axum::body::Body;
 
async fn consume_stream(body: Body) -> String {
    let mut stream = body.into_data_stream();
 
    let mut total = 0usize;
    while let Some(chunk) = stream.next().await {
        match chunk {
            Ok(bytes) => {
                total += bytes.len();
                // proses per chunk, tanpa menyimpan semuanya
            }
            Err(err) => {
                tracing::error!(error = %err, "error saat membaca body");
                break;
            }
        }
    }
    format!("total {total} bytes diterima")
}

Body::into_data_stream() mengubah body menjadi stream Bytes. Berbeda dari String/Bytes extractor yang menunggu seluruh body, pola ini mulai memproses begitu chunk pertama tiba. Gunakan untuk: menulis upload langsung ke disk, hashing file saat diupload, atau menghitung ukuran.

#RequestBodyLimit: Kontrol Ukuran

Body stream punya batas default 2 MB dari Axum. Dua cara mengubahnya:

Batasi body per route

use axum::extract::DefaultBodyLimit;
 
Router::new()
    .route("/upload", axum::routing::post(upload))
    .layer(DefaultBodyLimit::max(100 * 1024 * 1024)) // 100 MB

Atau batasi streaming dengan RequestBodyLimit layer — berhenti menerima setelah limit tercapai:

Batas pada stream

use axum::extract::{BodyLimit, DefaultBodyLimit};
 
async fn upload(body: Body) -> &'static str {
    let mut stream = body.into_data_stream();
    let mut total = 0usize;
    while let Some(chunk) = stream.next().await {
        total += chunk.map(|b| b.len()).unwrap_or(0);
        if total > 100 * 1024 * 1024 {
            return "terlalu besar";
        }
    }
    "oke"
}

Tip

Setel body limit per kebutuhan endpoint: DefaultBodyLimit::max kecil untuk JSON API (mencegah body raksasa menghabiskan parsing), besar hanya untuk route upload. Limit global yang seragam jarang cocok untuk semua endpoint.

#Multipart Upload

Upload file web klasik memakai multipart/form-data. Axum menyediakan Multipart extractor yang membaca field dan file satu per satu:

Handler multipart

use axum::extract::Multipart;
 
pub async fn upload(
    State(state): State<AppState>,
    mut multipart: Multipart,
) -> AppResult<(StatusCode, Json<serde_json::Value>)> {
    let mut saved = Vec::new();
 
    while let Some(field) = multipart.next_field().await.map_err(AppError::from)? {
        let name = field.name().unwrap_or("").to_string();
        let file_name = field.file_name().unwrap_or("").to_string();
 
        match name.as_str() {
            "file" => {
                // stream file langsung ke disk — tanpa memuat penuh
                let path = std::path::Path::new("/tmp/uploads").join(&file_name);
                let mut out = tokio::fs::File::create(&path).await.map_err(AppError::from)?;
 
                let mut stream = field.bytes_stream();
                while let Some(chunk) = stream.next().await {
                    let chunk = chunk.map_err(AppError::from)?;
                    out.write_all(&chunk).await.map_err(AppError::from)?;
                }
                saved.push(path.to_string_lossy().to_string());
            }
            "description" => {
                let text = field.text().await.map_err(AppError::from)?;
                tracing::info!(description = %text, "metadata diterima");
            }
            _ => {
                // field tak dikenal: lewati
            }
        }
    }
 
    Ok((StatusCode::CREATED, Json(serde_json::json!({ "files": saved }))))
}

Poin penting multipart:

field.bytes_stream() — stream file per field; kita tulis ke disk per chunk (pola streaming dari atas). File 1 GB tidak pernah penuh di RAM.
Field dan file diproses urut — next_field() mengonsumsi stream sampai habis; jangan mengembalikan response di tengah kecuali sengaja membatalkan upload.
field.text() untuk field teks, bytes_stream() untuk file.
Validasi file: cek ekstensi/MIME dan batasi ukuran — jangan menerima file apa pun tanpa pemeriksaan (keamanan di episode 18).

Warning

Sanitasi nama file! file_name datang dari klien — path traversal seperti ../../etc/passwd bisa menimpa file server. Selalu validasi: gunakan hanya basename, buang karakter berbahaya, dan tulis di direktori yang diizinkan. Jangan pernah menaruh nama file mentah ke path.

#Kombinasi: Upload + Antrian + Status

Mari rangkai streaming, multipart, dan background tasks (episode 22) menjadi satu alur realistis:

Upload dengan antrian proses

use axum::extract::{DefaultBodyLimit, Multipart};
use std::path::PathBuf;
use tokio::io::AsyncWriteExt;
 
pub async fn upload_video(
    State(state): State<AppState>,
    mut multipart: Multipart,
) -> AppResult<(StatusCode, Json<serde_json::Value>)> {
    let mut job_id = None;
 
    while let Some(field) = multipart.next_field().await.map_err(AppError::from)? {
        if field.name() == Some("video") {
            let file_name = sanitize_name(field.file_name().unwrap_or("upload.bin"));
            let target = PathBuf::from("/data/uploads").join(file_name);
 
            let mut out = tokio::fs::File::create(&target).await.map_err(AppError::from)?;
            let mut stream = field.bytes_stream();
            while let Some(chunk) = stream.next().await {
                out.write_all(&chunk.bytes).await.map_err(AppError::from)?;
            }
            out.flush().await.map_err(AppError::from)?;
 
            // jadwalkan transcoding sebagai background job (episode 22)
            job_id = Some(state.jobs.send(Job::TranscodeVideo { path: target }).await);
        }
    }
 
    Ok((StatusCode::ACCEPTED, Json(serde_json::json!({ "job_id": job_id }))))
}
 
fn sanitize_name(name: &str) -> String {
    name.split('/')
        .last()
        .unwrap_or("upload")
        .chars()
        .map(|c| if c.is_ascii_alphanumeric() || c == '.' || c == '-' || c == '_' { c } else { '_' })
        .collect()
}

Alur lengkap: stream upload ke disk → sanitize nama → antri proses berat → balas 202 dengan job ID. Klien bisa mengecek status via endpoint yang membaca job state (implementasi dari episode 22).

#Kapan Tidak Memakai Streaming

Streaming bukan selalu jawaban. Untuk payload kecil, buffering penuh (Json, Bytes) lebih sederhana dan cepat. Streaming menambah kompleksitas: management error per-chunk, pembatalan koneksi, dan penyimpanan antara. Pilih streaming saat:

Ukuran tidak diketahui atau bisa sangat besar (file upload, export).
Klien butuh data bertahap (log, progress).
Data lebih besar dari batas memori yang masuk akal.

Jika ukuran selalu kecil dan sudah dibatasi (DefaultBodyLimit), extractor biasa tetap pilihan terbaik.

#Penutup

Pada episode 23 ini aplikasi kalian bisa menangani data besar:

Streaming response dengan Body::from_stream + ReaderStream (file tanpa memuat RAM).
Streaming request dengan into_data_stream() per chunk.
DefaultBodyLimit dan RequestBodyLimit untuk kontrol ukuran.
Multipart upload ke disk per chunk + sanitasi nama file.
Alur upload → antrian → response 202 dengan job ID.

Di episode 24 selanjutnya kita bawa aplikasi ke dunia nyata: deployment & Docker — multi-stage Dockerfile (scratch/distroless), pengaturan worker, dan deploy ke managed platform atau VPS. Sampai jumpa di episode 24!