textwrap: fix text accumulating for longer than accumulate-time

By draining before *and* after the word wrapping loop.

We don't do that during the loop in order not to break up sentences
unnecessarily.

Also adds tests

Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs/-/merge_requests/1930>

text/wrap/src/gsttextwrap/imp.rs

@@ -125,6 +125,43 @@ impl TextWrap {
         state.options = Some(options);
     }
 
+    fn try_drain(
+        &self,
+        state: &mut State,
+        pts: gst::ClockTime,
+        accumulate_time: gst::ClockTime,
+        bufferlist: &mut gst::BufferList,
+    ) {
+        let add_buffer = state
+            .start_ts
+            .opt_add(accumulate_time)
+            .opt_le(pts)
+            .unwrap_or(false);
+
+        if add_buffer {
+            let drained = mem::take(&mut state.current_text);
+            let duration = state.end_ts.opt_checked_sub(state.start_ts).ok().flatten();
+            gst::debug!(
+                CAT,
+                imp = self,
+                "Outputting contents {}, ts: {}, duration: {}",
+                drained,
+                state.start_ts.display(),
+                duration.display(),
+            );
+            let mut buf = gst::Buffer::from_mut_slice(drained.into_bytes());
+            {
+                let buf_mut = buf.get_mut().unwrap();
+                buf_mut.set_pts(state.start_ts);
+                buf_mut.set_duration(duration);
+            }
+            bufferlist.get_mut().unwrap().add(buf);
+
+            state.start_ts = None;
+            state.end_ts = None;
+        }
+    }
+
     fn sink_chain(
         &self,
         _pad: &gst::Pad,
@@ -166,6 +203,8 @@ impl TextWrap {
             let mut bufferlist = gst::BufferList::new();
             let n_lines = std::cmp::max(self.settings.lock().unwrap().lines, 1);
 
+            self.try_drain(&mut state, pts, accumulate_time, &mut bufferlist);
+
             let add_buffer = state
                 .start_ts
                 .opt_add(accumulate_time)
@@ -199,10 +238,10 @@ impl TextWrap {
             let duration_per_word = (num_words != 0).then(|| duration / num_words);
 
             if state.start_ts.is_none() {
-                state.start_ts = buffer.pts();
+                state.start_ts = Some(pts);
             }
 
-            state.end_ts = buffer.pts();
+            state.end_ts = Some(pts);
 
             let words = data.split_ascii_whitespace();
             let mut current_text = state.current_text.to_string();
@@ -260,6 +299,11 @@ impl TextWrap {
             }
 
             state.current_text = current_text;
+            state.end_ts = Some(pts + duration);
+
+            if let Some(pts) = state.end_ts {
+                self.try_drain(&mut state, pts, accumulate_time, &mut bufferlist);
+            }
 
             if state.current_text.is_empty() {
                 state.start_ts = None;

text/wrap/tests/textwrap.rs

@@ -112,3 +112,114 @@ fn test_lines() {
         std::str::from_utf8(expected_output.as_ref())
     );
 }
+
+#[test]
+fn test_accumulate_time_gaps() {
+    init();
+
+    let input = b"First buffer content";
+
+    let mut h = gst_check::Harness::new("textwrap");
+
+    {
+        let wrap = h.element().expect("Could not create textwrap");
+        wrap.set_property("accumulate-time", 2_000_000_000u64);
+        wrap.set_property("columns", u32::MAX);
+    }
+
+    h.set_src_caps_str("text/x-raw, format=utf8");
+
+    let buf = {
+        let mut buf = gst::Buffer::from_mut_slice(Vec::from(&input[..]));
+        let buf_ref = buf.get_mut().unwrap();
+        buf_ref.set_pts(gst::ClockTime::ZERO);
+        buf_ref.set_duration(1.seconds());
+        buf
+    };
+
+    assert_eq!(h.push(buf), Ok(gst::FlowSuccess::Ok));
+
+    assert_eq!(h.buffers_in_queue(), 0);
+
+    // A long-enough gap should cause the first buffer to come out
+
+    let gap = gst::event::Gap::builder(1.seconds())
+        .duration(1_000.mseconds())
+        .build();
+
+    assert!(h.push_event(gap));
+
+    assert_eq!(h.buffers_in_queue(), 1);
+    let buf = h.pull().expect("Couldn't pull buffer");
+
+    assert_eq!(buf.pts(), Some(gst::ClockTime::ZERO));
+    assert_eq!(buf.duration(), Some(2.seconds()));
+
+    let map = buf.map_readable().expect("Couldn't map buffer readable");
+
+    let expected_output = b"First buffer content";
+
+    assert_eq!(
+        std::str::from_utf8(map.as_ref()),
+        std::str::from_utf8(expected_output.as_ref())
+    );
+}
+
+#[test]
+fn test_accumulate_time_long_second_buffer() {
+    init();
+
+    let input = b"First buffer content";
+
+    let mut h = gst_check::Harness::new("textwrap");
+
+    {
+        let wrap = h.element().expect("Could not create textwrap");
+        wrap.set_property("accumulate-time", 2_000_000_000u64);
+        wrap.set_property("columns", u32::MAX);
+    }
+
+    h.set_src_caps_str("text/x-raw, format=utf8");
+
+    let buf = {
+        let mut buf = gst::Buffer::from_mut_slice(Vec::from(&input[..]));
+        let buf_ref = buf.get_mut().unwrap();
+        buf_ref.set_pts(gst::ClockTime::ZERO);
+        buf_ref.set_duration(1.seconds());
+        buf
+    };
+
+    assert_eq!(h.push(buf), Ok(gst::FlowSuccess::Ok));
+
+    assert_eq!(h.buffers_in_queue(), 0);
+
+    let input = b"Second buffer content";
+
+    let buf = {
+        let mut buf = gst::Buffer::from_mut_slice(Vec::from(&input[..]));
+        let buf_ref = buf.get_mut().unwrap();
+        buf_ref.set_pts(1.seconds());
+        buf_ref.set_duration(2.seconds());
+        buf
+    };
+
+    assert_eq!(h.push(buf), Ok(gst::FlowSuccess::Ok));
+
+    assert_eq!(h.buffers_in_queue(), 1);
+
+    let buf = h.pull().expect("Couldn't pull buffer");
+
+    // The contents of the second buffer should not be broken up
+
+    assert_eq!(buf.pts(), Some(gst::ClockTime::ZERO));
+    assert_eq!(buf.duration(), Some(3.seconds()));
+
+    let map = buf.map_readable().expect("Couldn't map buffer readable");
+
+    let expected_output = b"First buffer content Second buffer content";
+
+    assert_eq!(
+        std::str::from_utf8(map.as_ref()),
+        std::str::from_utf8(expected_output.as_ref())
+    );
+}