Rewrite ISOM muxer on TimestampQueue (#111)

* Rewrite ISOM Muxer on auto demands with TimestampQueue

* WIP remove dependency to Track

* Revert "WIP remove dependency to Track"

This reverts commit ef837ec.

* Remove leftovers

* Delete manual ISOM

* Format

* Fix deps

* Refactor due to CR

Author: FelonEkonom

lib/membrane_mp4/muxer/isom.ex

@@ -4,16 +4,15 @@ defmodule Membrane.MP4.Muxer.ISOM do
   """
   use Membrane.Filter
 
-  alias Membrane.{Buffer, File, MP4, RemoteStream, Time}
+  alias Membrane.{Buffer, File, MP4, RemoteStream, Time, TimestampQueue}
   alias Membrane.MP4.{Container, FileTypeBox, MediaDataBox, MovieBox, Track}
 
   @ftyp FileTypeBox.assemble("isom", ["isom", "iso2", "avc1", "mp41"])
   @ftyp_size @ftyp |> Container.serialize!() |> byte_size()
   @mdat_header_size 8
 
   def_input_pad :input,
-    flow_control: :manual,
-    demand_unit: :buffers,
+    flow_control: :auto,
     accepted_format:
       any_of(
         %Membrane.AAC{config: {:esds, _esds}},
@@ -31,7 +30,7 @@ defmodule Membrane.MP4.Muxer.ISOM do
 
   def_output_pad :output,
     accepted_format: %RemoteStream{type: :bytestream, content_format: MP4},
-    flow_control: :manual
+    flow_control: :auto
 
   def_options fast_start: [
                 spec: boolean(),
@@ -61,27 +60,36 @@ defmodule Membrane.MP4.Muxer.ISOM do
 
   @impl true
   def handle_init(_ctx, options) do
+    queue =
+      TimestampQueue.new(
+        chunk_duration: options.chunk_duration,
+        pause_demand_boundary: 3 * options.chunk_duration,
+        pause_demand_boundary_unit: :time
+      )
+
     state =
       options
       |> Map.from_struct()
       |> Map.merge(%{
         mdat_size: 0,
         next_track_id: 1,
         pad_order: [],
-        pad_to_track: %{}
+        pad_to_track: %{},
+        queue: queue
       })
 
     {[], state}
   end
 
   @impl true
-  def handle_pad_added(Pad.ref(:input, pad_ref), _ctx, state) do
+  def handle_pad_added(Pad.ref(:input, pad_ref) = pad, _ctx, state) do
     {track_id, state} = Map.get_and_update!(state, :next_track_id, &{&1, &1 + 1})
 
     state =
       state
       |> Map.update!(:pad_order, &[pad_ref | &1])
       |> put_in([:pad_to_track, pad_ref], track_id)
+      |> Map.update!(:queue, &TimestampQueue.register_pad(&1, pad))
 
     {[], state}
   end
@@ -126,34 +134,59 @@ defmodule Membrane.MP4.Muxer.ISOM do
   end
 
   @impl true
-  def handle_demand(:output, _size, :buffers, _ctx, state) do
-    next_ref = hd(state.pad_order)
+  def handle_event(Pad.ref(:input, _pad_ref) = pad, event, _ctx, state) do
+    state.queue
+    |> TimestampQueue.push_event(pad, event)
+    |> TimestampQueue.pop_available_items()
+    |> handle_queue_output(state)
+  end
+
+  @impl true
+  def handle_event(:output, event, _ctx, state) do
+    {[forward: event], state}
+  end
 
-    {[demand: {Pad.ref(:input, next_ref), 1}], state}
+  @impl true
+  def handle_buffer(Pad.ref(:input, _pad_ref) = pad, buffer, _ctx, state) do
+    state.queue
+    |> TimestampQueue.push_buffer_and_pop_available_items(pad, buffer)
+    |> handle_queue_output(state)
   end
 
   @impl true
-  def handle_buffer(Pad.ref(:input, pad_ref), buffer, _ctx, state) do
+  def handle_end_of_stream(Pad.ref(:input, _pad_ref) = pad, _ctx, state) do
+    state.queue
+    |> TimestampQueue.push_end_of_stream(pad)
+    |> TimestampQueue.pop_available_items()
+    |> handle_queue_output(state)
+  end
+
+  defp handle_queue_output({suggested_actions, batch, queue}, state) do
+    {actions, state} = Enum.flat_map_reduce(batch, state, &handle_queue_item/2)
+    {suggested_actions ++ actions, %{state | queue: queue}}
+  end
+
+  defp handle_queue_item({_pad_ref, {:event, event}}, state) do
+    {[event: {:output, event}], state}
+  end
+
+  defp handle_queue_item({Pad.ref(:input, pad_ref), {:buffer, buffer}}, state) do
     # In case DTS is not set, use PTS. This is the case for audio tracks or H264 originated
     # from an RTP stream. ISO base media file format specification uses DTS for calculating
     # decoding deltas, and so is the implementation of sample table in this plugin.
     buffer = %Buffer{buffer | dts: Buffer.get_dts_or_pts(buffer)}
 
-    {maybe_buffer, state} =
-      state
-      |> update_in([:pad_to_track, pad_ref], &Track.store_sample(&1, buffer))
-      |> maybe_flush_chunk(pad_ref)
-
-    {maybe_buffer ++ [redemand: :output], state}
+    state
+    |> update_in([:pad_to_track, pad_ref], &Track.store_sample(&1, buffer))
+    |> maybe_flush_chunk(pad_ref)
   end
 
-  @impl true
-  def handle_end_of_stream(Pad.ref(:input, pad_ref), _ctx, state) do
+  defp handle_queue_item({Pad.ref(:input, pad_ref), :end_of_stream}, state) do
     {buffer, state} = do_flush_chunk(state, pad_ref)
     state = Map.update!(state, :pad_order, &List.delete(&1, pad_ref))
 
     if state.pad_order != [] do
-      {buffer ++ [redemand: :output], state}
+      {buffer, state}
     else
       actions = finalize_mp4(state)
       {buffer ++ actions ++ [end_of_stream: :output], state}

mix.exs

@@ -45,6 +45,7 @@ defmodule Membrane.MP4.Plugin.MixProject do
       {:membrane_h265_format, "~> 0.2.0"},
       {:membrane_opus_format, "~> 0.3.0"},
       {:membrane_file_plugin, "~> 0.17.0"},
+      {:membrane_timestamp_queue, "~> 0.1.0"},
       {:bunch, "~> 1.5"},
       {:membrane_h26x_plugin, "~> 0.10.0", only: :test},
       {:membrane_aac_plugin, "~> 0.18.0", only: :test},

mix.lock

@@ -14,6 +14,7 @@
   "ex_doc": {:hex, :ex_doc, "0.31.1", "8a2355ac42b1cc7b2379da9e40243f2670143721dd50748bf6c3b1184dae2089", [:mix], [{:earmark_parser, "~> 1.4.39", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_c, ">= 0.1.1", [hex: :makeup_c, repo: "hexpm", optional: true]}, {:makeup_elixir, "~> 0.14", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1", [hex: :makeup_erlang, repo: "hexpm", optional: false]}], "hexpm", "3178c3a407c557d8343479e1ff117a96fd31bafe52a039079593fb0524ef61b0"},
   "file_system": {:hex, :file_system, "1.0.0", "b689cc7dcee665f774de94b5a832e578bd7963c8e637ef940cd44327db7de2cd", [:mix], [], "hexpm", "6752092d66aec5a10e662aefeed8ddb9531d79db0bc145bb8c40325ca1d8536d"},
   "finch": {:hex, :finch, "0.18.0", "944ac7d34d0bd2ac8998f79f7a811b21d87d911e77a786bc5810adb75632ada4", [:mix], [{:castore, "~> 0.1 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: false]}, {:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:mint, "~> 1.3", [hex: :mint, repo: "hexpm", optional: false]}, {:nimble_options, "~> 0.4 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:nimble_pool, "~> 0.2.6 or ~> 1.0", [hex: :nimble_pool, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "69f5045b042e531e53edc2574f15e25e735b522c37e2ddb766e15b979e03aa65"},
+  "heap": {:hex, :heap, "2.0.2", "d98cb178286cfeb5edbcf17785e2d20af73ca57b5a2cf4af584118afbcf917eb", [:mix], [], "hexpm", "ba9ea2fe99eb4bcbd9a8a28eaf71cbcac449ca1d8e71731596aace9028c9d429"},
   "hpax": {:hex, :hpax, "0.1.2", "09a75600d9d8bbd064cdd741f21fc06fc1f4cf3d0fcc335e5aa19be1a7235c84", [:mix], [], "hexpm", "2c87843d5a23f5f16748ebe77969880e29809580efdaccd615cd3bed628a8c13"},
   "jason": {:hex, :jason, "1.4.1", "af1504e35f629ddcdd6addb3513c3853991f694921b1b9368b0bd32beb9f1b63", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "fbb01ecdfd565b56261302f7e1fcc27c4fb8f32d56eab74db621fc154604a7a1"},
   "logger_backends": {:hex, :logger_backends, "1.0.0", "09c4fad6202e08cb0fbd37f328282f16539aca380f512523ce9472b28edc6bdf", [:mix], [], "hexpm", "1faceb3e7ec3ef66a8f5746c5afd020e63996df6fd4eb8cdb789e5665ae6c9ce"},
@@ -36,6 +37,7 @@
   "membrane_precompiled_dependency_provider": {:hex, :membrane_precompiled_dependency_provider, "0.1.1", "a0d5b7942f8be452c30744207f78284f6a0e0c84c968aba7d76e206fbf75bc5d", [:mix], [{:bundlex, "~> 1.4", [hex: :bundlex, repo: "hexpm", optional: false]}], "hexpm", "87ad44752e2cf0fa3b31c5aac15b863343c2f6e0f0fd201f5ec4c0bcda8c6fa3"},
   "membrane_raw_audio_format": {:hex, :membrane_raw_audio_format, "0.12.0", "b574cd90f69ce2a8b6201b0ccf0826ca28b0fbc8245b8078d9f11cef65f7d5d5", [:mix], [{:bimap, "~> 1.1", [hex: :bimap, repo: "hexpm", optional: false]}, {:bunch, "~> 1.0", [hex: :bunch, repo: "hexpm", optional: false]}, {:membrane_core, "~> 1.0", [hex: :membrane_core, repo: "hexpm", optional: false]}], "hexpm", "6e6c98e3622a2b9df19eab50ba65d7eb45949b1ba306fa8423df6cdb12fd0b44"},
   "membrane_stream_plugin": {:hex, :membrane_stream_plugin, "0.4.0", "0c4ab72a4e13bf0faa0f1166fbaf68d2e34167dbec345aedb74ce1eb7497bdda", [:mix], [{:membrane_core, "~> 1.0", [hex: :membrane_core, repo: "hexpm", optional: false]}], "hexpm", "5a9a9c17783e18ad740e6ddfed364581bdb7ebdab8e61ba2c19a1830356f7eb8"},
+  "membrane_timestamp_queue": {:hex, :membrane_timestamp_queue, "0.1.0", "98cfc057b845d690da9d57dd5f80e32a3aae2c43b52fac9aade07146113f3a89", [:mix], [{:heap, "~> 2.0", [hex: :heap, repo: "hexpm", optional: false]}, {:membrane_core, "~> 1.0", [hex: :membrane_core, repo: "hexpm", optional: false]}], "hexpm", "7aa2912839a85e93b5c92cd52a1e53b719e39f3231dc3e49b7cadf182e7345b8"},
   "mime": {:hex, :mime, "2.0.5", "dc34c8efd439abe6ae0343edbb8556f4d63f178594894720607772a041b04b02", [:mix], [], "hexpm", "da0d64a365c45bc9935cc5c8a7fc5e49a0e0f9932a761c55d6c52b142780a05c"},
   "mint": {:hex, :mint, "1.5.2", "4805e059f96028948870d23d7783613b7e6b0e2fb4e98d720383852a760067fd", [:mix], [{:castore, "~> 0.1.0 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:hpax, "~> 0.1.1", [hex: :hpax, repo: "hexpm", optional: false]}], "hexpm", "d77d9e9ce4eb35941907f1d3df38d8f750c357865353e21d335bdcdf6d892a02"},
   "nimble_options": {:hex, :nimble_options, "1.1.0", "3b31a57ede9cb1502071fade751ab0c7b8dbe75a9a4c2b5bbb0943a690b63172", [:mix], [], "hexpm", "8bbbb3941af3ca9acc7835f5655ea062111c9c27bcac53e004460dfd19008a99"},

test/fixtures/isom/ref_two_tracks.mp4

@@ -345,6 +345,8 @@ defmodule Membrane.MP4.Muxer.CMAF.IntegrationTest do
       assert_in_delta buffer.metadata.duration, Membrane.Time.seconds(2), @video_sample_duration
 
       assert_end_of_stream(pipeline, :sink)
+
+      Testing.Pipeline.terminate(pipeline)
     end
 
     @tag segment_min_duration: Membrane.Time.seconds(6)
@@ -362,6 +364,7 @@ defmodule Membrane.MP4.Muxer.CMAF.IntegrationTest do
       assert_in_delta buffer.metadata.duration, ctx.segment_min_duration, @video_sample_duration
 
       assert_end_of_stream(pipeline, :sink)
+      Testing.Pipeline.terminate(pipeline)
     end
 
     @tag segment_min_duration: Membrane.Time.seconds(6)
@@ -405,6 +408,7 @@ defmodule Membrane.MP4.Muxer.CMAF.IntegrationTest do
       assert buffer.metadata.last_chunk?
 
       assert_end_of_stream(pipeline, :sink)
+      Testing.Pipeline.terminate(pipeline)
     end
   end
 

test/fixtures/isom/ref_two_tracks_fast_start.mp4

@@ -196,7 +196,10 @@ defmodule Membrane.MP4.Muxer.ISOM.IntegrationTest do
           out_encapsulation: :none,
           output_config: :esds
         }),
-        child(:muxer, %Membrane.MP4.Muxer.ISOM{chunk_duration: Time.seconds(1), fast_start: true})
+        child(:muxer, %Membrane.MP4.Muxer.ISOM{
+          chunk_duration: Time.seconds(1),
+          fast_start: true
+        })
         |> child(:sink, %Membrane.File.Sink{location: out_path_for("two_tracks_fast_start")}),
         get_child(:video_parser) |> get_child(:muxer),
         get_child(:audio_parser) |> get_child(:muxer)
@@ -228,8 +231,9 @@ defmodule Membrane.MP4.Muxer.ISOM.IntegrationTest do
 
       assert_receive {:DOWN, ^monitor_ref, :process, ^pid,
                       {:membrane_child_crash, :muxer,
-                       {%RuntimeError{message: "ISOM Muxer doesn't support variable parameters"},
-                        _stacktrace}}},
+                       {%RuntimeError{
+                          message: "ISOM Muxer doesn't support variable parameters"
+                        }, _stacktrace}}},
                      1_000
     end
   end