Async Integration

Three patterns for using the synchronous edifact-rs parser from tokio without blocking the runtime.

edifact-rs is intentionally synchronous — it exposes std::io::Read-based APIs. This is the right design for EDIFACT:

• Parsing is CPU-bound after the first read; async would add scheduling overhead without benefit.
• Imposing an async runtime on every downstream user is a heavy design tax.
• Integration is easy: the patterns below bridge to any async runtime cleanly.

This guide covers three canonical patterns for Tokio. The same patterns apply to other runtimes (async-std, smol) with minor API substitutions.

Dependencies

cargo add edifact-rs
cargo add tokio --features rt-multi-thread,macros,fs,sync

Pattern A — Read into memory, parse synchronously

Best for: payloads that fit comfortably in memory (typical EDIFACT messages are well under 1 MB).

use edifact_rs::{from_bytes, Segment};

async fn process_edi_file(path: &str) -> Result<(), Box<dyn std::error::Error>> {
    // Async I/O: read the file without blocking the runtime
    let bytes = tokio::fs::read(path).await?;

    // Synchronous parsing on the async thread — safe because it's fast
    let segments: Vec<_> = from_bytes(&bytes).collect::<Result<_, _>>()?;

    for seg in &segments {
        println!("{}", seg.tag);
    }
    Ok(())
}

This is the simplest pattern. Because from_bytes is O(n) in CPU (no blocking I/O), it is safe to call from an async context without spawn_blocking.

Pattern B — spawn_blocking for large files or blocking sources

Best for: large files (multi-MB interchanges), blocking file systems, or reader-based streaming.

use edifact_rs::{OwnedMessageWindow, message_windows_from_reader};

async fn process_large_interchange(path: String)
    -> Result<Vec<OwnedMessageWindow>, Box<dyn std::error::Error>>
{
    let windows = tokio::task::spawn_blocking(move || {
        let f = std::fs::File::open(&path)?;
        message_windows_from_reader(f)
            .collect::<Result<Vec<_>, _>>()
    })
    .await??;  // propagates both JoinError and EdifactError

    Ok(windows)
}

spawn_blocking moves the work onto Tokio's dedicated blocking thread pool, keeping the async worker threads free.

Pattern C — Channel bridge (backpressure-safe streaming)

Best for: very large interchanges where you want to process each UNH..UNT window on the async side as soon as it is parsed, without buffering all windows.

use edifact_rs::{EdifactError, OwnedMessageWindow, message_windows_from_reader};
use tokio::sync::mpsc;

async fn stream_interchange(
    bytes: Vec<u8>,
) -> Result<Vec<OwnedMessageWindow>, EdifactError> {
    // Channel with backpressure (capacity = 8 windows in flight)
    let (tx, mut rx) = mpsc::channel::<Result<OwnedMessageWindow, EdifactError>>(8);

    // Producer: blocking thread
    tokio::task::spawn_blocking(move || {
        let cursor = std::io::Cursor::new(bytes);
        for window in message_windows_from_reader(cursor) {
            if tx.blocking_send(window).is_err() {
                break; // consumer dropped; stop parsing early
            }
        }
    });

    // Consumer: async side — processes windows as they arrive
    let mut all: Vec<OwnedMessageWindow> = Vec::new();
    while let Some(result) = rx.recv().await {
        all.push(result?);
    }
    Ok(all)
}

The channel capacity controls backpressure: if the consumer is slow, the producer blocks until there is space in the channel buffer.

Pattern D — Typed message streaming with channel

Combine Pattern C with deserialize_messages_from_reader for typed output:

use edifact_rs::{EdifactError, EdifactDeserialize, deserialize_messages_from_reader};
use tokio::sync::mpsc;

# #[derive(Debug, EdifactDeserialize)]
# #[edifact(segment = "BGM")]
# struct Bgm { #[edifact(element = 0)] doc_code: String }
# #[derive(Debug, EdifactDeserialize)]
# struct OrderMessage { bgm: Option<Bgm> }
async fn stream_typed_messages(
    bytes: Vec<u8>,
) -> Result<Vec<OrderMessage>, EdifactError> {
    let (tx, mut rx) = mpsc::channel::<Result<OrderMessage, EdifactError>>(8);

    tokio::task::spawn_blocking(move || {
        let cursor = std::io::Cursor::new(bytes);
        for msg in deserialize_messages_from_reader::<OrderMessage, _>(cursor) {
            if tx.blocking_send(msg).is_err() {
                break;
            }
        }
    });

    let mut messages = Vec::new();
    while let Some(result) = rx.recv().await {
        messages.push(result?);
    }
    Ok(messages)
}

Pattern E — Process bytes from AsyncRead

When your async input is an AsyncRead (e.g. tokio::net::TcpStream), buffer it into memory before parsing, or use spawn_blocking with a synchronous wrapper:

use tokio::io::{AsyncReadExt};
use edifact_rs::from_bytes;

async fn parse_from_async_reader<R: AsyncReadExt + Unpin>(
    mut reader: R,
) -> Result<usize, Box<dyn std::error::Error>> {
    let mut buf = Vec::new();
    reader.read_to_end(&mut buf).await?;

    // Now parse synchronously
    let segments: Vec<_> = from_bytes(&buf).collect::<Result<_, _>>()?;
    Ok(segments.len())
}

Why not AsyncBufRead? edifact-rs exposes std::io::BufRead internally. The simplest bridge is read_to_end into a Vec<u8>. For byte-by-byte bridging see the async-compat crate or write an std::io::Read adapter.

Choosing a pattern

Error propagation

spawn_blocking returns a JoinError if the worker thread panics. Map it to your own error type:

use edifact_rs::EdifactError;

async fn safe_parse(bytes: Vec<u8>) -> Result<usize, Box<dyn std::error::Error + Send + Sync>> {
    let count = tokio::task::spawn_blocking(move || {
        edifact_rs::from_bytes(&bytes).count()
    })
    .await?;   // propagates JoinError if the worker thread panicked
    Ok(count)
}

Complete example

cargo run -p edifact-rs --example cookbook_async_tokio_integration

See cookbook_async_tokio_integration.rs for a runnable example demonstrating Patterns A, B, and C.

Next steps

• Streaming — synchronous streaming APIs
• Performance — memory budgets and benchmarking

Edit this page on GitHub