Manually mark how much data has been consumed from the source in muxer. (#3720)

Author: toots

src/core/operators/muxer.ml

@@ -87,7 +87,7 @@ class muxer tracks =
       let pos, frame =
         List.fold_left
           (fun (pos, frame) { fields; source } ->
-            let buf = source#get_frame in
+            let buf = source#peek_frame in
             ( min pos (Frame.position buf),
               List.fold_left
                 (fun frame { source_field; target_field; processor } ->
@@ -97,7 +97,10 @@ class muxer tracks =
           (length, Frame.create ~length Frame.Fields.empty)
           tracks
       in
-      Frame.slice frame pos
+      let frame = Frame.slice frame pos in
+      let consumed = Frame.position frame in
+      List.iter (fun { source } -> source#consumed consumed) tracks;
+      frame
   end
 
 let muxer_operator p =

src/core/source.ml

@@ -613,6 +613,7 @@ class virtual operator ?pos ?(name = "src") sources =
       consumed <- max consumed (Frame.position data);
       data
 
+    method consumed n = consumed <- max consumed n
     method get_frame = self#get_partial_frame (fun f -> f)
 
     method get_mutable_content field =

src/core/source.mli

@@ -258,12 +258,21 @@ class virtual source :
            that was effectively used. This method is used when a consumer of the source's data
            only uses an initial chunk of the frame. In this case, the remaining data is cached
            whenever possible and returned during the next streaming cycle. Final returned value
-           is the same as the partial chunk returned for the callback for easy method call chaining. *)
+           is the same as the partial chunk returned from the callback for easy method call chaining.
+
+           Calling this method is equivalent to doing: {[
+             let frame = Frame.slice source#peek_frame len in
+             source#consumed (Frame.position frame);
+             frame
+           ]} *)
        method get_partial_frame : (Frame.t -> Frame.t) -> Frame.t
 
        (** Check a frame without consuming any of its data. *)
        method peek_frame : Frame.t
 
+       (** Manually mark amount of consumed data from the source. *)
+       method consumed : int -> unit
+
        (** This method requests a specific field of the frame that can be mutated. It is used
            by a consumer of the source that will modify the source's data (e.g. [amplify]). The
            source will do its best to minimize data copy according to the streaming context. Typically,