flowly-mpegts 0.2.2

Mpeg2Ts format streaming parser
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133

use bytes::{BufMut, BytesMut};
use flowly::Fourcc;
use std::collections::HashMap;

use bytes::Buf;
use mpeg2ts::{
    es::StreamType,
    ts::{ContinuityCounter, Pid, ReadTsPacket, TsPacket, TsPacketReader, TsPayload},
};

use crate::error::Error;

pub struct Mpeg2TsDemux {
    ignore: bool,
    mappings: Option<HashMap<Pid, PidKind>>,
    base_ts: u64,

    #[allow(dead_code)]
    cc: ContinuityCounter,
    fragm_idx: u64,
    frame_idx: u64,
    codec: Fourcc,
    is_keyframe: bool,
    is_params_updated: bool,
    tags: BytesMut,
    pts: u64,
    body: BytesMut,
}

impl Mpeg2TsDemux {
    pub fn new(base_ts: u64, fragm_idx: u64) -> Self {
        Self {
            ignore: true,
            mappings: None,
            base_ts,
            fragm_idx,
            frame_idx: 0,
            is_params_updated: false,
            tags: BytesMut::with_capacity(64 * 1024 * 1024),
            is_keyframe: false,
            codec: Fourcc::default(),
            pts: 0,
            cc: ContinuityCounter::new(),
            body: BytesMut::with_capacity(64 * 1024 * 1024),
        }
    }
}

impl Mpeg2TsDemux {
    fn parse(&mut self, src: &mut BytesMut) -> Result<Option<Mpeg2TsFrame>, Error> {
        let mut reader = if let Some(map) = self.mappings.take() {
            TsPacketReader::with_mappings(src.reader(), map)
        } else {
            TsPacketReader::new(src.reader())
        };

        while reader.stream().get_ref().remaining() >= TsPacket::SIZE {
            let pkt = reader.read_ts_packet()?.unwrap();
            let is_keyframe = pkt
                .adaptation_field
                .as_ref()
                .map(|x| x.random_access_indicator)
                .unwrap_or(false);

            if let Some(p) = pkt.payload {
                match p {
                    TsPayload::Pat(_pat) => {}
                    TsPayload::Pmt(pmt) => {
                        for es in pmt.es_info {
                            match es.stream_type {
                                StreamType::H264 => self.codec = Fourcc::VIDEO_AVC,
                                StreamType::H265 => self.codec = Fourcc::VIDEO_HEVC,
                                _ => (),
                            }
                        }
                    }

                    TsPayload::Pes(pes) => {
                        let frame = (!self.ignore).then(|| {
                            Frame::new(
                                self.codec,
                                self.fragm_idx,
                                self.frame_idx,
                                self.pts,
                                self.pts + self.base_ts,
                                self.is_keyframe,
                                self.is_params_updated,
                                self.tags.clone().into(),
                                true,
                                (0, 0),
                                self.source.clone(),
                                None::<Vec<Vec<u8>>>,
                                false,
                                self.body.to_vec(),
                            )
                        });

                        if pes.header.stream_id.as_u8() != PES_VIDEO_STREAM_ID {
                            self.ignore = true;
                            continue;
                        } else {
                            self.ignore = false;
                        }

                        self.pts = (pes.header.pts.unwrap().as_u64() * 1_000_000) / 90_000;
                        self.is_keyframe = is_keyframe;

                        self.body.clear();
                        self.body.put_slice(&pes.data);

                        if frame.is_some() {
                            self.frame_idx += 1;
                            self.mappings.replace(reader.into_mappings());
                            return Ok(frame);
                        }
                    }

                    TsPayload::Section(_) => println!("section"),
                    TsPayload::Null(_) => println!("null"),
                    TsPayload::Raw(raw) => {
                        if !self.ignore {
                            self.body.put_slice(&raw);
                        }
                    }
                }
            }
        }

        self.mappings.replace(reader.into_mappings());

        Ok(None)
    }
}