wip3

Author: sgfn

lib/ex_webrtc/rtp/av1/depayloader.ex

@@ -14,26 +14,16 @@ defmodule ExWebRTC.RTP.Depayloader.AV1 do
 
   alias ExWebRTC.RTP.AV1.{OBU, Payload}
 
-  @temporal_delimiter %OBU{
-                        type: 2,
-                        x: 0,
-                        s: 1,
-                        payload: <<>>
-                      }
-                      |> OBU.serialize()
-
   @type t :: %__MODULE__{
           current_temporal_unit: [OBU.t()],
           current_obu: binary() | nil,
           current_timestamp: ExRTP.Packet.uint32() | nil
         }
 
-  defstruct [current_temporal_unit: [], current_obu: nil, current_timestamp: nil]
+  defstruct current_temporal_unit: [], current_obu: nil, current_timestamp: nil
 
-  # XXX where warnings, where debugs?
-  # XXX MUST ignore TD OBU, Tile List OBU
   @impl true
-  def new() do
+  def new do
     %__MODULE__{}
   end
 
@@ -45,21 +35,51 @@ defmodule ExWebRTC.RTP.Depayloader.AV1 do
   def depayload(depayloader, packet) do
     case Payload.parse(packet.payload) do
       {:ok, av1_payload} ->
-        do_depayload2(depayloader, packet, av1_payload)
+        do_depayload(depayloader, packet, av1_payload)
 
       {:error, reason} ->
         Logger.warning("""
         Couldn't parse payload, reason: #{reason}. \
         Resetting depayloader state. Payload: #{inspect(packet.payload)}.\
         """)
 
-        {:ok, %{depayloader | current_temporal_unit: [], current_timestamp: nil}}
+        {:ok, %__MODULE__{}}
+    end
+  end
+
+  defp do_depayload(depayloader, packet, %Payload{z: z, y: y} = av1_payload) do
+    {obus, current_obu_fragment, next_obu_fragment} =
+      av1_payload
+      |> Payload.depayload_obu_elements()
+      |> parse_obu_elements(z, y)
+
+    # TODO: handle marker, or not (?)
+    #         av1-rtp-spec sec. 4.2.: It is possible for a receiver to receive the last packet of a temporal unit
+    #         without the marker bit being set equal to 1, and a receiver should be able to handle this case.
+    #       at the moment, we're looking at the timestamps only, and it seems to work
+    #
+    # TODO: handle the case where depayloader.current_timestamp > packet.timestamp
+    new_temporal_unit? = depayloader.current_timestamp != packet.timestamp
+
+    {depayloader, obus, next_obu_fragment} =
+      depayloader
+      |> update_current_obu(current_obu_fragment, new_temporal_unit?)
+      |> maybe_flush_current_obu(obus, next_obu_fragment, y)
+
+    if new_temporal_unit? do
+      {temporal_unit, depayloader} = flush_temporal_unit(depayloader)
+
+      {temporal_unit,
+       update_temporal_unit(depayloader, obus, next_obu_fragment, packet.timestamp)}
+    else
+      {nil, update_temporal_unit(depayloader, obus, next_obu_fragment, packet.timestamp)}
     end
   end
 
   defp parse_obu_elements(obu_elements, z, y)
 
   defp parse_obu_elements([], _, _) do
+    # TODO: decide where to use debug logs and where warnings
     Logger.debug("AV1 payload contains no valid OBU elements. Dropping packet.")
     {[], nil, nil}
   end
@@ -85,7 +105,7 @@ defmodule ExWebRTC.RTP.Depayloader.AV1 do
     {obus, current_obu_fragment, next_obu_fragment}
   end
 
-  defp update_current_obu(depayloader, current_obu_fragment, new_frame?)
+  defp update_current_obu(depayloader, current_obu_fragment, new_temporal_unit?)
 
   defp update_current_obu(depayloader, current_obu_fragment, true) do
     if depayloader.current_obu != nil do
@@ -95,7 +115,9 @@ defmodule ExWebRTC.RTP.Depayloader.AV1 do
     end
 
     if current_obu_fragment != nil do
-      Logger.debug("Received ")
+      Logger.debug(
+        "Received middle OBU fragment from a new temporal unit without beginning the OBU. Dropping this OBU fragment."
+      )
     end
 
     %{depayloader | current_obu: nil}
@@ -136,7 +158,7 @@ defmodule ExWebRTC.RTP.Depayloader.AV1 do
   end
 
   # Packet contained exactly 1 OBU fragment, current_obu will be continued. Do not flush
-  # XXX check the nil
+  # TODO: make sure `nil` is correct here
   defp maybe_flush_current_obu(%__MODULE__{current_obu: incomplete_obu} = depayloader, [], nil, 1) do
     {depayloader, [], incomplete_obu}
   end
@@ -165,100 +187,43 @@ defmodule ExWebRTC.RTP.Depayloader.AV1 do
     }
   end
 
-  # XXX UWAGA NA TIMESTAMPA ŚMIERDZIELA
-  defp flush_temporal_unit(%__MODULE__{current_temporal_unit: tu} = depayloader) when tu != [] do
-    # Force s=1 for the low overhead bitstring format
+  defp flush_temporal_unit(%__MODULE__{current_temporal_unit: tu}) when tu != [] do
+    # Force s=1 for the low overhead bitstream format
     tu_binary =
       tu
       |> Stream.map(&%OBU{&1 | s: 1})
       |> Stream.map(&OBU.serialize/1)
       |> Enum.reverse()
       |> :erlang.iolist_to_binary()
 
-    # if current obu not nil, log
-    {@temporal_delimiter <> tu_binary,
-     %{depayloader | current_temporal_unit: [], current_obu: nil, current_timestamp: nil}}
+    # TODO: is it possible that `current_obu != nil` here?
+    {OBU.temporal_delimiter() <> tu_binary, %__MODULE__{}}
   end
 
   defp flush_temporal_unit(depayloader) do
-    Logger.debug("WRITEME")
-    # XXX maybe zero?
+    Logger.debug("Previous temporal unit is empty, nothing to flush")
     {nil, depayloader}
   end
 
-  defp do_depayload2(depayloader, packet, %Payload{z: z, y: y} = av1_payload) do
-    {obus, current_obu_fragment, next_obu_fragment} =
-      av1_payload
-      |> Payload.depayload_obu_elements()
-      |> parse_obu_elements(z, y)
-
-    # {[A], 0, 0} -> {[A], nil, nil}
-    # {[A], 0, 1} -> {[], nil, A}
-    # {[A], 1, 0} -> {[], A, nil}
-    # {[A], 1, 1} -> {[], A, nil}
-    #
-    # {[B, C, D], 0, 0} -> {[B, C, D], nil, nil}
-    # {[B, C, D], 0, 1} -> {[B, C], nil, D}
-    # {[B, C, D], 1, 0} -> {[C, D], B, nil}
-    # {[B, C, D], 1, 1} -> {[C], B, D}
-
-    new_temporal_unit? = depayloader.current_timestamp != packet.timestamp
-
-    {depayloader, obus, next_obu_fragment} =
-      depayloader
-      |> update_current_obu(current_obu_fragment, new_temporal_unit?)
-      |> maybe_flush_current_obu(obus, next_obu_fragment, y)
-
-    # {[A], 0, 0} -> nil
-    # {[A], 0, 1} -> nil
-    # {[A], 1, 0} -> if CO != nil, do: CO <> A, else: nil
-    # {[A], 1, 1} -> if CO != nil, do: CO <> A, else: nil
-    #
-    # {[B, C, D], 0, 0} -> nil
-    # {[B, C, D], 0, 1} -> nil
-    # {[B, C, D], 1, 0} -> if CO != nil, do: CO <> B, else: nil
-    # {[B, C, D], 1, 1} -> if CO != nil, do: CO <> B, else: nil
-
-    # {[A], 0, 0} -> {nil, [A]}
-    # {[A], 0, 1} -> {A, []}
-    # {[A], 1, _}, cat err -> {nil, []}
-    # {[A], 1, 0}, cat ok -> {nil, [CO]}
-    # {[A], 1, 1}, cat ok -> {CO, []}
-    #
-    # {[B, C, D], 0, 0} -> {nil, [B, C, D]}
-    # {[B, C, D], 0, 1} -> {D, [B, C]}
-    # {[B, C, D], 1, 0}, cat err -> {nil, [C, D]}
-    # {[B, C, D], 1, 1}, cat err -> {D, [C]}
-    # {[B, C, D], 1, 0}, cat ok -> {nil, [CO, C, D]}
-    # {[B, C, D], 1, 1}, cat ok -> {D, [CO, C]}
-
-    # XXX what to do if we're ready to flush two?
-    # depayloader
-    # |> update_temporal_unit(obus, next_obu_fragment)
-    # |> maybe_flush_temporal_unit(packet)
-
-    if new_temporal_unit? do
-      {temporal_unit, depayloader} = flush_temporal_unit(depayloader)
-
-      {temporal_unit,
-       update_temporal_unit(depayloader, obus, next_obu_fragment, packet.timestamp)}
-    else
-      {nil, update_temporal_unit(depayloader, obus, next_obu_fragment, packet.timestamp)}
-    end
-  end
-
   defp append_obus([], tu), do: tu
 
   defp append_obus([obu_binary | rest], tu) do
-    case OBU.parse(obu_binary) do
-      {:ok, obu, rest_of_binary} ->
-        if rest_of_binary != <<>>, do: Logger.debug("WRITEME")
-
-        append_obus(rest, [obu | tu])
-
+    with {:ok, obu, rest_of_binary} <- OBU.parse(obu_binary),
+         true <- OBU.should_be_transmitted?(obu) do
+      if rest_of_binary != <<>>,
+        do:
+          Logger.debug(
+            "OBU binary contains additional data after the decoded OBU, dropping the additional data"
+          )
+
+      append_obus(rest, [obu | tu])
+    else
       {:error, :invalid_av1_bitstream} ->
-        Logger.debug("WRITEME")
+        Logger.debug("Unable to parse OBU from binary data, dropping")
+        append_obus(rest, tu)
 
+      false ->
+        Logger.debug("Dropping temporal delimiter/tile list OBU")
         append_obus(rest, tu)
     end
   end

lib/ex_webrtc/rtp/av1/obu.ex

@@ -24,6 +24,7 @@ defmodule ExWebRTC.RTP.AV1.OBU do
 
   @obu_sequence_header 1
   @obu_temporal_delimiter 2
+  @obu_tile_list 8
   @obu_padding 15
 
   @type t :: %__MODULE__{
@@ -145,4 +146,19 @@ defmodule ExWebRTC.RTP.AV1.OBU do
   end
 
   def disable_dropping_in_decoder_if_applicable(obu), do: obu
+
+  @spec temporal_delimiter :: binary()
+  def temporal_delimiter,
+    do: %__MODULE__{type: @obu_temporal_delimiter, x: 0, s: 1, payload: <<>>} |> serialize()
+
+  @doc """
+  Determines whether the OBU should be removed when transmitting, and must be ignored when receiving
+  in accordance with av1-rtp-spec sec. 5.
+  """
+  @spec should_be_transmitted?(t()) :: boolean()
+  def should_be_transmitted?(obu)
+
+  def should_be_transmitted?(%__MODULE__{type: @obu_temporal_delimiter}), do: false
+  def should_be_transmitted?(%__MODULE__{type: @obu_tile_list}), do: false
+  def should_be_transmitted?(_obu), do: true
 end

lib/ex_webrtc/rtp/av1/payload.ex

@@ -112,7 +112,7 @@ defmodule ExWebRTC.RTP.AV1.Payload do
 
   defp parse_obu_elements(<<>>, count, obu_elements) do
     if count > 0,
-      do: Logger.warning("Invalid AV1 RTP payload: expected #{count} more OBU elements.")
+      do: Logger.debug("Invalid AV1 RTP aggregation header: expected #{count} more OBU elements.")
 
     Enum.reverse(obu_elements)
   end

lib/ex_webrtc/rtp/av1/payloader.ex

@@ -16,7 +16,6 @@ defmodule ExWebRTC.RTP.Payloader.AV1 do
   alias ExWebRTC.Utils
 
   @obu_sequence_header 1
-  @obu_temporal_delimiter 2
 
   @aggregation_header_size_bytes 1
 
@@ -36,25 +35,14 @@ defmodule ExWebRTC.RTP.Payloader.AV1 do
   def payload(payloader, temporal_unit) when temporal_unit != <<>> do
     # In AV1, a temporal unit consists of all OBUs associated with a specific time instant.
     # Temporal units always start with a temporal delimiter OBU. They may contain multiple AV1 frames.
-    #   av1-rtp-spec sec. 5: The temporal delimiter OBU should be removed when transmitting.
-    obus =
-      case parse_obus(temporal_unit) do
-        [%OBU{type: @obu_temporal_delimiter} | next_obus] ->
-          next_obus
-
-        _ ->
-          Logger.warning("""
-          Invalid AV1 temporal unit: does not start with temporal delimiter OBU. \
-          Dropping temporal unit.\
-          """)
-
-          []
-      end
 
     # With the current implementation, each RTP packet will contain one OBU element.
     # This element can be an entire OBU, or a fragment of an OBU bigger than max_payload_size.
     rtp_packets =
-      Stream.flat_map(obus, fn obu ->
+      temporal_unit
+      |> parse_obus()
+      |> Stream.filter(&OBU.should_be_transmitted?/1)
+      |> Stream.flat_map(fn obu ->
         n_bit = Utils.to_int(obu.type == @obu_sequence_header)
 
         obu

lib/ex_webrtc/rtp/vp8/depayloader.ex

@@ -38,7 +38,7 @@ defmodule ExWebRTC.RTP.Depayloader.VP8 do
         Resetting depayloader state. Payload: #{inspect(packet.payload)}.\
         """)
 
-        {:ok, %{depayloader | current_frame: nil, current_timestamp: nil}}
+        {nil, %{depayloader | current_frame: nil, current_timestamp: nil}}
     end
   end