Streaming IO

Datum's io module provides byte-stream IO for the most common sources and sinks: synchronous file IO (FileIO), async Tokio-backed file and TCP IO (TokioFileIO, TokioTcp), byte-stream framing (Framing), compression (Compression), adapters for any std::io::Read or std::io::Write (StreamConverters), and blocking-handle bridges (as_input_stream/as_output_stream). All of these compose with the standard Source/Flow/Sink DSL.

All IO types are re-exported at the crate root (use datum::{Framing, TokioFileIO, …}). They are not part of datum::prelude, so import them explicitly — the examples below do.

File IO

FileIO is a synchronous, low-overhead file adapter backed by StreamConverters under the hood. It runs the reader on Datum's thread pool and keeps a bounded queue between the reader thread and the stream consumer to avoid unbounded read-ahead.

FileIO::from_path(path, chunk_size)  →  Source<Vec<u8>>
FileIO::from_path_default(path)      →  Source<Vec<u8>>   (chunk_size = 8192)
FileIO::to_path(path)                →  Sink<Vec<u8>, StreamCompletion<NotUsed>>

to_path creates the file if it does not exist and truncates it if it does. The materialized StreamCompletion<NotUsed> resolves when the file is fully flushed and closed.

use datum::{FileIO, Sink, Source};
use std::fs;

// Use a unique temp path to avoid conflicts between parallel test runs.
let path = std::env::temp_dir().join(format!("datum_docs_file_io_{}.bin", std::process::id()));

// Write a file with FileIO::to_path, then read it back with FileIO::from_path.
Source::single(b"datum streaming io test".to_vec())
    .run_with(FileIO::to_path(&path))
    .unwrap()
    .wait()
    .unwrap();

let chunks: Vec<Vec<u8>> = FileIO::from_path(&path, 8)
    .run_with(Sink::collect())
    .unwrap()
    .wait()
    .unwrap();

let data: Vec<u8> = chunks.into_iter().flatten().collect();
assert_eq!(data, b"datum streaming io test");

Performance. Sync file source and sink are competitive to significantly faster than Akka on both wall-clock and CPU. No hidden busy-spin cost — CPU tracks wall closely. Current numbers are maintained in roadmap/benchmarks/streaming-io.md.

Tokio file IO

TokioFileIO uses Tokio's async file IO. Compared with FileIO, it adds one producer Tokio task and uses ConsumerWaker::unpark to wake the stream thread after each chunk arrives. The materialized value carries an IoResult with a byte count and terminal status — useful for observability and auditing.

TokioFileIO::from_path(path, chunk_size)  →  TokioByteSource  (= Source<Vec<u8>, StreamCompletion<IoResult>>)
TokioFileIO::from_path_default(path)      →  TokioByteSource
TokioFileIO::to_path(path)                →  TokioByteSink    (= Sink<Vec<u8>, StreamCompletion<IoResult>>)

IoResult fields:

Method	Description
.bytes()	Bytes successfully transferred
.is_success()	true when no error occurred
.status()	StreamResult<()> — Ok(()) on success, Err(StreamError) on failure

use datum::{Keep, Sink, TokioFileIO};

let (io_result, frames) = TokioFileIO::from_path("events.log", 8192)
    .via(datum::Framing::delimiter(b"\n".to_vec(), 4096, true))
    .to_mat(Sink::collect(), Keep::both)
    .run()
    .unwrap();

let lines = frames.wait().unwrap();
let result = io_result.wait().unwrap();
println!("{} bytes read, {} lines", result.bytes(), lines.len());

Performance. Tokio file source and sink are at parity or better vs Akka on wall-clock. The source's CPU-to-wall gap is tracked-acceptable and attributable to the two-thread producer/consumer model (the named lever). The sink's CPU tracks wall closely. Current numbers are maintained in roadmap/benchmarks/streaming-io.md.

Framing

Framing turns a raw byte stream (possibly delivered in arbitrary chunks) into a stream of complete logical frames. All framing flows have type Flow<Vec<u8>, Vec<u8>> and compose with any byte source.

Delimiter framing

use datum::{Framing, Sink, Source};

// Framing::delimiter(delimiter, max_frame_length, allow_truncation)
// splits a byte stream on a byte sequence and strips the delimiter.
// allow_truncation = true emits a partial final frame when the stream ends
// without a terminating delimiter; false would fail with StreamError.
let chunks: Vec<Vec<u8>> = vec![
    b"hello\nworld\n".to_vec(),
    b"foo".to_vec(), // no trailing newline — emitted as a truncated frame
];

let frames: Vec<Vec<u8>> = Source::from_iter(chunks)
    .via(Framing::delimiter(b"\n".to_vec(), 256, true))
    .run_with(Sink::collect())
    .unwrap()
    .wait()
    .unwrap();

allow_truncation:

• true — emit the remaining buffer as a final frame when the stream ends without a terminating delimiter.
• false — fail the stream with StreamError if the stream ends mid-frame.

JSON object framing

use datum::{Framing, Sink, Source};

// Framing::json(max_object_length) splits concatenated JSON objects.
// Input chunks may split anywhere — even inside a key string.
let input: Vec<Vec<u8>> = vec![b"{\"a\":1}{\"b\"".to_vec(), b":2}".to_vec()];

let frames: Vec<Vec<u8>> = Source::from_iter(input)
    .via(Framing::json(1024))
    .run_with(Sink::collect())
    .unwrap()
    .wait()
    .unwrap();

Framing::json(max_object_length) extracts top-level JSON objects from a concatenated byte stream. It handles objects split across arbitrary chunk boundaries and advances past outer array brackets and commas so it works on both {...}{...} and [{...},{...}] input.

Framing methods

Method	Splits on
Framing::delimiter(delimiter, max_len, allow_truncation)	Byte sequence; strips the delimiter
Framing::length_field(field_len, field_offset, max_len, byte_order)	Length-prefixed frames; field_len is 1–4 bytes
Framing::json(max_len)	Top-level JSON objects

FramingByteOrder::BigEndian and FramingByteOrder::LittleEndian control how length_field interprets the length header.

Performance. Delimiter framing is at parity or better vs Akka with significantly lower allocation. JSON framing is at parity. Current numbers are maintained in roadmap/benchmarks/streaming-io.md.

Compression

Compression flows compress or decompress a Vec<u8> byte stream in-process using flate2. All four variants return Flow<Vec<u8>, Vec<u8>>.

Flow	Direction
Compression::gzip()	Compress (gzip format)
Compression::gunzip()	Decompress (gzip format)
Compression::deflate()	Compress (zlib/deflate format)
Compression::inflate()	Decompress (zlib/deflate format)

use datum::{Compression, Sink, Source};

// Compress a byte stream with gzip, then decompress it back.
let compressed: Vec<Vec<u8>> = Source::from_iter([b"hello ".to_vec(), b"world".to_vec()])
    .via(Compression::gzip())
    .run_with(Sink::collect())
    .unwrap()
    .wait()
    .unwrap();

let recovered: Vec<u8> = Source::from_iter(compressed)
    .via(Compression::gunzip())
    .run_with(Sink::collect())
    .unwrap()
    .wait()
    .unwrap()
    .into_iter()
    .flatten()
    .collect();

Compression and decompression flows compose directly with framing — for example, gzip a newline-delimited log file after framing it by line:

use datum::{Compression, FileIO, Framing, Sink};

let compressed_lines = FileIO::from_path_default("events.log")
    .via(Framing::delimiter(b"\n".to_vec(), 4096, true))
    .via(Compression::gzip())
    .run_with(Sink::collect())
    .unwrap()
    .wait()
    .unwrap();

StreamConverters

StreamConverters wraps any std::io::Read or std::io::Write implementation as a Source<Vec<u8>> or Sink<Vec<u8>, StreamCompletion<NotUsed>>. This is the primary extension point for custom IO: HTTP response bodies, in-memory cursors, pipes, and any other Read/Write all work without extra adapters.

The factory closure is called at materialization, not at blueprint construction. Building the source or sink has no side effects.

StreamConverters::from_reader(factory, chunk_size)  →  Source<Vec<u8>>
    where factory: Fn() -> std::io::Result<R> + Send + Sync + 'static
          R: std::io::Read + Send + 'static

StreamConverters::to_writer(factory)  →  Sink<Vec<u8>, StreamCompletion<NotUsed>>
    where factory: Fn() -> std::io::Result<W> + Send + Sync + 'static
          W: std::io::Write + Send + 'static

use datum::{Sink, StreamConverters};
use std::io::Cursor;

// StreamConverters::from_reader(factory, chunk_size) wraps any std::io::Read.
// The factory is called at materialization (blueprint-safe, not at construction).
let chunks: Vec<Vec<u8>> = StreamConverters::from_reader(
    || Ok(Cursor::new(b"hello world".to_vec())),
    4, // emit at most 4 bytes per chunk
)
.run_with(Sink::collect())
.unwrap()
.wait()
.unwrap();

let combined: Vec<u8> = chunks.into_iter().flatten().collect();
assert_eq!(combined, b"hello world");

The writer sink flushes exactly once — either on stream completion or on error — and drops the writer after flushing. Upstream errors are forwarded to the materialized StreamCompletion.

as_input_stream — blocking read handle from a sink

StreamConverters::as_input_stream(read_timeout) creates a Sink<Vec<u8>, InputStreamHandle>. When materialized, the InputStreamHandle implements std::io::Read — synchronous blocking code calls read() to consume bytes from the Datum stream. This is the reverse direction of from_reader: instead of adapting a Read into a source, it adapts a sink into a Read handle.

• Blocking semantics: read() blocks the calling thread (up to read_timeout) until the stream produces data.
• Partial reads: leftover bytes from a previous chunk are retained and served on the next call, so a single stream chunk can span multiple read() calls.
• EOF: read() returns Ok(0) when the stream completes.
• Errors: stream errors surface as io::Error.
• Cancellation: dropping the InputStreamHandle cancels the stream.

use datum::{Source, StreamConverters};
use std::io::Read;
use std::time::Duration;

let mut handle = Source::from_iter([b"hello".to_vec(), b"world".to_vec()])
    .run_with(StreamConverters::as_input_stream(Duration::from_secs(5)))
    .unwrap();

let mut buf = [0_u8; 32];
let n = handle.read(&mut buf).unwrap();
assert_eq!(&buf[..n], b"helloworld");

as_output_stream — blocking write handle into a source

StreamConverters::as_output_stream(write_timeout) creates a Source<Vec<u8>, OutputStreamHandle>. When materialized, the OutputStreamHandle implements std::io::Write — synchronous blocking code calls write() to inject bytes into the Datum stream. This is the reverse direction of to_writer: instead of adapting a source into a Write, it adapts a source from a Write handle.

• Blocking semantics: write() blocks the calling thread (up to write_timeout) when the stream is not ready, respecting backpressure.
• Stream elements: each write() call produces one Vec<u8> stream element.
• flush() is a no-op (there's no guarantee data has been accepted downstream).
• close() completes the stream, signalling EOF to downstream consumers. After close(), subsequent writes return ErrorKind::BrokenPipe.
• Cancellation: dropping the OutputStreamHandle cancels the stream.

use datum::{Keep, Sink, StreamConverters};
use std::io::Write;
use std::time::Duration;

let (mut handle, completion) = StreamConverters::as_output_stream(Duration::from_secs(5))
    .to_mat(Sink::collect(), Keep::both)
    .run()
    .unwrap();

handle.write_all(b"alpha").unwrap();
handle.write_all(b"beta").unwrap();
handle.close().unwrap();

let chunks = completion.wait().unwrap();
assert_eq!(chunks, vec![b"alpha".to_vec(), b"beta".to_vec()]);

Round-trip pattern

as_output_stream and as_input_stream can be chained together to bridge two blocking threads through a Datum stream with backpressure:

use datum::{InputStreamHandle, Keep, OutputStreamHandle, StreamConverters};
use std::io::{Read, Write};
use std::time::Duration;

let (mut out, mut in_): (OutputStreamHandle, InputStreamHandle) =
    StreamConverters::as_output_stream(Duration::from_secs(5))
        .to_mat(StreamConverters::as_input_stream(Duration::from_secs(5)), Keep::both)
        .run()
        .unwrap();

out.write_all(b"roundtrip").unwrap();
out.close().unwrap();

let mut buf = [0_u8; 16];
let n = in_.read(&mut buf).unwrap();
assert_eq!(&buf[..n], b"roundtrip");

TCP

TokioTcp provides plain TCP server and client sources backed by Tokio's async runtime. TLS, UDP, QUIC, and connection-lifecycle helpers live in the datum-net satellite.

Binding a server

TokioTcp::bind_default(addr) returns a Source<TcpIncomingConnection, StreamCompletion<TcpBinding>>. The materialized TcpBinding is available as soon as the listener is bound — before any connections are accepted. Each element in the source is one accepted TcpIncomingConnection.

use datum::{Keep, Sink, TokioTcp};

// Bind on an OS-assigned port.
let (binding_completion, incoming) = TokioTcp::bind_default("127.0.0.1:0")
    .to_mat(Sink::head(), Keep::both)
    .run()
    .unwrap();

let binding = binding_completion.wait().unwrap();
println!("Listening on {}", binding.local_addr());

// Accept one connection and split it into independent byte source and sink.
let conn = incoming.wait().unwrap();
let (source, sink) = conn.into_parts();
// source: TokioByteSource — reads bytes from the connection
// sink:   TokioByteSink   — writes bytes to the connection

TcpIncomingConnection methods:

Method	Returns
.local_addr()	SocketAddr of the server side
.remote_addr()	SocketAddr of the connected client
.connection()	TcpConnection metadata struct
.into_parts()	(TokioByteSource, TokioByteSink) — independent halves
.into_flow()	Flow<Vec<u8>, Vec<u8>, NotUsed> — coupled echo flow

Outgoing connections

TokioTcp::outgoing_connection_default(addr) returns Flow<Vec<u8>, Vec<u8>, StreamCompletion<TcpConnection>>. The upstream input is written to the socket; bytes received from the socket are emitted downstream. The materialized TcpConnection carries the local and remote addresses.

use datum::{Keep, Sink, Source, TokioTcp};

// Send "ping" to a server and collect whatever it echoes back.
let (conn_completion, response) = Source::single(b"ping".to_vec())
    .via(TokioTcp::outgoing_connection_default("127.0.0.1:8080"))
    .to_mat(Sink::collect(), Keep::both)
    .run()
    .unwrap();

let bytes = response.wait().unwrap();
let conn = conn_completion.wait().unwrap();
println!("Connected {} → {}", conn.local_addr(), conn.remote_addr());

Minimal echo server

use datum::{Keep, Sink, Source, TokioTcp};

// Bind the server.
let (binding_completion, first_incoming) = TokioTcp::bind_default("127.0.0.1:0")
    .to_mat(Sink::head(), Keep::both)
    .run()
    .unwrap();
let binding = binding_completion.wait().unwrap();

// Accept one connection and echo everything back.
let conn = first_incoming.wait().unwrap();
let echo_flow = conn.into_flow();
Source::empty::<Vec<u8>>()
    .via(echo_flow)
    .run_with(Sink::ignore())
    .unwrap()
    .wait()
    .ok();

Performance. TCP echo round-trip (64-byte payload) is competitive to faster than Akka on wall-clock. The consumer thread spins to catch fast completions within the spin window; the spin budget constant is the named lever. Current numbers are maintained in roadmap/benchmarks/streaming-io.md.

Performance summary

All streaming IO paths are at parity or better vs Akka on wall-clock. Notable observations:

• Sync file IO and IO adapter round-trip (as_output_stream → as_input_stream) are significantly faster.
• Tokio file source and TCP echo show a CPU-to-wall gap that is tracked-acceptable (two-thread model and spin budget, respectively).
• Delimiter framing and JSON framing are at parity or better, with delimiter framing also showing significantly lower allocation.

Full per-scenario numbers — including wall-clock µs/op, CPU µs/op, allocation, and Akka comparisons — are the source of truth in roadmap/benchmarks/streaming-io.md.

Next steps

• Futures Interop — integrating async/await with streams
• Execution Model — how the thread pool, fused paths, and spin-then-park interact