Rename RTP.TCP.Depayloader to RTP.RTSP.Decapsulator

README.md

@@ -15,7 +15,7 @@ The package can be installed by adding `membrane_rtp_plugin` to your list of dep
 ```elixir
 def deps do
   [
-    {:membrane_rtp_plugin, "~> 0.25.0"},
+    {:membrane_rtp_plugin, "~> 0.26.0"},
     {:ex_libsrtp, ">= 0.0.0"} # required only if SRTP/SRTCP support is needed
   ]
 end

lib/membrane/rtp/tcp_depayloader.ex → lib/membrane/rtp/rtsp_decapsulator.ex

@@ -1,12 +1,13 @@
-defmodule Membrane.RTP.TCP.Depayloader do
+defmodule Membrane.RTP.RTSP.Decapsulator do
   @moduledoc """
-  This element provides functionality of depayloading RTP Packets received by TCP and redirecting
-  RTSP messages received in the same stream. The encapsulation is described in RFC 7826 Section 14.
+  This element provides functionality of decapsulating RTP Packets and redirecting RTSP messages
+  received in the same TCP stream established with RTSP. The encapsulation is described in
+  RFC 7826 Section 14.
 
-  Encapsulated packets interleaved in the stream will have the following structure:
+  Encapsulated RTP packets interleaved in the stream will have the following structure:
   ["$" = 36 :: 1 byte][Channel id :: 1 byte][Length :: 2 bytes][packet :: <Length> bytes]
 
-  RTSP Messages
+  RTSP Messages are not encapsulated this way, but can only be present between RTP packets.
   """
   use Membrane.Filter
 
@@ -95,23 +96,21 @@ defmodule Membrane.RTP.TCP.Depayloader do
   end
 
   defp get_complete_packets(
-         <<"$", _rest::binary>> = packets_binary,
+         <<"$", received_channel_id, payload_length::size(16), rest::binary>> = packets_binary,
          channel_id,
          complete_packets
        ) do
-    <<"$", received_channel_id, payload_length::size(16), rest::binary>> = packets_binary
+    case rest do
+      <<complete_packet_binary::binary-size(payload_length)-unit(8), rest::binary>> ->
+        complete_packets =
+          if channel_id != received_channel_id,
+            do: complete_packets,
+            else: [complete_packet_binary | complete_packets]
 
-    if payload_length > byte_size(rest) do
-      {packets_binary, Enum.reverse(complete_packets)}
-    else
-      <<complete_packet_binary::binary-size(payload_length)-unit(8), rest::binary>> = rest
+        get_complete_packets(rest, channel_id, complete_packets)
 
-      complete_packets =
-        if channel_id != received_channel_id,
-          do: complete_packets,
-          else: [complete_packet_binary | complete_packets]
-
-      get_complete_packets(rest, channel_id, complete_packets)
+      _incomplete_packet_binary ->
+        {packets_binary, Enum.reverse(complete_packets)}
     end
   end
 

mix.exs

@@ -1,7 +1,7 @@
 defmodule Membrane.RTP.Plugin.MixProject do
   use Mix.Project
 
-  @version "0.25.0"
+  @version "0.26.0"
   @github_url "https://github.com/membraneframework/membrane_rtp_plugin"
 
   def project do

test/membrane/rtp/rtsp_decapsulator_test.exs

@@ -0,0 +1,84 @@
+defmodule Membrane.RTP.RTSPDecapsulatorTest do
+  use ExUnit.Case
+
+  import Membrane.Testing.Assertions
+  import Membrane.ChildrenSpec
+
+  alias Membrane.RTP.RTSP.Decapsulator
+  alias Membrane.Testing.{Pipeline, Sink, Source}
+
+  @header_length 4
+
+  defp encapsulate_rtp_packets(rtp_packets) do
+    Enum.map(rtp_packets, &<<"$", 0, byte_size(&1)::size(16), &1::binary>>)
+  end
+
+  defp create_tcp_segments(encapsulated_rtp_packets, tcp_segments_lengths) do
+    assert Enum.sum(tcp_segments_lengths) ==
+             Enum.sum(Enum.map(encapsulated_rtp_packets, &byte_size(&1)))
+
+    encaplsulated_rtp_packets_binary = Enum.join(encapsulated_rtp_packets)
+
+    {tcp_segments, _length} =
+      Enum.map_reduce(tcp_segments_lengths, 0, fn len, pos ->
+        {:binary.part(encaplsulated_rtp_packets_binary, pos, len), pos + len}
+      end)
+
+    tcp_segments
+  end
+
+  defp perform_standard_test(rtp_packets_lengths, tcp_segments_lengths) do
+    rtp_packets = Enum.map(rtp_packets_lengths, &<<0::size(&1)-unit(8)>>)
+
+    tcp_segments =
+      rtp_packets |> encapsulate_rtp_packets() |> create_tcp_segments(tcp_segments_lengths)
+
+    pipeline =
+      Pipeline.start_link_supervised!(
+        spec:
+          child(:source, %Source{
+            output: tcp_segments
+          })
+          |> child(:decapsulator, Decapsulator)
+          |> child(:sink, Sink)
+      )
+
+    Enum.each(rtp_packets, fn packet ->
+      assert_sink_buffer(pipeline, :sink, %Membrane.Buffer{payload: ^packet})
+    end)
+
+    Pipeline.terminate(pipeline)
+  end
+
+  describe "RTSP Decapsulator decapsulates correctly" do
+    test "when one tcp segment is one rtp packet" do
+      rtp_packets_lengths = 10..20
+      tcp_segments_lengths = Enum.map(rtp_packets_lengths, &(&1 + @header_length))
+
+      perform_standard_test(rtp_packets_lengths, tcp_segments_lengths)
+    end
+
+    test "when there are multiple (3) rtp packets in one tcp segment" do
+      rtp_packets_lengths = 10..40
+
+      tcp_segments_lengths =
+        rtp_packets_lengths
+        |> Enum.chunk_every(3)
+        |> Enum.map(&(Enum.sum(&1) + length(&1) * @header_length))
+
+      perform_standard_test(rtp_packets_lengths, tcp_segments_lengths)
+    end
+
+    test "when rtp packets are spread across multiple (3) tcp segments" do
+      rtp_packets_lengths = 11..41//3
+
+      tcp_segments_lengths =
+        Enum.flat_map(rtp_packets_lengths, fn len ->
+          tcp_segment_base_length = div(len + @header_length, 3)
+          [tcp_segment_base_length - 1, tcp_segment_base_length, tcp_segment_base_length + 1]
+        end)
+
+      perform_standard_test(rtp_packets_lengths, tcp_segments_lengths)
+    end
+  end
+end