Start forwarding timestamps if they're present in the stream (#50)

* Start forwarding timestamps if they're present in the stream

* Add file comparison test
---------

Co-authored-by: Mateusz Front <[email protected]>

README.md

@@ -14,7 +14,7 @@ Documentation is available at [HexDocs](https://hexdocs.pm/membrane_mp3_mad_plug
 Add the following line to your `deps` in `mix.exs`. Run `mix deps.get`.
 
 ```elixir
-	{:membrane_mp3_mad_plugin, "~> 0.18.2"}
+	{:membrane_mp3_mad_plugin, "~> 0.18.3"}
 ```
 
 This package depends on the [MAD](https://www.underbit.com/products/mad/) library. The precompiled build will be pulled and linked automatically. However, should there be any problems, consider installing it manually.

lib/membrane_mp3_mad_plugin/decoder.ex

@@ -9,6 +9,8 @@ defmodule Membrane.MP3.MAD.Decoder do
   alias Membrane.{Buffer, Logger, MPEGAudio, RawAudio, RemoteStream}
   alias Membrane.Event.Discontinuity
 
+  @samples_per_frame 1152
+
   def_input_pad :input, accepted_format: any_of(RemoteStream, MPEGAudio)
 
   def_output_pad :output, accepted_format: %RawAudio{sample_format: :s24le}
@@ -53,7 +55,7 @@ defmodule Membrane.MP3.MAD.Decoder do
   def handle_buffer(:input, buffer, ctx, state) do
     to_decode = state.queue <> buffer.payload
 
-    case decode_buffer(state.native, to_decode, ctx.pads.output.stream_format) do
+    case decode_buffer(state.native, to_decode, buffer.pts, ctx.pads.output.stream_format) do
       {:ok, {new_queue, actions}} ->
         {actions, %{state | queue: new_queue}}
 
@@ -108,13 +110,13 @@ defmodule Membrane.MP3.MAD.Decoder do
     |> Enum.reduce(0, fn {el, index}, acc -> acc ||| el <<< (index * 7) end)
   end
 
-  defp decode_buffer(native, buffer, stream_format, acc \\ [])
+  defp decode_buffer(native, buffer, pts, stream_format, acc \\ [])
 
-  defp decode_buffer(_native, <<>>, _stream_format, acc) do
+  defp decode_buffer(_native, <<>>, _pts, _stream_format, acc) do
     {:ok, {<<>>, Enum.reverse(acc)}}
   end
 
-  defp decode_buffer(native, buffer, stream_format, acc) when byte_size(buffer) > 0 do
+  defp decode_buffer(native, buffer, pts, stream_format, acc) when byte_size(buffer) > 0 do
     with {:ok, {decoded_frame, frame_size, sample_rate, channels}} <-
            Native.decode_frame(buffer, native) do
       new_stream_format = %RawAudio{
@@ -128,10 +130,19 @@ defmodule Membrane.MP3.MAD.Decoder do
           do: [],
           else: [stream_format: {:output, new_stream_format}]
 
-      buffer_action = [buffer: {:output, %Buffer{payload: decoded_frame}}]
+      buffer_action = [buffer: {:output, %Buffer{payload: decoded_frame, pts: pts}}]
 
       <<_used::binary-size(frame_size), rest::binary>> = buffer
-      decode_buffer(native, rest, new_stream_format, buffer_action ++ stream_format_action ++ acc)
+
+      next_pts = get_next_timestamp(pts, new_stream_format)
+
+      decode_buffer(
+        native,
+        rest,
+        next_pts,
+        new_stream_format,
+        buffer_action ++ stream_format_action ++ acc
+      )
     else
       {:error, :buflen} ->
         {:ok, {buffer, Enum.reverse(acc)}}
@@ -140,19 +151,31 @@ defmodule Membrane.MP3.MAD.Decoder do
         Logger.warning("Skipping malformed frame (#{bytes_to_skip} bytes)")
         <<_used::binary-size(bytes_to_skip), new_buffer::binary>> = buffer
 
+        next_pts = get_next_timestamp(pts, stream_format)
+
         case acc do
           [{:event, {:output, %Discontinuity{}}} | _actions] ->
             # send only one discontinuity event in a row
-            decode_buffer(native, new_buffer, stream_format, acc)
+            decode_buffer(native, new_buffer, next_pts, stream_format, acc)
 
           _no_event_on_top ->
             discontinuity = [event: {:output, %Discontinuity{}}]
-            decode_buffer(native, new_buffer, stream_format, discontinuity ++ acc)
+            decode_buffer(native, new_buffer, next_pts, stream_format, discontinuity ++ acc)
         end
 
       {:error, :malformed} ->
         Logger.warning("Terminating stream because of malformed frame")
         {:error, :malformed}
     end
   end
+
+  @spec get_next_timestamp(timestamp :: Membrane.Time.t() | nil, stream_format :: RawAudio.t()) ::
+          Membrane.Time.t() | nil
+  defp get_next_timestamp(nil, _stream_format) do
+    nil
+  end
+
+  defp get_next_timestamp(timestamp, stream_format) do
+    timestamp + RawAudio.frames_to_time(@samples_per_frame, stream_format)
+  end
 end

mix.exs

@@ -1,7 +1,7 @@
 defmodule Membrane.MP3.MAD.Plugin.Mixfile do
   use Mix.Project
 
-  @version "0.18.2"
+  @version "0.18.3"
   @github_url "https://github.com/membraneframework/membrane_mp3_mad_plugin"
 
   def project do
@@ -62,7 +62,9 @@ defmodule Membrane.MP3.MAD.Plugin.Mixfile do
       {:membrane_precompiled_dependency_provider, "~> 0.1.0"},
       {:ex_doc, "~> 0.28", only: :dev, runtime: false},
       {:credo, "~> 1.6", only: :dev, runtime: false},
-      {:dialyxir, "~> 1.1", only: :dev, runtime: false}
+      {:dialyxir, "~> 1.1", only: :dev, runtime: false},
+      # testing deps
+      {:membrane_file_plugin, "~> 0.17.0", only: :test}
     ]
   end
 end