Struct NdjsonStream 

pub struct NdjsonStream;

NDJSON streaming utilities.

Provides methods for parsing NDJSON streams into typed structs. Works with any type implementing DeserializeOwned.

Type Safety

Parsing is type-safe - invalid JSON or incompatible types result in StreamError::JsonParse rather than panics.

Implementations

impl NdjsonStream

pub fn parse<T>( lines: impl Stream<Item = Result<String, Error>>, ) -> impl Stream<Item = Result<T, StreamError>>

where T: DeserializeOwned,

Parse stream of lines into typed structs.

Each line should contain a valid JSON object matching type T. Invalid lines result in StreamError::JsonParse.

Type Parameters

• T - Target type, must implement DeserializeOwned

Example

use wme_stream::NdjsonStream;
use wme_models::Article;
use futures::StreamExt;
use std::pin::pin;

let articles = NdjsonStream::parse::<Article>(lines);
let mut pinned = pin!(articles);
while let Some(result) = pinned.next().await {
    match result {
        Ok(article) => println!("{}", article.name),
        Err(e) => eprintln!("Parse error: {}", e),
    }
}

pub fn parse_articles( lines: impl Stream<Item = Result<String, Error>>, ) -> impl Stream<Item = Result<Article, StreamError>>

Parse stream specifically for Articles.

Convenience method equivalent to parse::<Article>(lines).

Example

use wme_stream::NdjsonStream;
use futures::StreamExt;
use std::pin::pin;

let articles = NdjsonStream::parse_articles(lines);
let mut pinned = pin!(articles);
while let Some(result) = pinned.next().await {
    if let Ok(article) = result {
        println!("{}: {}", article.identifier, article.name);
    }
}

pub fn from_reader<R: BufRead + Send + 'static>( reader: R, ) -> impl Stream<Item = Result<String, Error>>

Read lines from a buffered reader.

Creates an async stream from any BufRead implementor. Useful for reading NDJSON from files or network streams.

Performance

The reader is read synchronously but yields to the async runtime between lines. For large files, this provides good throughput without blocking the runtime.

Example

use wme_stream::NdjsonStream;
use std::io::BufReader;
use std::fs::File;

let file = File::open("data.ndjson").unwrap();
let reader = BufReader::new(file);
let lines = NdjsonStream::from_reader(reader);