Qubit IO

Qubit IO provides a generic input/output stream system for Rust, built around Input and Output. Their associated Item type extends stream I/O from bytes to characters, business records, and other Rust types. Library and application authors can write transfer and processing algorithms directly over those items, without serializing them at each processing boundary.

Standard Read and Write values integrate with this system automatically. Buffers, limits, counters, checksums, digests, and tee wrappers compose around the stream interfaces. AsyncInput and AsyncOutput extend the model to runtime-neutral asynchronous I/O, with optional Tokio and futures-io adapters.

Core model: streams of items

An input reads items into a slice; an output writes items from a slice. With Item = Sale, a read count of 10 means ten sales records. With Item = u8, it means ten bytes. Limits and counters follow the same item unit.

The core traits do not require Item: Clone + Default; individual adapters and wrappers impose the bounds they need. Generic buffering requires Clone + Default, while checksum and digest wrappers operate on bytes.

Installation

Requires Rust 1.94 or later. The default feature set enables no async ecosystem adapter or digest dependency.

[dependencies]
qubit-io = "0.19"

Quick start: process business records

Consider a sales pipeline that emits category revenue records. Encoding every record into bytes between operators adds a serialization boundary where the algorithm only needs typed data. Instead, the mapper accepts a sales input and a revenue output. Its caller selects the source, destination, and policies:

sales source -> limit (2 records) -> buffer -> counter -> mapper -> counter -> revenue sink

This complete program uses application-defined in-memory record adapters. Qubit IO supplies the traits, buffer, limit, counters, and checked read/write helpers:

use std::io;
use qubit_io::{
    BufferedInput, CountingInput, CountingOutput, Input, LimitInput, Output,
};

#[derive(Clone, Debug, Default, PartialEq)]
struct Sale {
    category: u16,
    cents: u64,
}

#[derive(Clone, Debug, Default, PartialEq)]
struct Revenue {
    category: u16,
    cents: u64,
}

// Application adapters: these transfer whole records, not encoded bytes.
struct RecordInput<T> {
    records: Vec<T>,
    position: usize,
}

impl<T: Clone> Input for RecordInput<T> {
    type Item = T;

    // SAFETY: The caller must provide a valid destination range.
    unsafe fn read_unchecked(
        &mut self, output: &mut [T], index: usize, count: usize,
    ) -> io::Result<usize> {
        let n = count.min(self.records.len() - self.position);
        output[index..index + n]
            .clone_from_slice(&self.records[self.position..self.position + n]);
        self.position += n;
        Ok(n)
    }
}

struct RecordOutput<T>(Vec<T>);

impl<T: Clone> Output for RecordOutput<T> {
    type Item = T;

    // SAFETY: The caller must provide a valid source range.
    unsafe fn write_unchecked(
        &mut self, input: &[T], index: usize, count: usize,
    ) -> io::Result<usize> {
        self.0.extend_from_slice(&input[index..index + count]);
        Ok(count)
    }

    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
}

fn map_sales<I, O>(input: &mut I, output: &mut O) -> io::Result<()>
where
    I: Input<Item = Sale>,
    O: Output<Item = Revenue>,
{
    let mut batch = [Sale::default(); 1];
    while input.read(&mut batch)? != 0 {
        let sale = &batch[0];
        output.write_fully(&[Revenue {
            category: sale.category,
            cents: sale.cents,
        }])?;
    }
    Ok(())
}

fn main() -> io::Result<()> {
    let source = RecordInput {
        records: vec![
            Sale { category: 7, cents: 300 },
            Sale { category: 7, cents: 500 },
            Sale { category: 9, cents: 900 },
        ],
        position: 0,
    };
    let limited = LimitInput::new(source, 2);
    let buffered = BufferedInput::with_capacity(limited, 2);
    let mut input = CountingInput::new(buffered);
    let mut output = CountingOutput::new(RecordOutput(Vec::new()));

    map_sales(&mut input, &mut output)?;
    output.flush()?;
    assert_eq!(input.items_read(), 2);
    assert_eq!(output.items_written(), 2);
    assert_eq!(output.into_inner().0, vec![
        Revenue { category: 7, cents: 300 },
        Revenue { category: 7, cents: 500 },
    ]);
    Ok(())
}

The assertions check both the resulting records and the transfer counts: the third sale is excluded by the limit. No record serialization occurs in this pipeline. The mapper uses safe methods; the adapters implement the low-level indexed transfer contract, whose unsafe methods require valid slice ranges.

A real application supplies adapters for its own record sources and sinks; Qubit IO does not provide a record storage engine. The typed-record example also mirrors output records to two sinks with TeeOutput.

Shorter bounds for byte streams

Use ByteInput and ByteOutput when an API transfers bytes. They provide shorter bounds for Input<Item = u8> and Output<Item = u8>, respectively, with the same inherited operations:

use std::io::{self, Cursor};
use qubit_io::{ByteInput, ByteOutput};

fn input() -> impl ByteInput {
    Cursor::new(b"QBIT")
}

fn copy_magic<I: ByteInput + ?Sized, O: ByteOutput + ?Sized>(
    input: &mut I, output: &mut O,
) -> io::Result<()> {
    let mut magic = [0; 4];
    input.read_exactly(&mut magic)?;
    output.write_fully(&magic)
}

fn main() -> io::Result<()> {
    let mut bytes = Vec::new();
    copy_magic(&mut input(), &mut bytes)?;
    assert_eq!(bytes, b"QBIT");
    Ok(())
}

Existing byte inputs and outputs automatically satisfy these traits, including standard Read and Write implementations. Implementers still implement Input or Output; no separate byte-trait implementation is needed. Return impl ByteInput / impl ByteOutput to hide a concrete type, or use dyn ByteInput / dyn ByteOutput behind a reference or box for dynamic dispatch. See the byte-stream guide for borrowed outputs and standard-library conversion.

The same short-bound pattern covers asynchronous bytes and Unicode scalar values:

Each trait automatically applies to implementations of its corresponding base trait with the matching Item. For example, use I: CharInput instead of I: Input<Item = char>, or O: AsyncByteOutput instead of O: AsyncOutput<Item = u8>. Opaque return types use impl CharInput or impl AsyncByteOutput in the same way. Character streams transfer Rust char values (Unicode scalar values), without encoding or decoding text. Async polling still uses Pin and Poll, and does not require Unpin.

Compatible with standard Read and Write

Every std::io::Read value already implements Input<Item = u8>, and every std::io::Write value implements Output<Item = u8>. Files, sockets, Cursor, and Vec<u8> can therefore participate without an extra input/output adapter. In the reverse direction, StdRead and StdWrite expose Qubit byte streams through the standard traits:

use std::io::{self, Cursor, Read};
use qubit_io::{CountingInput, Input, Output, StdRead};

fn read_magic<I: Input<Item = u8>>(input: &mut I) -> io::Result<[u8; 4]> {
    let mut magic = [0; 4];
    input.read_exactly(&mut magic)?;
    Ok(magic)
}

fn main() -> io::Result<()> {
    let mut bytes = Vec::new();
    bytes.write_fully(b"QBITpayload")?;
    let mut input = CountingInput::new(Cursor::new(bytes));
    assert_eq!(read_magic(&mut input)?, *b"QBIT");
    assert_eq!(input.bytes_read(), 4);

    // Present the remaining Qubit input to an API that expects std::io::Read.
    let mut reader = StdRead::new(input);
    let mut rest = String::new();
    reader.read_to_string(&mut rest)?;
    assert_eq!(rest, "payload");
    assert_eq!(reader.get_ref().bytes_read(), 11);
    Ok(())
}

InputRef and OutputRef support borrowing an existing Qubit stream, including a trait object, before conversion. Standard-library compatibility applies to byte streams; converting typed records to bytes still requires an encoding.

Compose stream wrappers

A wrapper implements the corresponding stream interface, so algorithms can accept a wrapped stream without knowing which policies surround it. The quick start's CountingInput<BufferedInput<LimitInput<_>>> is still an Input<Item = Sale>.

Tee wrappers preserve progress already reported by the primary stream. An Ok(n) therefore confirms n primary items, not a complete mirror. Check branch_error() after finite input transfers and check output flush(); all three into_parts() methods now return (primary, branch, error). A branch failure does not roll back the primary, so do not blindly replay its prefix. The user guide covers this recovery boundary.

Wrapper order determines what is observed. A counter outside an input buffer counts items consumed by the algorithm; a counter inside it also counts prefetched items. A limit bounds the items exposed or accepted by its wrapper; it does not replace application validation such as checking a message's declared length before allocation. See the bounded-frame example.

Flush buffered output explicitly and check the result. Dropping synchronous buffered output only attempts a best-effort drain; dropping asynchronous buffered output cannot perform asynchronous I/O. The user guide explains wrapper order, partial progress, and recovery of unread or pending items.

Synchronous buffered output exposes recovery as Pending or Indeterminate. The latter follows a panic during inner I/O; check external side effects before deciding whether any retained items can be replayed. See the user guide for the recovery path and its limits.

Content digests

For a SHA-256 fingerprint, enable digest and add a compatible algorithm:

[dependencies]
qubit-io = { version = "0.19", features = ["digest"] }
sha2 = "0.11"

use std::io::{self, Cursor};
use qubit_io::DigestInput;
use sha2::{Digest, Sha256};

fn main() -> io::Result<()> {
    let mut input = Cursor::new(b"hello world");
    let fingerprint = DigestInput::<_, Sha256>::digest_all(&mut input)?;
    assert_eq!(fingerprint, Sha256::digest(b"hello world"));
    Ok(())
}

digest_all reads from the current position to EOF using a fixed-size buffer; it fingerprints the entire original input only when started at its beginning. An Interrupted read is retried without losing the digest accumulated from earlier reads. Other read errors are returned without a partial digest, and the input remains at the position already reached. For incremental tracking, wrap a byte stream with DigestInput::new or DigestOutput::new. Algorithms must use the compatible digest 0.11 interface; Qubit IO does not choose the algorithm for the application.

Digest wrappers track successful transfers in order, including repeated reads or overwriting writes after seeking. Their result is not automatically a fingerprint of final file contents. finalize() does not read to EOF, flush, or close. See the digest guide.

Asynchronous input and output

AsyncInput and AsyncOutput retain the generic item model and expose readiness through Pin and Poll. Callers normally use helpers such as read_exactly_async and write_fully_async; pinned !Unpin values and trait objects use PinnedAsyncInputExt and PinnedAsyncOutputExt.

An asynchronous algorithm can accept either Tokio or futures-io sources via adapters. For example, this complete program runs a generic async reader with Tokio:

[dependencies]
qubit-io = { version = "0.19", features = ["tokio"] }
tokio = { version = "1", features = ["macros", "rt", "io-util"] }

use std::io;
use qubit_io::{AsyncCountingInput, AsyncInput, TokioInput};

async fn read_magic<I>(input: &mut I) -> io::Result<[u8; 4]>
where
    I: AsyncInput<Item = u8> + Unpin,
{
    let mut magic = [0; 4];
    input.read_exactly_async(&mut magic).await?;
    Ok(magic)
}

#[tokio::main(flavor = "current_thread")]
async fn main() -> io::Result<()> {
    let source = TokioInput::new(&b"QBIT"[..]);
    let mut input = AsyncCountingInput::new(source);
    assert_eq!(read_magic(&mut input).await?, *b"QBIT");
    assert_eq!(input.bytes_read(), 4);
    Ok(())
}

The algorithm depends on AsyncInput, while the application selects Tokio. A futures-io caller can pass FuturesInput to the same function. Synchronous and asynchronous algorithms have separate entry points.

AsyncClose is a separate capability for graceful output closure, including any required buffered output and underlying close operation. Flushing alone does not close a transport. The user guide covers polling, pinning, and cancellation behavior.

Other stream tools

Scope and when to use it

Use Qubit IO when an algorithm should operate on typed item streams, transfer policies need to compose independently, or an async library must leave runtime selection to its caller. An application using only ordinary byte streams and one runtime may need only the native I/O traits.

The crate provides stream transfer and related tools. Binary values and text encodings belong to qubit-io-binary and qubit-io-text, respectively. Paths, file identity, metadata, commit, abort, and persistence belong to a higher-level file-system abstraction such as qubit-fs.

Features and documentation

• API documentation — built with all declared features enabled.
• User guide | 中文用户指南
• Typed-record example | bounded-frame example
• 中文 README

Testing

# Run tests with the default feature set
cargo test

# Run tests with all declared features
cargo test --all-features

# Project CI checks
./.infra/bin/ci-check.sh

# Check code coverage
./.infra/bin/coverage.sh

License

Copyright (c) 2025 - 2026. Haixing Hu. All rights reserved.

Licensed under the Apache License, Version 2.0. See LICENSE for the full license text.

Contributing

Contributions are welcome. Please follow the Rust API guidelines, keep public API documentation and tests current, and run ./.infra/bin/align-ci.sh to format code and ./.infra/bin/ci-check.sh to satisfy CI requirements before submitting a pull request.

Author

Haixing Hu - Qubit Co. Ltd.

Repository: https://github.com/qubit-ltd/rs-io