Crate dvb_stream 

Async/tokio stream adapters for DVB SI and T2-MI processing.

This crate wraps the synchronous dvb_si::demux::SiDemux and dvb_t2mi::pump::T2miPump as futures_core::Stream implementations, quarantining tokio and futures-core away from the parser crates.

Streams

• SectionStream — wraps dvb_si::demux::SiDemux over any tokio::io::AsyncRead byte source (file, TCP socket). Each item is an owned dvb_si::demux::SectionEvent ('static, no borrow of the read buffer).

• T2miEventStream — wraps dvb_t2mi::pump::T2miPump over any tokio::io::AsyncRead. Each item is an owned dvb_t2mi::pump::T2miEvent.

Both streams are also constructable from a UDP multicast socket (the dominant real-world DVB transport) via the bind_multicast constructor when the udp feature is enabled.

Ownership and cancellation

The adapter owns the read buffer and feeds bytes into the synchronous pump on each poll_next call. Events are buffered in a small per-packet queue and drained before the next read is attempted. There are no internal tasks or spawning; cancellation is simply dropping the stream.

188-byte TS framing and resync

The adapter reads raw bytes from the AsyncRead source and performs 188-byte TS packet alignment via a sync-byte (0x47) resync on the read buffer. The resync logic is implemented once in resync and shared by both streams.

Feature flags

Feature	Default	Description
udp	on	UDP/multicast constructors (bind_multicast) via tokio::net::UdpSocket.

MSRV

dvb-stream 1.86 (mirrors the workspace). This crate is versioned and released independently from the dvb-si / dvb-t2mi lockstep because tokio’s own MSRV moves faster.

Examples

Two runnable examples ship with this crate (cargo run -p dvb-stream --example <name>).

count_sections

//! Basic: drive the async `SectionStream` over an in-memory TS and count the
//! SI sections it yields.
//!
//! Run with: `cargo run -p dvb-stream --example count_sections`
//!
//! Reads the committed `m6-single.ts` fixture (from the sibling `dvb-si` crate)
//! into memory and feeds it through the stream via a `Cursor`.

use dvb_stream::SectionStream;
use futures_util::StreamExt;

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let path = concat!(env!("CARGO_MANIFEST_DIR"), "/../fixtures/ts/m6-single.ts");
    let data = match std::fs::read(path) {
        Ok(b) => b,
        Err(e) => {
            eprintln!("fixture not available ({e}); nothing to do");
            return;
        }
    };

    let reader = tokio::io::BufReader::new(std::io::Cursor::new(data));
    let mut stream = SectionStream::new(reader);

    let mut sections = 0u32;
    while let Some(_event) = stream.next().await {
        sections += 1;
    }

    println!("section events: {sections}");
}

stream_stats

//! Advanced: stream SI sections asynchronously, tally the table types, and
//! report the demux + resync statistics.
//!
//! Run with: `cargo run -p dvb-stream --example stream_stats`

use dvb_si::tables::AnyTableSection;
use dvb_stream::SectionStream;
use futures_util::StreamExt;
use std::collections::BTreeMap;

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let path = concat!(env!("CARGO_MANIFEST_DIR"), "/../fixtures/ts/m6-single.ts");
    let data = match std::fs::read(path) {
        Ok(b) => b,
        Err(e) => {
            eprintln!("fixture not available ({e}); nothing to do");
            return;
        }
    };

    let reader = tokio::io::BufReader::new(std::io::Cursor::new(data));
    let mut stream = SectionStream::new(reader);

    let mut tables: BTreeMap<&'static str, u32> = BTreeMap::new();
    while let Some(event) = stream.next().await {
        let name = match event.table_section() {
            Ok(AnyTableSection::PatSection(_)) => "PAT",
            Ok(AnyTableSection::PmtSection(_)) => "PMT",
            Ok(AnyTableSection::SdtSection(_)) => "SDT",
            Ok(AnyTableSection::NitSection(_)) => "NIT",
            Ok(AnyTableSection::EitSection(_)) => "EIT",
            Ok(_) => "other",
            Err(_) => "malformed",
        };
        *tables.entry(name).or_default() += 1;
    }

    println!("table sections:");
    for (name, n) in &tables {
        println!("  {name:<10} {n}");
    }

    let stats = stream.stats();
    let resync = stream.resync_stats();
    println!("\ndemux  : {stats:?}");
    println!(
        "resync : {} resyncs, {} bytes discarded",
        resync.resyncs, resync.bytes_discarded
    );
}

Re-exports

pub use section_stream::SectionStream;

pub use t2mi_stream::T2miEventStream;

Modules

resync

TS byte-stream resynchronisation helpers for dvb-stream.

section_stream

SectionStream — async futures_core::Stream of owned SI section events.

t2mi_stream

T2miEventStream — async futures_core::Stream of owned T2-MI events.

Structs

ResyncStats

Statistics tracking resynchronisation events and discarded bytes in a TS byte stream.