closedcaption: Add SCC parser and encoder plugins

gst-plugin-closedcaption/Cargo.toml

@@ -15,17 +15,17 @@ lazy_static = "1.2"
 
 [dependencies.gst]
 git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs"
-features = ["subclassing", "v1_10"]
+features = ["subclassing", "v1_12"]
 package="gstreamer"
 
 [dependencies.gst-base]
 git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs"
-features = ["subclassing", "v1_10"]
+features = ["subclassing", "v1_12"]
 package="gstreamer-base"
 
 [dependencies.gst-video]
 git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs"
-features = ["v1_10"]
+features = ["v1_12"]
 package="gstreamer-video"
 
 [dev-dependencies]

gst-plugin-closedcaption/src/lib.rs

@@ -37,10 +37,15 @@ mod line_reader;
 mod mcc_enc;
 mod mcc_parse;
 mod mcc_parser;
+mod scc_enc;
+mod scc_parse;
+mod scc_parser;
 
 fn plugin_init(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
     mcc_parse::register(plugin)?;
     mcc_enc::register(plugin)?;
+    scc_parse::register(plugin)?;
+    scc_enc::register(plugin)?;
     Ok(())
 }
 

gst-plugin-closedcaption/src/scc_enc.rs

@@ -0,0 +1,451 @@
+// Copyright (C) 2019 Sebastian Dröge <sebastian@centricular.com>
+// Copyright (C) 2019 Jordan Petridis <jordan@centricular.com>
+//
+// This library is free software; you can redistribute it and/or
+// modify it under the terms of the GNU Library General Public
+// License as published by the Free Software Foundation; either
+// version 2 of the License, or (at your option) any later version.
+//
+// This library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+// Library General Public License for more details.
+//
+// You should have received a copy of the GNU Library General Public
+// License along with this library; if not, write to the
+// Free Software Foundation, Inc., 51 Franklin Street, Suite 500,
+// Boston, MA 02110-1335, USA.
+
+use glib;
+use glib::prelude::*;
+use glib::subclass;
+use glib::subclass::prelude::*;
+use gst;
+use gst::prelude::*;
+use gst::subclass::prelude::*;
+use gst_video::{self, ValidVideoTimeCode};
+
+use std::io::Write;
+use std::sync::Mutex;
+
+lazy_static! {
+    static ref CAT: gst::DebugCategory = {
+        gst::DebugCategory::new(
+            "sccenc",
+            gst::DebugColorFlags::empty(),
+            "Scc Encoder Element",
+        )
+    };
+}
+
+#[derive(Debug)]
+struct State {
+    need_headers: bool,
+    expected_timecode: Option<ValidVideoTimeCode>,
+    internal_buffer: Vec<gst::Buffer>,
+    framerate: Option<gst::Fraction>,
+}
+
+impl Default for State {
+    fn default() -> Self {
+        Self {
+            need_headers: true,
+            expected_timecode: None,
+            internal_buffer: Vec::with_capacity(64),
+            framerate: None,
+        }
+    }
+}
+
+impl State {
+    // Write the header to the buffer and set need_headers to false
+    fn generate_headers(&mut self, buffer: &mut Vec<u8>) {
+        assert!(buffer.is_empty());
+        self.need_headers = false;
+        buffer.extend_from_slice(b"Scenarist_SCC V1.0\r\n\r\n");
+    }
+
+    fn encode_payload(outbuf: &mut Vec<u8>, slice: &[u8]) {
+        write!(outbuf, "{:02x}{:02x}", slice[0], slice[1]).unwrap();
+    }
+
+    fn generate_caption(
+        &mut self,
+        element: &gst::Element,
+        buffer: gst::Buffer,
+    ) -> Result<Option<gst::Buffer>, gst::FlowError> {
+        // Arbitrary number that was chosen to keep in order
+        // to batch pushes of smaller buffers
+        const MAXIMUM_PACKETES_PER_LINE: usize = 16;
+
+        assert!(self.internal_buffer.len() < MAXIMUM_PACKETES_PER_LINE);
+
+        if buffer.get_size() != 2 {
+            gst_element_error!(
+                element,
+                gst::StreamError::Format,
+                ["Wrongly sized CEA608 packet: {}", buffer.get_size()]
+            );
+
+            return Err(gst::FlowError::Error);
+        };
+
+        let mut timecode = buffer
+            .get_meta::<gst_video::VideoTimeCodeMeta>()
+            .ok_or_else(|| {
+                gst_element_error!(
+                    element,
+                    gst::StreamError::Format,
+                    ["Stream with timecodes on each buffer required"]
+                );
+
+                // If we neeed to skip a buffer, increment the frame if it exists
+                // to avoid getting out of sync
+                if let Some(ref mut timecode) = self.expected_timecode {
+                    timecode.increment_frame();
+                }
+
+                gst::FlowError::Error
+            })?
+            .get_tc();
+
+        if self.expected_timecode.is_none() {
+            self.expected_timecode = Some(timecode.clone());
+        }
+
+        // if the timecode is different from the expected one,
+        // flush the previous line into the buffer, and push
+        // the new packet to the, now empty, internal buffer
+        if Some(&timecode) != self.expected_timecode.as_ref() {
+            let res = self.write_line(element);
+            assert!(self.internal_buffer.is_empty());
+            self.internal_buffer.push(buffer);
+
+            timecode.increment_frame();
+            self.expected_timecode = Some(timecode);
+            return res;
+        } else if let Some(ref mut timecode) = self.expected_timecode {
+            timecode.increment_frame();
+        }
+
+        self.internal_buffer.push(buffer);
+
+        if self.internal_buffer.len() == MAXIMUM_PACKETES_PER_LINE {
+            return self.write_line(element);
+        }
+
+        Ok(None)
+    }
+
+    // Flush the internal buffers into a line
+    fn write_line(
+        &mut self,
+        element: &gst::Element,
+    ) -> Result<Option<gst::Buffer>, gst::FlowError> {
+        let mut outbuf = Vec::new();
+        let mut line_start = true;
+
+        assert!(!self.internal_buffer.is_empty());
+
+        if self.need_headers {
+            self.generate_headers(&mut outbuf);
+        }
+
+        let first_buf = self.internal_buffer.first().unwrap();
+        for buffer in self.internal_buffer.iter() {
+            let map = buffer.map_readable().ok_or_else(|| {
+                gst_element_error!(
+                    element,
+                    gst::StreamError::Format,
+                    ["Failed to map buffer readable"]
+                );
+
+                gst::FlowError::Error
+            })?;
+
+            // If its the first packet in the line, write the timecode first
+            // else, separate the packets with a space
+            if line_start {
+                let timecode = buffer
+                    .get_meta::<gst_video::VideoTimeCodeMeta>()
+                    // Checked already before the buffer has been pushed to the
+                    // internal_buffer
+                    .expect("Buffer without timecode")
+                    .get_tc();
+
+                let _ = write!(outbuf, "{}\t", timecode);
+                line_start = false;
+            } else {
+                outbuf.push(b' ');
+            }
+
+            Self::encode_payload(&mut outbuf, &*map);
+        }
+
+        outbuf.extend_from_slice(b"\r\n\r\n".as_ref());
+
+        let buffer = {
+            let mut buffer = gst::Buffer::from_mut_slice(outbuf);
+            let buf_mut = buffer.get_mut().unwrap();
+
+            // Something is seriously wrong else
+            assert!(self.framerate.is_some());
+            let framerate = self.framerate.unwrap();
+
+            let dur = gst::SECOND
+                .mul_div_floor(
+                    self.internal_buffer.len() as u64 * *framerate.denom() as u64,
+                    *framerate.numer() as u64,
+                )
+                .unwrap_or(gst::CLOCK_TIME_NONE);
+            buf_mut.set_duration(dur);
+
+            // Copy the metadata of the first buffer
+            first_buf
+                .copy_into(buf_mut, *gst::BUFFER_COPY_METADATA, 0, None)
+                .expect("Failed to copy buffer metadata");
+            buf_mut.set_pts(first_buf.get_pts());
+            buffer
+        };
+
+        // Clear the internal buffer
+        drop(first_buf);
+        self.internal_buffer.clear();
+
+        Ok(Some(buffer))
+    }
+}
+
+struct SccEnc {
+    srcpad: gst::Pad,
+    sinkpad: gst::Pad,
+    state: Mutex<State>,
+}
+
+impl SccEnc {
+    fn set_pad_functions(sinkpad: &gst::Pad, srcpad: &gst::Pad) {
+        sinkpad.set_chain_function(|pad, parent, buffer| {
+            SccEnc::catch_panic_pad_function(
+                parent,
+                || Err(gst::FlowError::Error),
+                |enc, element| enc.sink_chain(pad, element, buffer),
+            )
+        });
+        sinkpad.set_event_function(|pad, parent, event| {
+            SccEnc::catch_panic_pad_function(
+                parent,
+                || false,
+                |enc, element| enc.sink_event(pad, element, event),
+            )
+        });
+
+        srcpad.set_event_function(|pad, parent, event| {
+            SccEnc::catch_panic_pad_function(
+                parent,
+                || false,
+                |enc, element| enc.src_event(pad, element, event),
+            )
+        });
+        srcpad.set_query_function(|pad, parent, query| {
+            SccEnc::catch_panic_pad_function(
+                parent,
+                || false,
+                |enc, element| enc.src_query(pad, element, query),
+            )
+        });
+    }
+
+    fn sink_chain(
+        &self,
+        pad: &gst::Pad,
+        element: &gst::Element,
+        buffer: gst::Buffer,
+    ) -> Result<gst::FlowSuccess, gst::FlowError> {
+        gst_log!(CAT, obj: pad, "Handling buffer {:?}", buffer);
+
+        let mut state = self.state.lock().unwrap();
+        let res = state.generate_caption(element, buffer)?;
+
+        if let Some(outbuf) = res {
+            gst_trace!(CAT, obj: pad, "Pushing buffer {:?} to the pad", &outbuf);
+
+            drop(state);
+            self.srcpad.push(outbuf)?;
+        };
+
+        Ok(gst::FlowSuccess::Ok)
+    }
+
+    fn sink_event(&self, pad: &gst::Pad, element: &gst::Element, event: gst::Event) -> bool {
+        use gst::EventView;
+
+        gst_log!(CAT, obj: pad, "Handling event {:?}", event);
+
+        match event.view() {
+            EventView::Caps(ev) => {
+                let caps = ev.get_caps();
+                let s = caps.get_structure(0).unwrap();
+                let framerate = s.get::<gst::Fraction>("framerate");
+                if framerate.is_none() {
+                    gst_error!(CAT, obj: pad, "Caps without framerate");
+                    return false;
+                };
+
+                let mut state = self.state.lock().unwrap();
+                state.framerate = framerate;
+
+                // We send our own caps downstream
+                let caps = gst::Caps::builder("application/x-scc").build();
+                self.srcpad.push_event(gst::Event::new_caps(&caps).build())
+            }
+            EventView::Eos(_) => {
+                let mut state = self.state.lock().unwrap();
+
+                let outbuf = state.write_line(element);
+
+                if let Ok(Some(buffer)) = outbuf {
+                    gst_trace!(CAT, obj: pad, "Pushing buffer {:?} to the pad", &buffer);
+
+                    drop(state);
+                    if self.srcpad.push(buffer).is_err() {
+                        gst_error!(CAT, obj: pad, "Failed to push buffer to the pad");
+                        return false;
+                    }
+                } else {
+                    gst_error!(CAT, obj: pad, "Failed to write a line after EOS");
+                    return false;
+                }
+                pad.event_default(element, event)
+            }
+            _ => pad.event_default(element, event),
+        }
+    }
+
+    fn src_event(&self, pad: &gst::Pad, element: &gst::Element, event: gst::Event) -> bool {
+        use gst::EventView;
+
+        gst_log!(CAT, obj: pad, "Handling event {:?}", event);
+        match event.view() {
+            EventView::Seek(_) => {
+                gst_log!(CAT, obj: pad, "Dropping seek event");
+                false
+            }
+            _ => pad.event_default(element, event),
+        }
+    }
+
+    fn src_query(&self, pad: &gst::Pad, element: &gst::Element, query: &mut gst::QueryRef) -> bool {
+        use gst::QueryView;
+
+        gst_log!(CAT, obj: pad, "Handling query {:?}", query);
+
+        match query.view_mut() {
+            QueryView::Seeking(mut q) => {
+                // We don't support any seeking at all
+                let fmt = q.get_format();
+                q.set(
+                    false,
+                    gst::GenericFormattedValue::Other(fmt, -1),
+                    gst::GenericFormattedValue::Other(fmt, -1),
+                );
+                true
+            }
+            _ => pad.query_default(element, query),
+        }
+    }
+}
+
+impl ObjectSubclass for SccEnc {
+    const NAME: &'static str = "RsSccEnc";
+    type ParentType = gst::Element;
+    type Instance = gst::subclass::ElementInstanceStruct<Self>;
+    type Class = subclass::simple::ClassStruct<Self>;
+
+    glib_object_subclass!();
+
+    fn new_with_class(klass: &subclass::simple::ClassStruct<Self>) -> Self {
+        let templ = klass.get_pad_template("sink").unwrap();
+        let sinkpad = gst::Pad::new_from_template(&templ, "sink");
+        let templ = klass.get_pad_template("src").unwrap();
+        let srcpad = gst::Pad::new_from_template(&templ, "src");
+
+        SccEnc::set_pad_functions(&sinkpad, &srcpad);
+
+        Self {
+            srcpad,
+            sinkpad,
+            state: Mutex::new(State::default()),
+        }
+    }
+
+    fn class_init(klass: &mut subclass::simple::ClassStruct<Self>) {
+        klass.set_metadata(
+            "Scc Encoder",
+            "Encoder/ClosedCaption",
+            "Encodes SCC Closed Caption Files",
+            "Sebastian Dröge <sebastian@centricular.com>, Jordan Petridis <jordan@centricular.com>",
+        );
+
+        let framerates =
+            gst::List::new(&[&gst::Fraction::new(30000, 1001), &gst::Fraction::new(30, 1)]);
+        let caps = gst::Caps::builder("closedcaption/x-cea-608")
+            .field("format", &"raw")
+            .field("framerate", &framerates)
+            .build();
+        let sink_pad_template = gst::PadTemplate::new(
+            "sink",
+            gst::PadDirection::Sink,
+            gst::PadPresence::Always,
+            &caps,
+        )
+        .unwrap();
+        klass.add_pad_template(sink_pad_template);
+
+        let caps = gst::Caps::builder("application/x-scc").build();
+        let src_pad_template = gst::PadTemplate::new(
+            "src",
+            gst::PadDirection::Src,
+            gst::PadPresence::Always,
+            &caps,
+        )
+        .unwrap();
+        klass.add_pad_template(src_pad_template);
+    }
+}
+
+impl ObjectImpl for SccEnc {
+    glib_object_impl!();
+
+    fn constructed(&self, obj: &glib::Object) {
+        self.parent_constructed(obj);
+
+        let element = obj.downcast_ref::<gst::Element>().unwrap();
+        element.add_pad(&self.sinkpad).unwrap();
+        element.add_pad(&self.srcpad).unwrap();
+    }
+}
+
+impl ElementImpl for SccEnc {
+    fn change_state(
+        &self,
+        element: &gst::Element,
+        transition: gst::StateChange,
+    ) -> Result<gst::StateChangeSuccess, gst::StateChangeError> {
+        gst_trace!(CAT, obj: element, "Changing state {:?}", transition);
+
+        match transition {
+            gst::StateChange::ReadyToPaused | gst::StateChange::PausedToReady => {
+                // Reset the whole state
+                let mut state = self.state.lock().unwrap();
+                *state = State::default();
+            }
+            _ => (),
+        }
+
+        self.parent_change_state(element, transition)
+    }
+}
+
+pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
+    gst::Element::register(plugin, "sccenc", 0, SccEnc::get_type())
+}

gst-plugin-closedcaption/src/scc_parse.rs

@@ -0,0 +1,558 @@
+// Copyright (C) 2019 Sebastian Dröge <sebastian@centricular.com>
+// Copyright (C) 2019 Jordan Petridis <jordan@centricular.com>
+//
+// This library is free software; you can redistribute it and/or
+// modify it under the terms of the GNU Library General Public
+// License as published by the Free Software Foundation; either
+// version 2 of the License, or (at your option) any later version.
+//
+// This library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+// Library General Public License for more details.
+//
+// You should have received a copy of the GNU Library General Public
+// License along with this library; if not, write to the
+// Free Software Foundation, Inc., 51 Franklin Street, Suite 500,
+// Boston, MA 02110-1335, USA.
+
+use glib;
+use glib::prelude::*;
+use glib::subclass;
+use glib::subclass::prelude::*;
+use gst;
+use gst::prelude::*;
+use gst::subclass::prelude::*;
+use gst_video;
+
+use std::sync::{Mutex, MutexGuard};
+
+use crate::line_reader::LineReader;
+use crate::scc_parser::{SccLine, SccParser, TimeCode};
+
+lazy_static! {
+    static ref CAT: gst::DebugCategory = {
+        gst::DebugCategory::new(
+            "sccparse",
+            gst::DebugColorFlags::empty(),
+            "Scc Parser Element",
+        )
+    };
+}
+
+#[derive(Debug)]
+struct State {
+    reader: LineReader<gst::MappedBuffer<gst::buffer::Readable>>,
+    parser: SccParser,
+    need_segment: bool,
+    pending_events: Vec<gst::Event>,
+    start_position: gst::ClockTime,
+    last_position: gst::ClockTime,
+    last_timecode: Option<gst_video::ValidVideoTimeCode>,
+}
+
+impl Default for State {
+    fn default() -> Self {
+        Self {
+            reader: LineReader::new(),
+            parser: SccParser::new(),
+            need_segment: true,
+            pending_events: Vec::new(),
+            start_position: gst::CLOCK_TIME_NONE,
+            last_position: gst::CLOCK_TIME_NONE,
+            last_timecode: None,
+        }
+    }
+}
+
+impl State {
+    fn get_line(
+        &mut self,
+        drain: bool,
+    ) -> Result<
+        Option<SccLine>,
+        (
+            &[u8],
+            combine::easy::Errors<u8, &[u8], combine::stream::PointerOffset>,
+        ),
+    > {
+        let line = match self.reader.get_line_with_drain(drain) {
+            None => {
+                return Ok(None);
+            }
+            Some(line) => line,
+        };
+
+        self.parser
+            .parse_line(line)
+            .map(Option::Some)
+            .map_err(|err| (line, err))
+    }
+
+    fn handle_timecode(
+        &mut self,
+        tc: TimeCode,
+        framerate: gst::Fraction,
+        element: &gst::Element,
+    ) -> Result<gst_video::ValidVideoTimeCode, gst::FlowError> {
+        let timecode = gst_video::VideoTimeCode::new(
+            framerate,
+            None,
+            if tc.drop_frame {
+                gst_video::VideoTimeCodeFlags::DROP_FRAME
+            } else {
+                gst_video::VideoTimeCodeFlags::empty()
+            },
+            tc.hours,
+            tc.minutes,
+            tc.seconds,
+            tc.frames,
+            0,
+        );
+
+        match timecode.try_into() {
+            Ok(timecode) => Ok(timecode),
+            Err(timecode) => {
+                let last_timecode =
+                    self.last_timecode
+                        .as_ref()
+                        .map(Clone::clone)
+                        .ok_or_else(|| {
+                            gst_element_error!(
+                                element,
+                                gst::StreamError::Decode,
+                                ["Invalid first timecode {:?}", timecode]
+                            );
+
+                            gst::FlowError::Error
+                        })?;
+
+                gst_warning!(
+                    CAT,
+                    obj: element,
+                    "Invalid timecode {:?}, using previous {:?}",
+                    timecode,
+                    last_timecode
+                );
+
+                Ok(last_timecode)
+            }
+        }
+    }
+
+    /// Calculate a timestamp from the timecode and make sure to
+    /// not produce timestamps jumping backwards
+    fn update_timestamp(
+        &mut self,
+        timecode: &gst_video::ValidVideoTimeCode,
+        element: &gst::Element,
+    ) {
+        let nsecs = gst::ClockTime::from(timecode.nsec_since_daily_jam());
+        if self.start_position.is_none() {
+            self.start_position = nsecs;
+        }
+
+        let nsecs = if nsecs < self.start_position {
+            gst_fixme!(
+                CAT,
+                obj: element,
+                "New position {} < start position {}",
+                nsecs,
+                self.start_position
+            );
+            self.start_position
+        } else {
+            nsecs - self.start_position
+        };
+
+        if nsecs >= self.last_position {
+            self.last_position = nsecs;
+        } else {
+            gst_fixme!(
+                CAT,
+                obj: element,
+                "New position {} < last position {}",
+                nsecs,
+                self.last_position
+            );
+        }
+    }
+
+    fn add_buffer_metadata(
+        &mut self,
+        buffer: &mut gst::buffer::Buffer,
+        timecode: &gst_video::ValidVideoTimeCode,
+        framerate: &gst::Fraction,
+        element: &gst::Element,
+    ) {
+        let buffer = buffer.get_mut().unwrap();
+        gst_video::VideoTimeCodeMeta::add(buffer, &timecode);
+
+        self.update_timestamp(timecode, element);
+
+        buffer.set_pts(self.last_position);
+        buffer.set_duration(
+            gst::SECOND
+                .mul_div_ceil(*framerate.denom() as u64, *framerate.numer() as u64)
+                .unwrap_or(gst::CLOCK_TIME_NONE),
+        );
+    }
+}
+
+struct SccParse {
+    srcpad: gst::Pad,
+    sinkpad: gst::Pad,
+    state: Mutex<State>,
+}
+
+impl SccParse {
+    fn set_pad_functions(sinkpad: &gst::Pad, srcpad: &gst::Pad) {
+        sinkpad.set_chain_function(|pad, parent, buffer| {
+            SccParse::catch_panic_pad_function(
+                parent,
+                || Err(gst::FlowError::Error),
+                |parse, element| parse.sink_chain(pad, element, buffer),
+            )
+        });
+        sinkpad.set_event_function(|pad, parent, event| {
+            SccParse::catch_panic_pad_function(
+                parent,
+                || false,
+                |parse, element| parse.sink_event(pad, element, event),
+            )
+        });
+
+        srcpad.set_event_function(|pad, parent, event| {
+            SccParse::catch_panic_pad_function(
+                parent,
+                || false,
+                |parse, element| parse.src_event(pad, element, event),
+            )
+        });
+        srcpad.set_query_function(|pad, parent, query| {
+            SccParse::catch_panic_pad_function(
+                parent,
+                || false,
+                |parse, element| parse.src_query(pad, element, query),
+            )
+        });
+    }
+
+    fn handle_buffer(
+        &self,
+        element: &gst::Element,
+        buffer: Option<gst::Buffer>,
+    ) -> Result<gst::FlowSuccess, gst::FlowError> {
+        let mut state = self.state.lock().unwrap();
+
+        let drain;
+        if let Some(buffer) = buffer {
+            let buffer = buffer.into_mapped_buffer_readable().map_err(|_| {
+                gst_element_error!(
+                    element,
+                    gst::ResourceError::Read,
+                    ["Failed to map buffer readable"]
+                );
+
+                gst::FlowError::Error
+            })?;
+
+            state.reader.push(buffer);
+            drain = false;
+        } else {
+            drain = true;
+        }
+
+        loop {
+            let line = state.get_line(drain);
+            match line {
+                Ok(Some(SccLine::Caption(tc, data))) => {
+                    state = self.handle_line(tc, data, element, state)?;
+                }
+                Ok(Some(line)) => {
+                    gst_debug!(CAT, obj: element, "Got line '{:?}'", line);
+                }
+                Err((line, err)) => {
+                    gst_element_error!(
+                        element,
+                        gst::StreamError::Decode,
+                        ["Couldn't parse line '{:?}': {:?}", line, err]
+                    );
+
+                    break Err(gst::FlowError::Error);
+                }
+                Ok(None) => break Ok(gst::FlowSuccess::Ok),
+            }
+        }
+    }
+
+    fn handle_line(
+        &self,
+        tc: TimeCode,
+        data: Vec<u8>,
+        element: &gst::Element,
+        mut state: MutexGuard<State>,
+    ) -> Result<MutexGuard<State>, gst::FlowError> {
+        gst_trace!(
+            CAT,
+            obj: element,
+            "Got caption buffer with timecode {:?} and size {}",
+            tc,
+            data.len()
+        );
+
+        // The framerate is defined as 30 or 30000/1001 according to:
+        // http://www.theneitherworld.com/mcpoodle/SCC_TOOLS/DOCS/SCC_FORMAT.HTML
+        let framerate = if tc.drop_frame {
+            gst::Fraction::new(30000, 1001)
+        } else {
+            gst::Fraction::new(30, 1)
+        };
+
+        let mut events = Vec::new();
+
+        let caps = gst::Caps::builder("closedcaption/x-cea-608")
+            .field("format", &"raw")
+            .field("framerate", &framerate)
+            .build();
+        events.push(gst::Event::new_caps(&caps).build());
+
+        if state.need_segment {
+            let segment = gst::FormattedSegment::<gst::format::Time>::new();
+            events.push(gst::Event::new_segment(&segment).build());
+            state.need_segment = false;
+        }
+
+        events.extend(state.pending_events.drain(..));
+
+        let mut timecode = state.handle_timecode(tc, framerate, element)?;
+        let mut buffers = gst::BufferList::new_sized(data.len() / 2);
+        for d in data.chunks_exact(2) {
+            let mut buffer = gst::Buffer::with_size(d.len()).unwrap();
+            {
+                let buf_mut = buffer.get_mut().unwrap();
+                buf_mut.copy_from_slice(0, d).unwrap();
+            }
+
+            state.add_buffer_metadata(&mut buffer, &timecode, &framerate, element);
+            timecode.increment_frame();
+            let buffers = buffers.get_mut().unwrap();
+            buffers.add(buffer);
+        }
+
+        // Update the last_timecode to the current one
+        state.last_timecode = Some(timecode);
+
+        // Drop our state mutex while we push out buffers or events
+        drop(state);
+
+        for event in events {
+            gst_debug!(CAT, obj: element, "Pushing event {:?}", event);
+            self.srcpad.push_event(event);
+        }
+
+        self.srcpad.push_list(buffers).map_err(|err| {
+            gst_error!(CAT, obj: element, "Pushing buffer returned {:?}", err);
+            err
+        })?;
+
+        Ok(self.state.lock().unwrap())
+    }
+
+    fn sink_chain(
+        &self,
+        pad: &gst::Pad,
+        element: &gst::Element,
+        buffer: gst::Buffer,
+    ) -> Result<gst::FlowSuccess, gst::FlowError> {
+        gst_log!(CAT, obj: pad, "Handling buffer {:?}", buffer);
+
+        self.handle_buffer(element, Some(buffer))
+    }
+
+    fn sink_event(&self, pad: &gst::Pad, element: &gst::Element, event: gst::Event) -> bool {
+        use gst::EventView;
+
+        gst_log!(CAT, obj: pad, "Handling event {:?}", event);
+
+        match event.view() {
+            EventView::Caps(_) => {
+                // We send a proper caps event from the chain function later
+                gst_log!(CAT, obj: pad, "Dropping caps event");
+                true
+            }
+            EventView::Segment(_) => {
+                // We send a gst::Format::Time segment event later when needed
+                gst_log!(CAT, obj: pad, "Dropping segment event");
+                true
+            }
+            EventView::FlushStop(_) => {
+                let mut state = self.state.lock().unwrap();
+                state.reader.clear();
+                state.parser.reset();
+                state.need_segment = true;
+                state.pending_events.clear();
+                state.start_position = gst::ClockTime::from_seconds(0);
+                state.last_position = gst::ClockTime::from_seconds(0);
+                state.last_timecode = None;
+
+                pad.event_default(element, event)
+            }
+            EventView::Eos(_) => {
+                gst_log!(CAT, obj: pad, "Draining");
+                if let Err(err) = self.handle_buffer(element, None) {
+                    gst_error!(CAT, obj: pad, "Failed to drain parser: {:?}", err);
+                }
+                pad.event_default(element, event)
+            }
+            _ => {
+                if event.is_sticky()
+                    && !self.srcpad.has_current_caps()
+                    && event.get_type() > gst::EventType::Caps
+                {
+                    gst_log!(CAT, obj: pad, "Deferring sticky event until we have caps");
+                    let mut state = self.state.lock().unwrap();
+                    state.pending_events.push(event);
+                    true
+                } else {
+                    pad.event_default(element, event)
+                }
+            }
+        }
+    }
+
+    fn src_event(&self, pad: &gst::Pad, element: &gst::Element, event: gst::Event) -> bool {
+        use gst::EventView;
+
+        gst_log!(CAT, obj: pad, "Handling event {:?}", event);
+        match event.view() {
+            EventView::Seek(_) => {
+                gst_log!(CAT, obj: pad, "Dropping seek event");
+                false
+            }
+            _ => pad.event_default(element, event),
+        }
+    }
+
+    fn src_query(&self, pad: &gst::Pad, element: &gst::Element, query: &mut gst::QueryRef) -> bool {
+        use gst::QueryView;
+
+        gst_log!(CAT, obj: pad, "Handling query {:?}", query);
+
+        match query.view_mut() {
+            QueryView::Seeking(mut q) => {
+                // We don't support any seeking at all
+                let fmt = q.get_format();
+                q.set(
+                    false,
+                    gst::GenericFormattedValue::Other(fmt, -1),
+                    gst::GenericFormattedValue::Other(fmt, -1),
+                );
+                true
+            }
+            QueryView::Position(ref mut q) => {
+                // For Time answer ourselfs, otherwise forward
+                if q.get_format() == gst::Format::Time {
+                    let state = self.state.lock().unwrap();
+                    q.set(state.last_position);
+                    true
+                } else {
+                    self.sinkpad.peer_query(query)
+                }
+            }
+            _ => pad.query_default(element, query),
+        }
+    }
+}
+
+impl ObjectSubclass for SccParse {
+    const NAME: &'static str = "RsSccParse";
+    type ParentType = gst::Element;
+    type Instance = gst::subclass::ElementInstanceStruct<Self>;
+    type Class = subclass::simple::ClassStruct<Self>;
+
+    glib_object_subclass!();
+
+    fn new_with_class(klass: &subclass::simple::ClassStruct<Self>) -> Self {
+        let templ = klass.get_pad_template("sink").unwrap();
+        let sinkpad = gst::Pad::new_from_template(&templ, "sink");
+        let templ = klass.get_pad_template("src").unwrap();
+        let srcpad = gst::Pad::new_from_template(&templ, "src");
+
+        SccParse::set_pad_functions(&sinkpad, &srcpad);
+
+        Self {
+            srcpad,
+            sinkpad,
+            state: Mutex::new(State::default()),
+        }
+    }
+
+    fn class_init(klass: &mut subclass::simple::ClassStruct<Self>) {
+        klass.set_metadata(
+            "Scc Parse",
+            "Parser/ClosedCaption",
+            "Parses SCC Closed Caption Files",
+            "Sebastian Dröge <sebastian@centricular.com>, Jordan Petridis <jordan@centricular.com>",
+        );
+
+        let caps = gst::Caps::builder("closedcaption/x-cea-608")
+            .field("format", &"raw")
+            .build();
+        let src_pad_template = gst::PadTemplate::new(
+            "src",
+            gst::PadDirection::Src,
+            gst::PadPresence::Always,
+            &caps,
+        )
+        .unwrap();
+        klass.add_pad_template(src_pad_template);
+
+        let caps = gst::Caps::builder("application/x-scc").build();
+        let sink_pad_template = gst::PadTemplate::new(
+            "sink",
+            gst::PadDirection::Sink,
+            gst::PadPresence::Always,
+            &caps,
+        )
+        .unwrap();
+        klass.add_pad_template(sink_pad_template);
+    }
+}
+
+impl ObjectImpl for SccParse {
+    glib_object_impl!();
+
+    fn constructed(&self, obj: &glib::Object) {
+        self.parent_constructed(obj);
+
+        let element = obj.downcast_ref::<gst::Element>().unwrap();
+        element.add_pad(&self.sinkpad).unwrap();
+        element.add_pad(&self.srcpad).unwrap();
+    }
+}
+
+impl ElementImpl for SccParse {
+    fn change_state(
+        &self,
+        element: &gst::Element,
+        transition: gst::StateChange,
+    ) -> Result<gst::StateChangeSuccess, gst::StateChangeError> {
+        gst_trace!(CAT, obj: element, "Changing state {:?}", transition);
+
+        match transition {
+            gst::StateChange::ReadyToPaused | gst::StateChange::PausedToReady => {
+                // Reset the whole state
+                let mut state = self.state.lock().unwrap();
+                *state = State::default();
+            }
+            _ => (),
+        }
+
+        self.parent_change_state(element, transition)
+    }
+}
+
+pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
+    gst::Element::register(plugin, "sccparse", 0, SccParse::get_type())
+}

gst-plugin-closedcaption/src/scc_parser.rs

@@ -0,0 +1,508 @@
+// Copyright (C) 2019 Sebastian Dröge <sebastian@centricular.com>
+// Copyright (C) 2019 Jordan Petridis <jordan@centricular.com>
+//
+// This library is free software; you can redistribute it and/or
+// modify it under the terms of the GNU Library General Public
+// License as published by the Free Software Foundation; either
+// version 2 of the License, or (at your option) any later version.
+//
+// This library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+// Library General Public License for more details.
+//
+// You should have received a copy of the GNU Library General Public
+// License along with this library; if not, write to the
+// Free Software Foundation, Inc., 51 Franklin Street, Suite 500,
+// Boston, MA 02110-1335, USA.
+
+use combine;
+use combine::parser::byte::hex_digit;
+use combine::parser::range::{range, take_while1};
+use combine::{choice, eof, from_str, many1, one_of, optional, token, unexpected_any, value};
+use combine::{ParseError, Parser, RangeStream};
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct TimeCode {
+    pub hours: u32,
+    pub minutes: u32,
+    pub seconds: u32,
+    pub frames: u32,
+    pub drop_frame: bool,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum SccLine {
+    Header,
+    Empty,
+    Caption(TimeCode, Vec<u8>),
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum State {
+    Init,
+    Header,
+    Empty,
+    Captions,
+}
+
+#[derive(Debug)]
+pub struct SccParser {
+    state: State,
+}
+
+/// Parser for parsing a run of ASCII, decimal digits and converting them into a `u32`
+fn digits<'a, I: 'a>() -> impl Parser<Input = I, Output = u32>
+where
+    I: RangeStream<Item = u8, Range = &'a [u8]>,
+    I::Error: ParseError<I::Item, I::Range, I::Position>,
+{
+    from_str(take_while1(|c: u8| c >= b'0' && c <= b'9').message("while parsing digits"))
+}
+
+/// Copy from std::ops::RangeBounds as it's not stabilized yet.
+///
+/// Checks if `item` is in the range `range`.
+fn contains<R: std::ops::RangeBounds<U>, U>(range: &R, item: &U) -> bool
+where
+    U: ?Sized + PartialOrd<U>,
+{
+    (match range.start_bound() {
+        std::ops::Bound::Included(ref start) => *start <= item,
+        std::ops::Bound::Excluded(ref start) => *start < item,
+        std::ops::Bound::Unbounded => true,
+    }) && (match range.end_bound() {
+        std::ops::Bound::Included(ref end) => item <= *end,
+        std::ops::Bound::Excluded(ref end) => item < *end,
+        std::ops::Bound::Unbounded => true,
+    })
+}
+
+/// Parser for a run of decimal digits, that converts them into a `u32` and checks if the result is
+/// in the allowed range.
+fn digits_range<'a, I: 'a, R: std::ops::RangeBounds<u32>>(
+    range: R,
+) -> impl Parser<Input = I, Output = u32>
+where
+    I: RangeStream<Item = u8, Range = &'a [u8]>,
+    I::Error: ParseError<I::Item, I::Range, I::Position>,
+{
+    digits().then(move |v| {
+        if contains(&range, &v) {
+            value(v).left()
+        } else {
+            unexpected_any("digits out of range").right()
+        }
+    })
+}
+
+/// Parser for a timecode in the form `hh:mm:ss:fs`
+fn timecode<'a, I: 'a>() -> impl Parser<Input = I, Output = TimeCode>
+where
+    I: RangeStream<Item = u8, Range = &'a [u8]>,
+    I::Error: ParseError<I::Item, I::Range, I::Position>,
+{
+    (
+        digits(),
+        token(b':'),
+        digits_range(0..60),
+        token(b':'),
+        digits_range(0..60),
+        one_of([b':', b'.', b';', b','].iter().cloned()),
+        digits(),
+    )
+        .map(|(hours, _, minutes, _, seconds, sep, frames)| TimeCode {
+            hours,
+            minutes,
+            seconds,
+            frames,
+            drop_frame: sep == b';' || sep == b',',
+        })
+        .message("while parsing timecode")
+}
+
+/// Parser that checks for EOF and optionally `\n` or `\r\n` before EOF
+fn end_of_line<'a, I: 'a>() -> impl Parser<Input = I, Output = ()> + 'a
+where
+    I: RangeStream<Item = u8, Range = &'a [u8]>,
+    I::Error: ParseError<I::Item, I::Range, I::Position>,
+{
+    (
+        optional(choice((range(b"\n".as_ref()), range(b"\r\n".as_ref())))),
+        eof(),
+    )
+        .map(|_| ())
+        .message("while parsing end of line")
+}
+
+/// Parser for the SCC header
+fn header<'a, I: 'a>() -> impl Parser<Input = I, Output = SccLine> + 'a
+where
+    I: RangeStream<Item = u8, Range = &'a [u8]>,
+    I::Error: ParseError<I::Item, I::Range, I::Position>,
+{
+    (range(b"Scenarist_SCC V1.0".as_ref()), end_of_line())
+        .map(|_| SccLine::Header)
+        .message("while parsing header")
+}
+
+/// Parser that accepts only an empty line
+fn empty_line<'a, I: 'a>() -> impl Parser<Input = I, Output = SccLine> + 'a
+where
+    I: RangeStream<Item = u8, Range = &'a [u8]>,
+    I::Error: ParseError<I::Item, I::Range, I::Position>,
+{
+    end_of_line()
+        .map(|_| SccLine::Empty)
+        .message("while parsing empty line")
+}
+
+/// A single SCC payload item. This is ASCII hex encoded bytes.
+/// It returns an tuple of `(u8, u8)` of the hex encoded bytes.
+fn scc_payload_item<'a, I: 'a>() -> impl Parser<Input = I, Output = (u8, u8)>
+where
+    I: RangeStream<Item = u8, Range = &'a [u8]>,
+    I::Error: ParseError<I::Item, I::Range, I::Position>,
+{
+    ((hex_digit(), hex_digit(), hex_digit(), hex_digit()).map(|(u, l, m, n)| {
+        let hex_to_u8 = |v: u8| match v {
+            v if v >= b'0' && v <= b'9' => v - b'0',
+            v if v >= b'A' && v <= b'F' => 10 + v - b'A',
+            v if v >= b'a' && v <= b'f' => 10 + v - b'a',
+            _ => unreachable!(),
+        };
+        let val = (hex_to_u8(u) << 4) | hex_to_u8(l);
+        let val2 = (hex_to_u8(m) << 4) | hex_to_u8(n);
+        (val, val2)
+    }))
+    .message("while parsing SCC payload")
+}
+
+/// A wrapper around `Vec<u8>` that implements `Extend` in a special way. It can be
+/// extended from a `(u8, u8)` while the default `Extend` implementation for
+/// `Vec` only allows to extend over vector items.
+struct VecExtend(Vec<u8>);
+
+impl Default for VecExtend {
+    fn default() -> Self {
+        VecExtend(Vec::with_capacity(256))
+    }
+}
+
+impl Extend<(u8, u8)> for VecExtend {
+    fn extend<T>(&mut self, iter: T)
+    where
+        T: IntoIterator<Item = (u8, u8)>,
+    {
+        for (item, item2) in iter {
+            self.0.extend_from_slice(&[item, item2]);
+        }
+    }
+}
+
+/// Parser for the whole SCC payload with conversion to the underlying byte values.
+fn scc_payload<'a, I: 'a>() -> impl Parser<Input = I, Output = Vec<u8>> + 'a
+where
+    I: RangeStream<Item = u8, Range = &'a [u8]>,
+    I::Error: ParseError<I::Item, I::Range, I::Position>,
+{
+    many1(
+        (
+            scc_payload_item(),
+            choice!(token(b' ').map(|_| ()), end_of_line()),
+        )
+            .map(|v| v.0),
+    )
+    .map(|v: VecExtend| v.0)
+    .message("while parsing SCC payloads")
+}
+
+/// Parser for a SCC caption line in the form `timecode\tpayload`.
+fn caption<'a, I: 'a>() -> impl Parser<Input = I, Output = SccLine> + 'a
+where
+    I: RangeStream<Item = u8, Range = &'a [u8]>,
+    I::Error: ParseError<I::Item, I::Range, I::Position>,
+{
+    (timecode(), token(b'\t'), scc_payload(), end_of_line())
+        .map(|(tc, _, value, _)| SccLine::Caption(tc, value))
+        .message("while parsing caption")
+}
+
+/// SCC parser the parses line-by-line and keeps track of the current state in the file.
+impl SccParser {
+    pub fn new() -> Self {
+        Self { state: State::Init }
+    }
+
+    pub fn reset(&mut self) {
+        self.state = State::Init;
+    }
+
+    pub fn parse_line<'a>(
+        &mut self,
+        line: &'a [u8],
+    ) -> Result<SccLine, combine::easy::Errors<u8, &'a [u8], combine::stream::PointerOffset>> {
+        match self.state {
+            State::Init => header()
+                .message("while in Init state")
+                .easy_parse(line)
+                .map(|v| {
+                    self.state = State::Header;
+                    v.0
+                }),
+            State::Header => empty_line()
+                .message("while in Header state")
+                .easy_parse(line)
+                .map(|v| {
+                    self.state = State::Empty;
+                    v.0
+                }),
+            State::Empty => caption()
+                .message("while in Empty state")
+                .easy_parse(line)
+                .map(|v| {
+                    self.state = State::Captions;
+                    v.0
+                }),
+            State::Captions => empty_line()
+                .message("while in Captions state")
+                .easy_parse(line)
+                .map(|v| {
+                    self.state = State::Empty;
+                    v.0
+                }),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use combine::error::UnexpectedParse;
+
+    #[test]
+    fn test_timecode() {
+        let mut parser = timecode();
+        assert_eq!(
+            parser.parse(b"11:12:13;14".as_ref()),
+            Ok((
+                TimeCode {
+                    hours: 11,
+                    minutes: 12,
+                    seconds: 13,
+                    frames: 14,
+                    drop_frame: true
+                },
+                b"".as_ref()
+            ))
+        );
+
+        assert_eq!(
+            parser.parse(b"11:12:13:14".as_ref()),
+            Ok((
+                TimeCode {
+                    hours: 11,
+                    minutes: 12,
+                    seconds: 13,
+                    frames: 14,
+                    drop_frame: false
+                },
+                b"".as_ref()
+            ))
+        );
+
+        assert_eq!(
+            parser.parse(b"11:12:13:14abcd".as_ref()),
+            Ok((
+                TimeCode {
+                    hours: 11,
+                    minutes: 12,
+                    seconds: 13,
+                    frames: 14,
+                    drop_frame: false
+                },
+                b"abcd".as_ref()
+            ))
+        );
+
+        assert_eq!(
+            parser.parse(b"abcd11:12:13:14".as_ref()),
+            Err(UnexpectedParse::Unexpected)
+        );
+    }
+
+    #[test]
+    fn test_header() {
+        let mut parser = header();
+        assert_eq!(
+            parser.parse(b"Scenarist_SCC V1.0".as_ref()),
+            Ok((SccLine::Header, b"".as_ref()))
+        );
+
+        assert_eq!(
+            parser.parse(b"Scenarist_SCC V1.0\n".as_ref()),
+            Ok((SccLine::Header, b"".as_ref()))
+        );
+
+        assert_eq!(
+            parser.parse(b"Scenarist_SCC V1.0\r\n".as_ref()),
+            Ok((SccLine::Header, b"".as_ref()))
+        );
+
+        assert_eq!(
+            parser.parse(b"Scenarist_SCC V1.1".as_ref()),
+            Err(UnexpectedParse::Unexpected)
+        );
+    }
+
+    #[test]
+    fn test_empty_line() {
+        let mut parser = empty_line();
+        assert_eq!(
+            parser.parse(b"".as_ref()),
+            Ok((SccLine::Empty, b"".as_ref()))
+        );
+
+        assert_eq!(
+            parser.parse(b"\n".as_ref()),
+            Ok((SccLine::Empty, b"".as_ref()))
+        );
+
+        assert_eq!(
+            parser.parse(b"\r\n".as_ref()),
+            Ok((SccLine::Empty, b"".as_ref()))
+        );
+
+        assert_eq!(
+            parser.parse(b" \r\n".as_ref()),
+            Err(UnexpectedParse::Unexpected)
+        );
+    }
+
+    #[test]
+    fn test_caption() {
+        let mut parser = caption();
+        assert_eq!(
+            parser.parse(b"01:02:53:14\t94ae 94ae 9420 9420 947a 947a 97a2 97a2 a820 68ef f26e 2068 ef6e 6be9 6e67 2029 942c 942c 8080 8080 942f 942f".as_ref()),
+            Ok((
+                SccLine::Caption(
+                    TimeCode {
+                        hours: 1,
+                        minutes: 2,
+                        seconds: 53,
+                        frames: 14,
+                        drop_frame: false
+                    },
+
+                    vec![
+                        0x94, 0xae, 0x94, 0xae, 0x94, 0x20, 0x94, 0x20, 0x94, 0x7a, 0x94, 0x7a, 0x97, 0xa2,
+                        0x97, 0xa2, 0xa8, 0x20, 0x68, 0xef, 0xf2, 0x6e, 0x20, 0x68, 0xef, 0x6e, 0x6b, 0xe9,
+                        0x6e, 0x67, 0x20, 0x29, 0x94, 0x2c, 0x94, 0x2c, 0x80, 0x80, 0x80, 0x80, 0x94, 0x2f,
+                        0x94, 0x2f,
+                    ]
+                ),
+                b"".as_ref()
+            ))
+        );
+
+        assert_eq!(
+            parser.parse(b"01:02:55;14	942c 942c".as_ref()),
+            Ok((
+                SccLine::Caption(
+                    TimeCode {
+                        hours: 1,
+                        minutes: 2,
+                        seconds: 55,
+                        frames: 14,
+                        drop_frame: true
+                    },
+                    vec![0x94, 0x2c, 0x94, 0x2c]
+                ),
+                b"".as_ref()
+            ))
+        );
+
+        assert_eq!(
+            parser.parse(b"01:03:27:29	94ae 94ae 9420 9420 94f2 94f2 c845 d92c 2054 c845 5245 ae80 942c 942c 8080 8080 942f 942f".as_ref()),
+            Ok((
+                SccLine::Caption(
+                    TimeCode {
+                        hours: 1,
+                        minutes: 3,
+                        seconds: 27,
+                        frames: 29,
+                        drop_frame: false
+                    },
+                    vec![
+                        0x94, 0xae, 0x94, 0xae, 0x94, 0x20, 0x94, 0x20, 0x94, 0xf2, 0x94, 0xf2, 0xc8, 0x45,
+                        0xd9, 0x2c, 0x20, 0x54, 0xc8, 0x45, 0x52, 0x45, 0xae, 0x80, 0x94, 0x2c, 0x94, 0x2c,
+                        0x80, 0x80, 0x80, 0x80, 0x94, 0x2f, 0x94, 0x2f,
+                    ]
+                ),
+                b"".as_ref()
+            ))
+        );
+
+        assert_eq!(
+            parser.parse(b"00:00:00;00\t942c 942c".as_ref()),
+            Ok((
+                SccLine::Caption(
+                    TimeCode {
+                        hours: 0,
+                        minutes: 0,
+                        seconds: 00,
+                        frames: 00,
+                        drop_frame: true,
+                    },
+                    vec![0x94, 0x2c, 0x94, 0x2c],
+                ),
+                b"".as_ref()
+            ))
+        );
+
+        assert_eq!(
+            parser.parse(b"00:00:00;00\t942c 942c\r\n".as_ref()),
+            Ok((
+                SccLine::Caption(
+                    TimeCode {
+                        hours: 0,
+                        minutes: 0,
+                        seconds: 00,
+                        frames: 00,
+                        drop_frame: true,
+                    },
+                    vec![0x94, 0x2c, 0x94, 0x2c],
+                ),
+                b"".as_ref()
+            ))
+        );
+    }
+
+    #[test]
+    fn test_parser() {
+        let scc_file = include_bytes!("../tests/dn2018-1217.scc");
+        let mut reader = crate::line_reader::LineReader::new();
+        let mut parser = SccParser::new();
+        let mut line_cnt = 0;
+
+        reader.push(Vec::from(scc_file.as_ref()));
+
+        while let Some(line) = reader.get_line() {
+            let res = match parser.parse_line(line) {
+                Ok(res) => res,
+                Err(err) => panic!("Couldn't parse line {}: {:?}", line_cnt, err),
+            };
+
+            match line_cnt {
+                0 => assert_eq!(res, SccLine::Header),
+                x if x % 2 != 0 => assert_eq!(res, SccLine::Empty),
+                _ => match res {
+                    SccLine::Caption(_, _) => (),
+                    res => panic!("Expected caption at line {}, got {:?}", line_cnt, res),
+                },
+            }
+
+            line_cnt += 1;
+        }
+    }
+}

gst-plugin-closedcaption/tests/scc_enc.rs

@@ -0,0 +1,227 @@
+// Copyright (C) 2019 Sebastian Dröge <sebastian@centricular.com>
+// Copyright (C) 2019 Jordan Petridis <jordan@centricular.com>
+//
+// This library is free software; you can redistribute it and/or
+// modify it under the terms of the GNU Library General Public
+// License as published by the Free Software Foundation; either
+// version 2 of the License, or (at your option) any later version.
+//
+// This library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+// Library General Public License for more details.
+//
+// You should have received a copy of the GNU Library General Public
+// License along with this library; if not, write to the
+// Free Software Foundation, Inc., 51 Franklin Street, Suite 500,
+// Boston, MA 02110-1335, USA.
+
+#[macro_use]
+extern crate pretty_assertions;
+
+fn init() {
+    use std::sync::{Once, ONCE_INIT};
+    static INIT: Once = ONCE_INIT;
+
+    INIT.call_once(|| {
+        gst::init().unwrap();
+        gstrsclosedcaption::plugin_register_static().unwrap();
+    });
+}
+
+/// Encode a single raw CEA608 packet and compare the output
+#[test]
+fn test_encode_single_packet() {
+    init();
+
+    let input = [148, 44];
+    let expected_output = b"Scenarist_SCC V1.0\r\n\r\n11:12:13;14\t942c\r\n\r\n";
+
+    let mut h = gst_check::Harness::new("sccenc");
+    h.set_src_caps_str("closedcaption/x-cea-608, format=raw, framerate=(fraction)30000/1001");
+    let tc = gst_video::ValidVideoTimeCode::new(
+        gst::Fraction::new(30000, 1001),
+        None,
+        gst_video::VideoTimeCodeFlags::DROP_FRAME,
+        11,
+        12,
+        13,
+        14,
+        0,
+    )
+    .unwrap();
+
+    let buf = {
+        let mut buf = gst::Buffer::from_mut_slice(Vec::from(&input[..]));
+        let buf_ref = buf.get_mut().unwrap();
+        gst_video::VideoTimeCodeMeta::add(buf_ref, &tc);
+        buf_ref.set_pts(gst::ClockTime::from_seconds(0));
+        buf
+    };
+
+    assert_eq!(h.push(buf), Ok(gst::FlowSuccess::Ok));
+    h.push_event(gst::Event::new_eos().build());
+
+    let buf = h.pull().expect("Couldn't pull buffer");
+
+    let timecode = buf
+        .get_meta::<gst_video::VideoTimeCodeMeta>()
+        .expect("No timecode for buffer")
+        .get_tc();
+    assert_eq!(timecode, tc);
+
+    let pts = buf.get_pts();
+    assert_eq!(pts, gst::ClockTime::from_seconds(0));
+
+    let map = buf.map_readable().expect("Couldn't map buffer readable");
+    assert_eq!(
+        std::str::from_utf8(map.as_ref()),
+        std::str::from_utf8(expected_output.as_ref())
+    );
+}
+
+/// Encode a multiple raw CEA608 packets and compare the output
+#[test]
+fn test_encode_multiple_packets() {
+    init();
+
+    let input1 = [148, 44];
+    let input2 = [
+        148, 32, 148, 32, 148, 174, 148, 174, 148, 84, 148, 84, 16, 174, 16, 174, 70, 242, 239,
+        109, 32, 206, 229, 247, 32, 217, 239, 242, 107, 44, 148, 242, 148, 242, 16, 174, 16, 174,
+        244, 104, 233, 115, 32, 233, 115, 32, 196, 229, 109, 239, 227, 242, 97, 227, 121, 32, 206,
+        239, 247, 161, 148, 47, 148, 47,
+    ];
+
+    let expected_output1 = b"Scenarist_SCC V1.0\r\n\r\n00:00:00;00\t942c 942c\r\n\r\n";
+    let expected_output2 = b"00:00:14;01\t9420 9420 94ae 94ae 9454 9454 10ae 10ae 46f2 ef6d 20ce e5f7 20d9 eff2 6b2c 94f2\r\n\r\n";
+    let expected_output3 = b"00:00:14;17\t94f2 10ae 10ae f468 e973 20e9 7320 c4e5 6def e3f2 61e3 7920 ceef f7a1 942f 942f\r\n\r\n";
+
+    let mut h = gst_check::Harness::new("sccenc");
+    h.set_src_caps_str("closedcaption/x-cea-608, format=raw, framerate=(fraction)30000/1001");
+    let tc1 = gst_video::ValidVideoTimeCode::new(
+        gst::Fraction::new(30000, 1001),
+        None,
+        gst_video::VideoTimeCodeFlags::DROP_FRAME,
+        00,
+        00,
+        00,
+        00,
+        0,
+    )
+    .unwrap();
+
+    let tc2 = gst_video::ValidVideoTimeCode::new(
+        gst::Fraction::new(30000, 1001),
+        None,
+        gst_video::VideoTimeCodeFlags::DROP_FRAME,
+        00,
+        00,
+        14,
+        01,
+        0,
+    )
+    .unwrap();
+
+    let buf1 = {
+        let mut buf = gst::Buffer::from_mut_slice(Vec::from(&input1[..]));
+        let buf_ref = buf.get_mut().unwrap();
+        gst_video::VideoTimeCodeMeta::add(buf_ref, &tc1);
+        buf_ref.set_pts(gst::ClockTime::from_seconds(0));
+        buf
+    };
+
+    let buf2 = {
+        let mut buf = gst::Buffer::from_mut_slice(Vec::from(&input1[..]));
+        let buf_ref = buf.get_mut().unwrap();
+        let mut tc = tc1.clone();
+        tc.increment_frame();
+        gst_video::VideoTimeCodeMeta::add(buf_ref, &tc);
+        buf_ref.set_pts(gst::ClockTime::from_seconds(0));
+        buf
+    };
+
+    let mut t = tc2.clone();
+    let mut buffers = input2
+        .chunks(2)
+        .map(move |bytes| {
+            let mut buf = gst::Buffer::from_mut_slice(Vec::from(&bytes[..]));
+            let buf_ref = buf.get_mut().unwrap();
+            gst_video::VideoTimeCodeMeta::add(buf_ref, &t);
+            t.increment_frame();
+            buf
+        })
+        .collect::<Vec<gst::Buffer>>();
+    buffers.insert(0, buf1);
+    buffers.insert(1, buf2);
+
+    buffers.iter().for_each(|buf| {
+        assert_eq!(h.push(buf.clone()), Ok(gst::FlowSuccess::Ok));
+    });
+    h.push_event(gst::Event::new_eos().build());
+
+    // Pull 1
+    let buf = h.pull().expect("Couldn't pull buffer");
+
+    let timecode = buf
+        .get_meta::<gst_video::VideoTimeCodeMeta>()
+        .expect("No timecode for buffer")
+        .get_tc();
+    assert_eq!(timecode, tc1);
+
+    let pts = buf.get_pts();
+    assert_eq!(pts, gst::ClockTime::from_seconds(0));
+
+    let map = buf.map_readable().expect("Couldn't map buffer readable");
+
+    assert_eq!(
+        std::str::from_utf8(map.as_ref()),
+        std::str::from_utf8(expected_output1.as_ref())
+    );
+
+    // Pull 2
+    let buf = h.pull().expect("Couldn't pull buffer");
+    let timecode = buf
+        .get_meta::<gst_video::VideoTimeCodeMeta>()
+        .expect("No timecode for buffer")
+        .get_tc();
+    assert_eq!(timecode, tc2);
+
+    // let pts = buf.get_pts();
+    // assert_eq!(pts, gst::ClockTime::from_seconds(0));
+
+    let map = buf.map_readable().expect("Couldn't map buffer readable");
+    assert_eq!(
+        std::str::from_utf8(map.as_ref()),
+        std::str::from_utf8(expected_output2.as_ref())
+    );
+
+    let tc3 = gst_video::ValidVideoTimeCode::new(
+        gst::Fraction::new(30000, 1001),
+        None,
+        gst_video::VideoTimeCodeFlags::DROP_FRAME,
+        00,
+        00,
+        14,
+        17,
+        0,
+    )
+    .unwrap();
+
+    // Pull 3
+    let buf = h.pull().expect("Couldn't pull buffer");
+    let timecode = buf
+        .get_meta::<gst_video::VideoTimeCodeMeta>()
+        .expect("No timecode for buffer")
+        .get_tc();
+    assert_eq!(timecode, tc3);
+
+    // let pts = buf.get_pts();
+    // assert_eq!(pts, gst::ClockTime::from_seconds(0));
+
+    let map = buf.map_readable().expect("Couldn't map buffer readable");
+    assert_eq!(
+        std::str::from_utf8(map.as_ref()),
+        std::str::from_utf8(expected_output3.as_ref())
+    );
+}

gst-plugin-closedcaption/tests/scc_parse.rs

@@ -0,0 +1,200 @@
+// Copyright (C) 2019 Sebastian Dröge <sebastian@centricular.com>
+// Copyright (C) 2019 Jordan Petridis <jordan@centricular.com>
+//
+// This library is free software; you can redistribute it and/or
+// modify it under the terms of the GNU Library General Public
+// License as published by the Free Software Foundation; either
+// version 2 of the License, or (at your option) any later version.
+//
+// This library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+// Library General Public License for more details.
+//
+// You should have received a copy of the GNU Library General Public
+// License along with this library; if not, write to the
+// Free Software Foundation, Inc., 51 Franklin Street, Suite 500,
+// Boston, MA 02110-1335, USA.
+
+#[macro_use]
+extern crate pretty_assertions;
+
+use gst::prelude::*;
+use gst_video::{ValidVideoTimeCode, VideoTimeCode};
+use rand::{Rng, SeedableRng};
+use std::collections::VecDeque;
+
+fn init() {
+    use std::sync::{Once, ONCE_INIT};
+    static INIT: Once = ONCE_INIT;
+
+    INIT.call_once(|| {
+        gst::init().unwrap();
+        gstrsclosedcaption::plugin_register_static().unwrap();
+    });
+}
+
+/// Randomized test passing buffers of arbitrary sizes to the parser
+#[test]
+fn test_parse() {
+    init();
+    let mut data = include_bytes!("dn2018-1217.scc").as_ref();
+
+    let mut rnd = if let Ok(seed) = std::env::var("SCC_PARSE_TEST_SEED") {
+        rand::rngs::SmallRng::seed_from_u64(
+            seed.parse::<u64>()
+                .expect("SCC_PARSE_TEST_SEED has to contain a 64 bit integer seed"),
+        )
+    } else {
+        let seed = rand::random::<u64>();
+        println!("seed {}", seed);
+        rand::rngs::SmallRng::seed_from_u64(seed)
+    };
+
+    let mut h = gst_check::Harness::new("sccparse");
+    h.set_src_caps_str("application/x-scc");
+
+    let mut input_len = 0;
+    let mut output_len = 0;
+    let mut checksum = 0u32;
+
+    while !data.is_empty() {
+        let l = if data.len() == 1 {
+            1
+        } else {
+            rnd.gen_range(1, data.len())
+        };
+        let buf = gst::Buffer::from_mut_slice(Vec::from(&data[0..l]));
+        input_len += buf.get_size();
+        assert_eq!(h.push(buf), Ok(gst::FlowSuccess::Ok));
+        while let Some(buf) = h.try_pull() {
+            output_len += buf.get_size();
+            checksum = checksum.wrapping_add(
+                buf.map_readable()
+                    .unwrap()
+                    .iter()
+                    .fold(0u32, |s, v| s.wrapping_add(*v as u32)),
+            );
+        }
+        data = &data[l..];
+    }
+
+    h.push_event(gst::Event::new_eos().build());
+    while let Some(buf) = h.try_pull() {
+        output_len += buf.get_size();
+        checksum = checksum.wrapping_add(
+            buf.map_readable()
+                .unwrap()
+                .iter()
+                .fold(0u32, |s, v| s.wrapping_add(*v as u32)),
+        );
+    }
+
+    assert_eq!(input_len, 241152);
+    assert_eq!(output_len, 89084);
+    assert_eq!(checksum, 12554799);
+
+    let caps = h
+        .get_sinkpad()
+        .expect("harness has no sinkpad")
+        .get_current_caps()
+        .expect("pad has no caps");
+    assert_eq!(
+        caps,
+        gst::Caps::builder("closedcaption/x-cea-608")
+            .field("format", &"raw")
+            .field("framerate", &gst::Fraction::new(30000, 1001))
+            .build()
+    );
+}
+
+/// Test that ensures timecode parsing is the expected one
+#[test]
+fn test_timecodes() {
+    init();
+    let data = include_bytes!("timecodes-cut-down-sample.scc").as_ref();
+
+    let mut h = gst_check::Harness::new("sccparse");
+    h.set_src_caps_str("application/x-scc");
+
+    let timecodes = [
+        "00:00:00;00",
+        "00:00:14;01",
+        "00:00:17;26",
+        "00:00:19;01",
+        "00:00:21;02",
+        "00:00:23;10",
+        "00:00:25;18",
+        "00:00:28;13",
+        "00:00:30;29",
+        "00:00:34;29",
+        "00:00:37;27",
+        "00:00:40;01",
+        "00:00:43;27",
+        "00:00:45;13",
+        "00:00:49;16",
+        "00:58:51;01",
+        "00:58:52;29",
+        "00:58:55;00",
+        "00:59:00;25",
+    ];
+
+    let mut valid_timecodes: VecDeque<ValidVideoTimeCode> = timecodes
+        .iter()
+        .map(|s| {
+            let mut t = VideoTimeCode::from_string(s).unwrap();
+            t.set_fps(gst::Fraction::new(30000, 1001));
+            t.set_flags(gst_video::VideoTimeCodeFlags::DROP_FRAME);
+            t
+        })
+        .map(|t| t.try_into().unwrap())
+        .collect();
+
+    let mut output_len = 0;
+    let mut checksum = 0u32;
+    let mut expected_timecode = valid_timecodes.pop_front().unwrap();
+
+    let buf = gst::Buffer::from_mut_slice(Vec::from(&data[..]));
+    assert_eq!(h.push(buf), Ok(gst::FlowSuccess::Ok));
+    while let Some(buf) = h.try_pull() {
+        output_len += buf.get_size();
+        checksum = checksum.wrapping_add(
+            buf.map_readable()
+                .unwrap()
+                .iter()
+                .fold(0u32, |s, v| s.wrapping_add(*v as u32)),
+        );
+
+        // get the timecode of the buffer
+        let tc = buf
+            .get_meta::<gst_video::VideoTimeCodeMeta>()
+            .expect("No timecode meta")
+            .get_tc();
+
+        // if the timecode matches one of expected codes,
+        // pop the valid_timecodes deque and set expected_timecode,
+        // to the next timecode.
+        if Some(&tc) == valid_timecodes.front() {
+            expected_timecode = valid_timecodes.pop_front().unwrap();
+        }
+
+        assert_eq!(tc, expected_timecode);
+        expected_timecode.increment_frame();
+    }
+
+    assert_eq!(output_len, 1268);
+    assert_eq!(checksum, 174295);
+
+    let caps = h
+        .get_sinkpad()
+        .expect("harness has no sinkpad")
+        .get_current_caps()
+        .expect("pad has no caps");
+    assert_eq!(
+        caps,
+        gst::Caps::builder("closedcaption/x-cea-608")
+            .field("format", &"raw")
+            .field("framerate", &gst::Fraction::new(30000, 1001))
+            .build()
+    );
+}

gst-plugin-closedcaption/tests/timecodes-cut-down-sample.scc

@@ -0,0 +1,40 @@
+Scenarist_SCC V1.0
+
+00:00:00;00	942c 942c
+
+00:00:14;01	9420 9420 94ae 94ae 9454 9454 10ae 10ae 46f2 ef6d 20ce e5f7 20d9 eff2 6b2c 94f2 94f2 10ae 10ae f468 e973 20e9 7320 c4e5 6def e3f2 61e3 7920 ceef f7a1 942f 942f
+
+00:00:17;26	9420 9420 94ae 94ae 9452 9452 97a1 97a1 10ae 10ae d9e5 732c 942c 942c 2049 a76d 2073 7570 70ef f2f4 e96e 6780 94f4 94f4 10ae 10ae c4ef 6e61 ec64 2054 f275 6d70 ae80 942f 942f
+
+00:00:19;01	9420 9420 94ae 94ae 94d0 94d0 10ae 10ae 49a7 6d20 64ef e96e 6720 73ef 2061 7320 e56e f468 7573 e961 73f4 e9e3 61ec ec79 94f4 94f4 97a2 97a2 10ae 10ae 6173 2049 20e3 616e 2c80 942f 942f
+
+00:00:21;02	9420 9420 94ae 94ae 9452 9452 10ae 10ae e576 e56e 20f4 68e5 20e6 61e3 f420 4920 f468 e96e 6b80 9470 9470 97a1 97a1 10ae 10ae 68e5 a773 2061 20f4 e5f2 f2e9 62ec e520 6875 6d61 6e20 62e5 e96e 67ae 942f 942f
+
+00:00:23;10	9420 9420 94ae 94ae 94d0 94d0 10ae 10ae c275 f420 f468 e520 e368 efe9 e3e5 20ef 6e20 f468 e520 eff4 68e5 f220 73e9 64e5 94f2 94f2 9723 9723 10ae 10ae e973 20ea 7573 f420 6173 2062 6164 ae80 942f 942f
+
+00:00:25;18	9420 9420 94ae 94ae 9454 9454 97a2 97a2 10ae 10ae 54f2 756d 7020 e973 2061 9470 9470 9723 9723 10ae 10ae a2f4 e5f2 f2e9 62ec e520 6875 6d61 6e20 62e5 e96e 67ae a280 942f 942f
+
+00:00:28;13	9420 9420 94ae 94ae 9452 9452 97a1 97a1 10ae 10ae 5468 ef73 e520 61f2 e520 f468 942c 942c e520 f7ef f264 7380 94f2 94f2 97a2 97a2 10ae 10ae efe6 20cd e9e3 6b20 cd75 ec76 616e e579 2c80 942f 942f
+
+00:00:30;29	9420 9420 94ae 94ae 94d0 94d0 9723 9723 10ae 10ae f468 e520 6d61 6e20 54f2 756d 7020 6861 7320 e368 ef73 e56e 9470 9470 10ae 10ae f4ef 942c 942c 2062 e520 68e9 7320 6ee5 f720 e368 e9e5 e620 efe6 2073 f461 e6e6 ae80 942f 942f
+
+00:00:34;29	9420 9420 94ae 94ae 94d0 94d0 9723 9723 10ae 10ae 57e5 a7ec ec20 7370 e561 6b20 f7e9 f468 20e6 eff2 942c 942c 6de5 f280 94f2 94f2 10ae 10ae 70f2 e573 e964 e56e f4e9 61ec 20e3 616e 64e9 6461 f4e5 942f 942f
+
+00:00:37;27	9420 9420 94ae 94ae 9454 9454 97a1 97a1 10ae 10ae 5261 ec70 6820 ce61 64e5 f280 94f2 94f2 9723 9723 10ae 10ae 6162 ef75 f420 cd75 ec76 616e e579 2c80 942f 942f
+
+00:00:40;01	9420 9420 94ae 94ae 9452 9452 97a2 97a2 10ae 10ae f468 e520 f2e5 73e9 676e 61f4 e9ef 6e20 efe6 9470 9470 9723 9723 10ae 10ae 496e f4e5 f2e9 eff2 20d3 e5e3 f2e5 f461 f279 2052 7961 6e80 942f 942f
+
+00:00:43;27	9420 9420 94ae 94ae 94f2 94f2 97a1 97a1 10ae 10ae dae9 6e6b e52c 20f4 68e5 2070 ef73 73e9 62ec e580 942f 942f
+
+00:00:45;13	9420 9420 94ae 94ae 94d0 94d0 10ae 10ae 4368 f2e9 73f4 6d61 7320 67ef 76e5 f26e 6de5 6ef4 2073 6875 f464 eff7 6e80 94f2 94f2 10ae 10ae ef76 e5f2 20f4 68e5 2062 eff2 64e5 f220 f761 ecec 2c80 942f 942f
+
+00:00:49;16	9420 9420 94ae 94ae 94d0 94d0 97a1 97a1 10ae 10ae 616e 6420 f468 e520 e675 f475 f2e5 20ef e620 4f62 616d 61e3 61f2 e580 94f2 94f2 97a1 97a1 10ae 10ae 61e6 f4e5 f220 6120 e3ef 6e73 e5f2 7661 f4e9 76e5 942f 942f
+
+00:58:51;01	9420 9420 94ae 94ae 9452 9452 9723 9723 10ae 10ae f7e5 20f7 e9ec ec20 ece9 6e6b 20f4 ef2c 9470 9470 9723 9723 10ae 10ae a246 f2ef 6d20 c1f2 e97a ef6e 6120 f4ef 20d9 e56d e56e ba80 942f 942f
+
+00:58:52;29	9420 9420 94ae 94ae 9454 9454 97a1 97a1 10ae 10ae 5468 e520 4aef 75f2 6ee5 7980 94f2 94f2 10ae 10ae efe6 2061 6e20 c16d e5f2 e9e3 616e 20c2 ef6d 62ae a280 942f 942f
+
+00:58:55;00	9420 9420 94ae 94ae 9452 9452 9723 9723 10ae 10ae 49a7 6d20 c16d 7920 c7ef ef64 6d61 6eae 9470 9470 10ae 10ae 5468 616e 6b73 2073 ef20 6d75 e368 20e6 eff2 20ea efe9 6ee9 6e67 2075 73ae 942f 942f
+
+00:59:00;25	942c 942c
+