Streaming

Process gigabyte interchanges in constant memory with reader iterators, message windows, and typed extraction.

edifact-rs provides multiple streaming APIs for processing large EDIFACT interchanges without loading the entire file into memory. All reader-based APIs are synchronous (std::io::Read) and can be bridged to async runtimes — see Async Integration.

API overview

Segment-level streaming

from_reader — raw segment stream

Process one OwnedSegment at a time without loading the interchange into memory:

use edifact_rs::from_reader;
use std::fs::File;

fn main() -> Result<(), edifact_rs::EdifactError> {
    let f = File::open("interchange.edi")?;

    for result in from_reader(f) {
        let seg = result?;
        println!("tag={} elements={}", seg.tag, seg.elements.len());
    }
    Ok(())
}

Memory: a single OwnedSegment is allocated per iteration. Previous segments are dropped before the next one is parsed.

Message windows

A message window is the slice of segments between a UNH and its matching UNT (inclusive). Envelope segments (UNB, UNZ, UNG, UNE) are skipped automatically.

from_bytes_windows — byte-slice source

use edifact_rs::from_bytes_windows;

let interchange = b"\
    UNB+UNOA:1+S+R+200101:0900+1'\
    UNH+1+ORDERS:D:96A:UN'BGM+220+PO-001+9'UNT+3+1'\
    UNH+2+ORDERS:D:96A:UN'BGM+220+PO-002+9'UNT+3+2'\
    UNZ+2+1'";

for result in from_bytes_windows(interchange) {
    let window = result?;
    // window.message_type  — Option<Cow<'_, str>> from UNH element 1, component 0
    // window.association_code — Option<Cow<'_, str>> from UNH DE 0057
    // call .as_deref() when you want an Option<&str>
    // window.segments     — [UNH, BGM, UNT]
    println!("{:?}: {} segments", window.message_type, window.segments.len());
}
# Ok::<(), edifact_rs::EdifactError>(())

message_windows_from_reader — reader source

use edifact_rs::message_windows_from_reader;
use std::fs::File;

fn main() -> Result<(), edifact_rs::EdifactError> {
    let f = File::open("multi_message.edi")?;
    for result in message_windows_from_reader(f) {
        let window = result?;
        // window.message_type is extracted from the UNH segment automatically
        println!("message type: {:?}", window.message_type);
    }
    Ok(())
}

Error propagation: if a UNH is opened but no UNT is found before end of input, the unclosed window is silently discarded (the iterator simply ends). Any I/O error from the underlying reader surfaces as EdifactError::Io.

Typed streaming — extract matching segments

Use deserialize_first_streaming / deserialize_all_streaming when you only care about specific segment types in an interchange.

use edifact_rs::{EdifactDeserialize, deserialize_first_streaming, deserialize_all_streaming};

#[derive(Debug, EdifactDeserialize)]
#[edifact(segment = "BGM")]
struct Bgm {
    #[edifact(element = 0)]
    doc_code: String,
    #[edifact(element = 1)]
    doc_id: String,
}

let input = b"UNH+1+ORDERS:D:11A:UN'BGM+220+PO-001+9'BGM+231+PO-002+9'UNT+4+1'";

// Stop after the first BGM:
let first: Bgm = deserialize_first_streaming(input)?;
assert_eq!(first.doc_id, "PO-001");

// Collect all BGM segments:
let all: Vec<Bgm> = deserialize_all_streaming(input)?;
assert_eq!(all.len(), 2);
assert_eq!(all[1].doc_id, "PO-002");
# Ok::<(), edifact_rs::EdifactError>(())

Reader variants

use edifact_rs::{deserialize_first_from_reader, deserialize_all_from_reader};
use std::io::Cursor;

# use edifact_rs::EdifactDeserialize;
# #[derive(Debug, EdifactDeserialize)]
# #[edifact(segment = "BGM")]
# struct Bgm { #[edifact(element = 0)] doc_code: String, #[edifact(element = 1)] doc_id: String }
let input = Cursor::new(b"BGM+220+PO-001+9'BGM+231+PO-002+9'".to_vec());

let first: Bgm = deserialize_first_from_reader(input.clone())?;
let all: Vec<Bgm> = deserialize_all_from_reader(input)?;
# Ok::<(), edifact_rs::EdifactError>(())

Message-level typed streaming

deserialize_messages_from_reader combines message-window iteration with typed deserialization. Each UNH..UNT window is deserialized into a message struct:

use edifact_rs::{EdifactDeserialize, deserialize_messages_from_reader};
use std::io::Cursor;

#[derive(Debug, EdifactDeserialize)]
#[edifact(segment = "BGM")]
struct Bgm {
    #[edifact(element = 0)]
    doc_code: String,
    #[edifact(element = 1)]
    doc_id: String,
}

#[derive(Debug, EdifactDeserialize)]
struct OrderMessage {
    bgm: Option<Bgm>,
}

let interchange = Cursor::new(b"\
    UNB+UNOA:1+S+R+200101:0900+1'\
    UNH+1+ORDERS:D:96A:UN'BGM+220+PO-001+9'UNT+3+1'\
    UNH+2+ORDERS:D:96A:UN'BGM+220+PO-002+9'UNT+3+2'\
    UNZ+2+1'".to_vec());

let messages: Vec<OrderMessage> =
    deserialize_messages_from_reader::<OrderMessage, _>(interchange)
        .collect::<Result<_, _>>()?;

assert_eq!(messages.len(), 2);
assert_eq!(messages[0].bgm.as_ref().unwrap().doc_id, "PO-001");
assert_eq!(messages[1].bgm.as_ref().unwrap().doc_id, "PO-002");
# Ok::<(), edifact_rs::EdifactError>(())

Zero-alloc owned deserialization path

deserialize_messages_from_reader calls T::edifact_deserialize_owned(&window) rather than converting OwnedSegment → Segment<'_>. The #[derive(EdifactDeserialize)] macro generates an override of edifact_deserialize_owned that accesses OwnedSegment::element_str and OwnedSegment::component_str directly — no intermediate Vec<Segment<'_>> is allocated.

This makes the reader path allocate at most:

• One OwnedMessageWindow per message window (released after deserialization)
• The deserialized T value itself

Progressive (per-window) validation

Combine message_windows_from_reader with ValidationContext to validate each message as it arrives, without buffering the whole interchange:

use edifact_rs::{
    ValidationContext, ProfileRulePack, ValidationIssue, ValidationSeverity,
    message_windows_from_reader, OwnedSegment,
};
use std::io::Cursor;

let input = Cursor::new(b"\
    UNH+1+ORDERS:D:11A:UN'BGM+220+PO-001+9'UNT+3+1'\
    UNH+2+ORDERS:D:11A:UN'BGM+220+PO-002+9'UNT+3+2'".to_vec());

let pack = ProfileRulePack::new("ORDERS-PROGRESSIVE")
    .for_message_type("ORDERS")
    .with_stateless_rule_fn(|segs, issues| {
        if !segs.iter().any(|s| s.tag == "BGM") {
            issues.push(
                ValidationIssue::new(
                    ValidationSeverity::Error,
                    "every ORDERS message must contain a BGM segment",
                )
                .with_rule_id("ORDERS-P001"),
            );
        }
    });

let ctx = ValidationContext::builder()
    .with_profile_pack(pack)
    .build();

for result in message_windows_from_reader(input) {
    let window = result?;
    // validate using borrowed views of the OwnedMessageWindow's segments
    let borrowed: Vec<_> = window.segments.iter().map(|s| s.as_borrowed()).collect();
    let report = ctx.validate_lenient(&borrowed);
    if !report.is_valid() {
        for e in report.errors() {
            eprintln!("❌ {}", e.message);
        }
    }
}
# Ok::<(), edifact_rs::EdifactError>(())

See the full example in cookbook_streamed_progressive_validation.rs.

Manual window assembly with from_reader

For full control over window boundaries (e.g. custom grouping logic), use the raw segment iterator:

use edifact_rs::{from_reader, OwnedSegment};
use std::io::Cursor;

let input = Cursor::new(b"\
    UNH+1+ORDERS:D:11A:UN'\
    BGM+220+PO-001+9'\
    UNT+3+1'".to_vec());

let mut current: Vec<OwnedSegment> = Vec::new();
let mut in_message = false;

for result in from_reader(input) {
    let seg = result?;
    match seg.tag.as_str() {
        "UNH" => {
            current.clear();
            in_message = true;
        }
        "UNT" => {
            if in_message {
                current.push(seg);
                println!("window complete: {} segments", current.len());
                in_message = false;
            }
            continue;
        }
        _ => {}
    }
    if in_message {
        current.push(seg);
    }
}
# Ok::<(), edifact_rs::EdifactError>(())

Memory budget summary

Next steps

• Validation — validate windows and messages
• Async Integration — bridging to tokio
• Performance — allocation analysis and benchmarks

Decoding a stream without making it eager

A UNOC…UNOK interchange has to be transcoded before it can be parsed, and the repertoire is declared inside the stream itself — so something has to read as far as the UNB before the first segment can come out.

decode_reader does that read up front and therefore returns a Result. That is honest, but a ? on it turns a lazy pipeline eager: a function returning impl Iterator<Item = Result<T, E>> suddenly has to return Result<impl Iterator<…>>, and callers end up boxing the iterator or wrapping the decode error in a one-item chain.

from_reader_decoded keeps the signature a plain Iterator by deferring the sniff to the first next():

use edifact_rs::from_reader_decoded;

let mut raw = b"UNB+UNOC:3+S+R+260101:0900+IC1'NAD+BY+M".to_vec();
raw.push(0xFC); // `ü` in ISO 8859-1
raw.extend_from_slice(b"ller'UNZ+0+IC1'");

// No `?` before the pipeline — construction cannot fail.
let segments: Vec<_> = from_reader_decoded(std::io::Cursor::new(raw))
    .collect::<Result<Vec<_>, _>>()?;
assert_eq!(segments[1].element_str(1), Some("Müller"));
# Ok::<(), edifact_rs::EdifactError>(())

A repertoire this crate cannot decode — UNOX, KECA — arrives as the first item of the iterator, in the same shape as a parse error, and the iterator then ends:

# use edifact_rs::{EdifactError, from_reader_decoded};
let raw = b"UNB+UNOX:3+S+R+260101:0900+IC1'UNZ+0+IC1'";
let mut stream = from_reader_decoded(std::io::Cursor::new(&raw[..]));

assert!(matches!(
    stream.next().unwrap(),
    Err(EdifactError::UnsupportedCharset { .. })
));
assert!(stream.next().is_none());
# Ok::<(), edifact_rs::EdifactError>(())

from_bytes_decoded is the slice counterpart. It returns owned segments and a Result, because an ISO 8859-1 payload has to be transcoded to exist as UTF-8 at all and the decoded buffer belongs to the call. When you want to keep the zero-copy path, hold the buffer yourself: decode_interchange then from_bytes.

Reach for the decoding entry points by default. A UNOC corpus that happens to be stored as UTF-8 parses fine without them — so the tests pass, and the first conformant counterparty message is the one that fails.

Edit this page on GitHub