mpeg2ts-reader

Rust编写的MPEG2传输流数据读取器

对MPEG传输流中的有效负载数据实现零拷贝访问。

本crate，

• 实现了一个低级状态机，能够识别传输流语法的结构元素
• 提供了你应实现的特性，以定义你应用特定的数据处理。

示例

将H264有效负载数据以十六进制形式输出。

#[macro_use]
extern crate mpeg2ts_reader;
extern crate hex_slice;

use hex_slice::AsHex;
use mpeg2ts_reader::demultiplex;
use mpeg2ts_reader::packet;
use mpeg2ts_reader::pes;
use mpeg2ts_reader::psi;
use mpeg2ts_reader::StreamType;
use std::cmp;
use std::env;
use std::fs::File;
use std::io::Read;

// This macro invocation creates an enum called DumpFilterSwitch, encapsulating all possible ways
// that this application may handle transport stream packets.  Each enum variant is just a wrapper
// around an implementation of the PacketFilter trait
packet_filter_switch! {
    DumpFilterSwitch<DumpDemuxContext> {
        // the DumpFilterSwitch::Pes variant will perform the logic actually specific to this
        // application,
        Pes: pes::PesPacketFilter<DumpDemuxContext,PtsDumpElementaryStreamConsumer>,

        // these definitions are boilerplate required by the framework,
        Pat: demultiplex::PatPacketFilter<DumpDemuxContext>,
        Pmt: demultiplex::PmtPacketFilter<DumpDemuxContext>,

        // this variant will be used when we want to ignore data in the transport stream that this
        // application does not care about
        Null: demultiplex::NullPacketFilter<DumpDemuxContext>,
    }
}

// This macro invocation creates a type called DumpDemuxContext, which is our application-specific
// implementation of the DemuxContext trait.
demux_context!(DumpDemuxContext, DumpFilterSwitch);

// When the de-multiplexing process needs to create a PacketFilter instance to handle a particular
// kind of data discovered within the Transport Stream being processed, it will send a
// FilterRequest to our application-specific implementation of the do_construct() method
impl DumpDemuxContext {
    fn do_construct(&mut self, req: demultiplex::FilterRequest<'_, '_>) -> DumpFilterSwitch {
        match req {
            // The 'Program Association Table' is is always on PID 0.  We just use the standard
            // handling here, but an application could insert its own logic if required,
            demultiplex::FilterRequest::ByPid(packet::Pid::PAT) => {
                DumpFilterSwitch::Pat(demultiplex::PatPacketFilter::default())
            }
            // 'Stuffing' data on PID 0x1fff may be used to pad-out parts of the transport stream
            // so that it has constant overall bitrate.  This causes it to be ignored if present.
            demultiplex::FilterRequest::ByPid(packet::Pid::STUFFING) => {
                DumpFilterSwitch::Null(demultiplex::NullPacketFilter::default())
            }
            // Some Transport Streams will contain data on 'well known' PIDs, which are not
            // announced in PAT / PMT metadata.  This application does not process any of these
            // well known PIDs, so we register NullPacketFiltet such that they will be ignored
            demultiplex::FilterRequest::ByPid(_) => {
                DumpFilterSwitch::Null(demultiplex::NullPacketFilter::default())
            }
            // This match-arm installs our application-specific handling for each H264 stream
            // discovered within the transport stream,
            demultiplex::FilterRequest::ByStream {
                stream_type: StreamType::H264,
                pmt,
                stream_info,
                ..
            } => PtsDumpElementaryStreamConsumer::construct(pmt, stream_info),
            // We need to have a match-arm to specify how to handle any other StreamType values
            // that might be present; we answer with NullPacketFilter so that anything other than
            // H264 (handled above) is ignored,
            demultiplex::FilterRequest::ByStream { .. } => {
                DumpFilterSwitch::Null(demultiplex::NullPacketFilter::default())
            }
            // The 'Program Map Table' defines the sub-streams for a particular program within the
            // Transport Stream (it is common for Transport Streams to contain only one program).
            // We just use the standard handling here, but an application could insert its own
            // logic if required,
            demultiplex::FilterRequest::Pmt {
                pid,
                program_number,
            } => DumpFilterSwitch::Pmt(demultiplex::PmtPacketFilter::new(pid, program_number)),
            // Ignore 'Network Information Table', if present,
            demultiplex::FilterRequest::Nit { .. } => {
                DumpFilterSwitch::Null(demultiplex::NullPacketFilter::default())
            }
        }
    }
}

// Implement the ElementaryStreamConsumer to just dump and PTS/DTS timestamps to stdout
pub struct PtsDumpElementaryStreamConsumer {
    pid: packet::Pid,
    len: Option<usize>,
}
impl PtsDumpElementaryStreamConsumer {
    fn construct(
        _pmt_sect: &psi::pmt::PmtSection,
        stream_info: &psi::pmt::StreamInfo,
    ) -> DumpFilterSwitch {
        let filter = pes::PesPacketFilter::new(PtsDumpElementaryStreamConsumer {
            pid: stream_info.elementary_pid(),
            len: None,
        });
        DumpFilterSwitch::Pes(filter)
    }
}
impl pes::ElementaryStreamConsumer for PtsDumpElementaryStreamConsumer {
    fn start_stream(&mut self) {}
    fn begin_packet(&mut self, header: pes::PesHeader) {
        match header.contents() {
            pes::PesContents::Parsed(Some(parsed)) => {
                match parsed.pts_dts() {
                    Ok(pes::PtsDts::PtsOnly(Ok(pts))) => {
                        print!("{:?}: pts {:#08x}                ", self.pid, pts.value())
                    }
                    Ok(pes::PtsDts::Both {
                        pts: Ok(pts),
                        dts: Ok(dts),
                    }) => print!(
                        "{:?}: pts {:#08x} dts {:#08x} ",
                        self.pid,
                        pts.value(),
                        dts.value()
                    ),
                    _ => (),
                }
                let payload = parsed.payload();
                self.len = Some(payload.len());
                println!(
                    "{:02x}",
                    payload[..cmp::min(payload.len(), 16)].plain_hex(false)
                )
            }
            pes::PesContents::Parsed(None) => (),
            pes::PesContents::Payload(payload) => {
                self.len = Some(payload.len());
                println!(
                    "{:?}:                               {:02x}",
                    self.pid,
                    payload[..cmp::min(payload.len(), 16)].plain_hex(false)
                )
            }
        }
    }
    fn continue_packet(&mut self, data: &[u8]) {
        println!(
            "{:?}:                     continues {:02x}",
            self.pid,
            data[..cmp::min(data.len(), 16)].plain_hex(false)
        );
        self.len = self.len.map(|l| l + data.len());
    }
    fn end_packet(&mut self) {
        println!("{:?}: end of packet length={:?}", self.pid, self.len);
    }
    fn continuity_error(&mut self) {}
}

fn main() {
    // open input file named on command line,
    let name = env::args().nth(1).unwrap();
    let mut f = File::open(&name).expect(&format!("file not found: {}", &name));

    // create the context object that stores the state of the transport stream demultiplexing
    // process
    let mut ctx = DumpDemuxContext::new();

    // create the demultiplexer, which will use the ctx to create a filter for pid 0 (PAT)
    let mut demux = demultiplex::Demultiplex::new(&mut ctx);

    // consume the input file,
    let mut buf = [0u8; 188 * 1024];
    loop {
        match f.read(&mut buf[..]).expect("read failed") {
            0 => break,
            n => demux.push(&mut ctx, &buf[0..n]),
        }
    }
}

性能竞赛

比较此crate与其他几个你可能用于读取传输流的crate —— mpeg2ts 和 ffmpg-sys

产生上述图表数据的基准测试在shootout 文件夹中。（如果基准测试给出了不公正的性能比较，那是一个错误 —— 请提交bug！）

测试条件为，

• 数据已在内存中（无网络/磁盘访问）
• 测试数据集大于CPU缓存
• 处理发生在单个核心上（无流的多进程）

支持的传输流功能

并非所有传输流功能都得到支持。以下是可用功能的摘要以及尚未到来功能

• 封装
  • ISO/IEC 13818-1 188字节包
  • m2ts 192字节包（如果外部crate能支持的话，那会很好）
  • 同步丢失后的恢复
• 传输流包
  • 固定头部
  • 适配字段
  • TS级加扰（除0以外的transport_scrambling_control值）不受支持
• 节目特定信息表
  • 部分语法
  • ‘多部分’表
  • PAT - 程序关联表
  • 节目映射表（PMT）
  • 传输流描述表（TSDT）
• 分组基本流语法
  • PES分组数据
  • PTS/DTS
  • ESCR
  • ES速率
  • DSM技巧模式
  • 附加复制信息
  • PES循环冗余校验（CRC）
  • PES扩展
• 描述符
  • 视频流描述符
  • 音频流描述符
  • 层次描述符
  • 注册描述符
  • 数据流对齐描述符
  • 目标背景网格描述符
  • 视频窗口描述符
  • 条件接收（CA）描述符
  • ISO 639语言描述符
  • 系统时钟描述符
  • 复用缓冲区利用率描述符
  • 版权描述符
  • 最大比特率描述符
  • 私有数据指示符描述符
  • 平滑缓冲区描述符
  • 标准描述符
  • IBP描述符
  • MPEG4视频描述符
  • MPEG4音频描述符
  • IOD描述符
  • SL描述符
  • FMC描述符
  • 外部ES ID描述符
  • MUX码描述符
  • FMX缓冲区大小描述符
  • 复用缓冲区描述符
  • AVC视频描述符