RTP output (#51)

* Create rtp output

* Add a working test

* Add Opus and H265 support

Author: Noarkhh

lib/boombox.ex

@@ -24,20 +24,20 @@ defmodule Boombox do
   @typedoc """
   Some encodings can/must be accompanied with encoding specific parameters:
     * AAC:
-      - bitrate_mode - has to be provided. Defines how the RTP stream was payloaded and should be depayloaded.
-      - audio_specific_config - has to be provided. Contains crucial information about the stream and has to be obtained from a side channel.
+      - bitrate_mode - MUST be provided for both RTP input and output. Defines which mode should be assumed/set when depayloading/payloading.
+      - audio_specific_config - MUST be provided for RTP input. Contains crucial information about the stream and has to be obtained from a side channel.
     * H264 and H265:
-      - vpss (H265 only), ppss, spss - optional. Parameter sets, could be obtained from a side channel. They contain information about the encoded stream.
+      - vpss (H265 only), ppss, spss - MAY be provided for RTP input. picture and sequence parameter sets, could be obtained from a side channel. They contain information about the encoded stream.
   """
   @type rtp_encoding_specific_params ::
-          {:AAC, [bitrate_mode: RTP.AAC.Utils.mode(), audio_specific_config: binary()]}
-          | {:H264, [ppss: [binary()], spss: [binary()]]}
-          | {:H265, [vpss: [binary()], ppss: [binary()], spss: [binary()]]}
+          {:AAC, [{:bitrate_mode, RTP.AAC.Utils.mode()} | {:audio_specific_config, binary()}]}
+          | {:H264, [{:ppss, [binary()]} | {:spss, [binary()]}]}
+          | {:H265, [{:vpss, [binary()]} | {:ppss, [binary()]} | {:spss, [binary()]}]}
 
   @typedoc """
   For each media type the following parameters are specified:
-    * encoding - has to be provided, some encodings require additional parameters, see `rtp_encoding_specific_params/0`.
-    * payload_type, clock rate - most likely should be provided, if not, then an unofficial default will be used.
+    * encoding - MUST be provided for both RTP input and output, some encodings require additional parameters, see `rtp_encoding_specific_params/0`.
+    * payload_type, clock rate - MAY be provided for both RTP input and output, if not, then an unofficial default will be used.
   """
   @type rtp_track_config :: [
           {:encoding, RTP.encoding_name() | rtp_encoding_specific_params()}
@@ -51,7 +51,25 @@ defmodule Boombox do
   """
   @type in_rtp_opts :: [
           {:port, :inet.port_number()}
-          | {:track_configs, [audio: rtp_track_config(), video: rtp_track_config()]}
+          | {:track_configs,
+             [
+               {:audio, rtp_track_config()}
+               | {:video, rtp_track_config()}
+             ]}
+        ]
+
+  @typedoc """
+  In order to configure RTP output the destination and media configurations must be provided.
+  At least one media type needs to be configured.
+  """
+  @type out_rtp_opts :: [
+          {:address, :inet.ip_address()}
+          | {:port, :inet.port_number()}
+          | {:track_configs,
+             [
+               {:audio, rtp_track_config()}
+               | {:video, rtp_track_config()}
+             ]}
         ]
 
   @type input ::
@@ -68,6 +86,7 @@ defmodule Boombox do
           | {:mp4, location :: String.t()}
           | {:webrtc, webrtc_signaling()}
           | {:hls, location :: String.t()}
+          | {:rtp, out_rtp_opts()}
           | {:stream, out_stream_opts()}
 
   @typep procs :: %{pipeline: pid(), supervisor: pid()}

lib/boombox/pipeline.ex

@@ -198,6 +198,11 @@ defmodule Boombox.Pipeline do
     {[terminate: :normal], state}
   end
 
+  @impl true
+  def handle_element_end_of_stream(:udp_rtp_sink, :input, _ctx, state) do
+    {[terminate: :normal], state}
+  end
+
   @impl true
   def handle_element_end_of_stream(:elixir_stream_sink, Pad.ref(:input, id), _ctx, state) do
     eos_info = List.delete(state.eos_info, id)
@@ -316,8 +321,8 @@ defmodule Boombox.Pipeline do
     Boombox.RTSP.create_input(uri)
   end
 
-  defp create_input({:rtp, params}, _ctx, _state) do
-    Boombox.RTP.create_input(params)
+  defp create_input({:rtp, opts}, _ctx, _state) do
+    Boombox.RTP.create_input(opts)
   end
 
   defp create_input({:stream, params}, _ctx, state) do
@@ -359,6 +364,10 @@ defmodule Boombox.Pipeline do
     Boombox.HLS.link_output(location, track_builders, spec_builder)
   end
 
+  defp link_output({:rtp, opts}, track_builders, spec_builder, _ctx, _state) do
+    Boombox.RTP.link_output(opts, track_builders, spec_builder)
+  end
+
   defp link_output({:stream, opts}, track_builders, spec_builder, _ctx, state) do
     Boombox.ElixirStream.link_output(state.parent, opts, track_builders, spec_builder)
   end

lib/boombox/rtp.ex

@@ -7,39 +7,72 @@ defmodule Boombox.RTP do
   alias Boombox.Pipeline.Ready
   alias Membrane.RTP
 
-  @required_opts [:port, :track_configs]
+  @required_opts %{
+    input: [:port, :track_configs],
+    output: [:address, :port, :track_configs]
+  }
+
   @required_encoding_specific_params %{
-    AAC: [bitrate_mode: [require?: true], audio_specific_config: [require?: true]],
-    OPUS: [],
-    H264: [ppss: [require?: false], spss: [require?: false]],
-    H265: [ppss: [require?: false], spss: [require?: false]]
+    input: %{
+      AAC: [bitrate_mode: [require?: true], audio_specific_config: [require?: true]],
+      H264: [ppss: [require?: false, default: []], spss: [require?: false, default: []]],
+      H265: [
+        vpss: [require?: false, default: []],
+        ppss: [require?: false, default: []],
+        spss: [require?: false, default: []]
+      ]
+    },
+    output: %{
+      AAC: [bitrate_mode: [require?: true]]
+    }
   }
 
-  @type parsed_encoding_specific_params ::
+  @type parsed_input_encoding_specific_params ::
           %{bitrate_mode: RTP.AAC.Utils.mode(), audio_specific_config: binary()}
-          | %{optional(:ppss) => [binary()], optional(:spss) => [binary()]}
-          | %{
-              optional(:vpss) => [binary()],
-              optional(:ppss) => [binary()],
-              optional(:spss) => [binary()]
-            }
+          | %{:ppss => [binary()], :spss => [binary()]}
+          | %{:vpss => [binary()], :ppss => [binary()], :spss => [binary()]}
+          | %{}
+
+  @type parsed_output_encoding_specific_params ::
+          %{bitrate_mode: RTP.AAC.Utils.mode()}
           | %{}
 
-  @type parsed_track_config :: %{
+  @type parsed_input_track_config :: %{
+          encoding_name: RTP.encoding_name(),
+          encoding_specific_params: parsed_input_encoding_specific_params(),
+          payload_type: RTP.payload_type(),
+          clock_rate: RTP.clock_rate()
+        }
+
+  @type parsed_output_track_config :: %{
           encoding_name: RTP.encoding_name(),
-          encoding_specific_params: parsed_encoding_specific_params(),
+          encoding_specific_params: parsed_output_encoding_specific_params(),
           payload_type: RTP.payload_type(),
           clock_rate: RTP.clock_rate()
         }
 
   @type parsed_in_opts :: %{
           port: :inet.port_number(),
-          track_configs: %{audio: parsed_track_config(), video: parsed_track_config()}
+          track_configs: %{
+            optional(:audio) => parsed_input_track_config(),
+            optional(:video) => parsed_input_track_config()
+          }
         }
 
+  @type parsed_out_opts :: %{
+          address: :inet.ip_address(),
+          port: :inet.port_number(),
+          track_configs: %{
+            optional(:audio) => parsed_output_track_config(),
+            optional(:video) => parsed_output_track_config()
+          }
+        }
+
+  @type parsed_track_config :: parsed_input_track_config() | parsed_output_track_config()
+
   @spec create_input(Boombox.in_rtp_opts()) :: Ready.t()
   def create_input(opts) do
-    parsed_options = validate_and_parse_options(opts)
+    parsed_options = validate_and_parse_options(:input, opts)
     payload_type_mapping = get_payload_type_mapping(parsed_options.track_configs)
 
     spec =
@@ -51,8 +84,8 @@ defmodule Boombox.RTP do
         {depayloader, parser} =
           case track_config.encoding_name do
             :H264 ->
-              ppss = Map.get(track_config.encoding_specific_params, :ppss, [])
-              spss = Map.get(track_config.encoding_specific_params, :spss, [])
+              ppss = track_config.encoding_specific_params.ppss
+              spss = track_config.encoding_specific_params.spss
               {Membrane.RTP.H264.Depayloader, %Membrane.H264.Parser{ppss: ppss, spss: spss}}
 
             :AAC ->
@@ -77,21 +110,86 @@ defmodule Boombox.RTP do
         spec =
           get_child(:rtp_demuxer)
           |> via_out(:output, options: [stream_id: {:encoding_name, track_config.encoding_name}])
-          |> child({:jitter_buffer, track_config.encoding_name}, %Membrane.RTP.JitterBuffer{
+          |> child({:jitter_buffer, media_type}, %Membrane.RTP.JitterBuffer{
             clock_rate: track_config.clock_rate
           })
-          |> child({:rtp_depayloader, track_config.encoding_name}, depayloader)
-          |> child({:rtp_in_parser, track_config.encoding_name}, parser)
+          |> child({:rtp_depayloader, media_type}, depayloader)
+          |> child({:rtp_in_parser, media_type}, parser)
 
         {media_type, spec}
       end)
 
     %Ready{spec_builder: spec, track_builders: track_builders}
   end
 
-  @spec validate_and_parse_options(Boombox.in_rtp_opts()) :: parsed_in_opts()
-  defp validate_and_parse_options(opts) do
-    Enum.each(@required_opts, fn required_option ->
+  @spec link_output(
+          Boombox.out_rtp_opts(),
+          Boombox.Pipeline.track_builders(),
+          Membrane.ChildrenSpec.t()
+        ) :: Ready.t()
+  def link_output(opts, track_builders, spec_builder) do
+    parsed_opts = validate_and_parse_options(:output, opts)
+
+    spec = [
+      spec_builder,
+      child(:rtp_muxer, Membrane.RTP.Muxer)
+      |> child(:udp_rtp_sink, %Membrane.UDP.Sink{
+        destination_address: parsed_opts.address,
+        destination_port_no: parsed_opts.port
+      }),
+      Enum.map(track_builders, fn {media_type, builder} ->
+        track_config = parsed_opts.track_configs[media_type]
+
+        {output_stream_format, parser, payloader} =
+          case track_config.encoding_name do
+            :H264 ->
+              {%Membrane.H264{stream_structure: :annexb, alignment: :nalu},
+               %Membrane.H264.Parser{output_stream_structure: :annexb, output_alignment: :nalu},
+               Membrane.RTP.H264.Payloader}
+
+            :AAC ->
+              {%Membrane.AAC{encapsulation: :none},
+               %Membrane.AAC.Parser{out_encapsulation: :none},
+               %Membrane.RTP.AAC.Payloader{
+                 mode: track_config.encoding_specific_params.bitrate_mode,
+                 frames_per_packet: 1
+               }}
+
+            :OPUS ->
+              {Membrane.Opus, %Membrane.Opus.Parser{delimitation: :undelimit},
+               Membrane.RTP.Opus.Payloader}
+
+            :H265 ->
+              {%Membrane.H265{stream_structure: :annexb, alignment: :nalu},
+               %Membrane.H265.Parser{output_stream_structure: :annexb, output_alignment: :nalu},
+               Membrane.RTP.H265.Payloader}
+          end
+
+        builder
+        |> child({:rtp_transcoder, media_type}, %Boombox.Transcoder{
+          output_stream_format: output_stream_format
+        })
+        |> child({:rtp_out_parser, media_type}, parser)
+        |> child({:rtp_payloader, media_type}, payloader)
+        |> child({:realtimer, media_type}, Membrane.Realtimer)
+        |> via_in(:input,
+          options: [
+            encoding: track_config.encoding_name,
+            payload_type: track_config.payload_type,
+            clock_rate: track_config.clock_rate
+          ]
+        )
+        |> get_child(:rtp_muxer)
+      end)
+    ]
+
+    %Ready{actions: [spec: spec]}
+  end
+
+  @spec validate_and_parse_options(:input, Boombox.in_rtp_opts()) :: parsed_in_opts()
+  @spec validate_and_parse_options(:output, Boombox.out_rtp_opts()) :: parsed_out_opts()
+  defp validate_and_parse_options(direction, opts) do
+    Enum.each(@required_opts[direction], fn required_option ->
       unless Keyword.has_key?(opts, required_option) do
         raise "Required option #{inspect(required_option)} not present in passed RTP options"
       end
@@ -103,16 +201,25 @@ defmodule Boombox.RTP do
 
     parsed_track_configs =
       Map.new(opts[:track_configs], fn {media_type, track_config} ->
-        {media_type, validate_and_parse_track_config!(track_config)}
+        {media_type, validate_and_parse_track_config!(direction, track_config)}
       end)
 
-    %{port: opts[:port], track_configs: parsed_track_configs}
+    case direction do
+      :input ->
+        %{port: opts[:port], track_configs: parsed_track_configs}
+
+      :output ->
+        %{address: opts[:address], port: opts[:port], track_configs: parsed_track_configs}
+    end
   end
 
-  @spec validate_and_parse_track_config!(Boombox.rtp_track_config()) :: parsed_track_config()
-  defp validate_and_parse_track_config!(track_config) do
+  @spec validate_and_parse_track_config!(:input, Boombox.rtp_track_config()) ::
+          parsed_input_track_config()
+  @spec validate_and_parse_track_config!(:output, Boombox.rtp_track_config()) ::
+          parsed_output_track_config()
+  defp validate_and_parse_track_config!(direction, track_config) do
     {encoding_name, encoding_specific_params} =
-      validate_and_parse_encoding!(track_config[:encoding])
+      validate_and_parse_encoding!(direction, track_config[:encoding])
 
     track_config = Keyword.put(track_config, :encoding_name, encoding_name)
 
@@ -135,18 +242,24 @@ defmodule Boombox.RTP do
     }
   end
 
-  @spec validate_and_parse_encoding!(RTP.encoding_name() | Boombox.rtp_encoding_specific_params()) ::
-          {RTP.encoding_name(), %{}} | parsed_encoding_specific_params()
-  defp validate_and_parse_encoding!(encoding) do
+  @spec validate_and_parse_encoding!(
+          :input,
+          RTP.encoding_name() | Boombox.rtp_encoding_specific_params()
+        ) :: {RTP.encoding_name(), parsed_input_encoding_specific_params()}
+  @spec validate_and_parse_encoding!(
+          :output,
+          RTP.encoding_name() | Boombox.rtp_encoding_specific_params()
+        ) :: {RTP.encoding_name(), parsed_output_encoding_specific_params()}
+  defp validate_and_parse_encoding!(direction, encoding) do
     case encoding do
       nil ->
         raise "Encoding name not provided"
 
       encoding when is_atom(encoding) ->
-        validate_and_parse_encoding!({encoding, []})
+        validate_and_parse_encoding!(direction, {encoding, []})
 
       {encoding, encoding_params} when is_atom(encoding) ->
-        field_specs = Map.get(@required_encoding_specific_params, encoding, [])
+        field_specs = Map.get(@required_encoding_specific_params[direction], encoding, [])
         {:ok, encoding_params} = Bunch.Config.parse(encoding_params, field_specs)
         {encoding, encoding_params}
     end