Transcoding

lib/boombox/elixir_stream.ex

@@ -3,7 +3,10 @@ defmodule Boombox.ElixirStream do
 
   import Membrane.ChildrenSpec
   require Membrane.Pad, as: Pad
+
+  alias __MODULE__.{Sink, Source}
   alias Boombox.Pipeline.Ready
+  alias Boombox.Transcoders
 
   @options_audio_keys [:audio_format, :audio_rate, :audio_channels]
 
@@ -53,15 +56,21 @@ defmodule Boombox.ElixirStream do
         Enum.map(track_builders, fn
           {:audio, builder} ->
             builder
+            |> child(:mp4_audio_transcoder, %Transcoders.Audio{
+              output_stream_format_module: Membrane.RawAudio
+            })
             |> then(&maybe_plug_resampler(&1, options))
             |> via_in(Pad.ref(:input, :audio))
             |> get_child(:elixir_stream_sink)
 
           {:video, builder} ->
             builder
-            |> child(%Membrane.H264.Parser{output_stream_structure: :annexb})
-            |> child(Membrane.H264.FFmpeg.Decoder)
-            |> child(%Membrane.FFmpeg.SWScale.Converter{format: :RGB})
+            |> child(:elixir_stream_video_transcoder, %Transcoders.Video{
+              output_stream_format: Membrane.RawVideo
+            })
+            |> child(:elixir_stream_rgb_converter, %Membrane.FFmpeg.SWScale.Converter{
+              format: :RGB
+            })
             |> via_in(Pad.ref(:input, :video))
             |> get_child(:elixir_stream_sink)
         end),

lib/boombox/elixir_stream/sink.ex

@@ -1,4 +1,4 @@
-defmodule Sink do
+defmodule Boombox.ElixirStream.Sink do
   @moduledoc false
   use Membrane.Sink
 

lib/boombox/elixir_stream/source.ex

@@ -1,4 +1,4 @@
-defmodule Source do
+defmodule Boombox.ElixirStream.Source do
   @moduledoc false
   use Membrane.Source
 

lib/boombox/hls.ex

@@ -5,6 +5,8 @@ defmodule Boombox.HLS do
 
   require Membrane.Pad, as: Pad
   alias Boombox.Pipeline.Ready
+  alias Boombox.Transcoders
+  alias Membrane.H264
   alias Membrane.Time
 
   @spec link_output(
@@ -42,14 +44,19 @@ defmodule Boombox.HLS do
         Enum.map(track_builders, fn
           {:audio, builder} ->
             builder
-            |> child(:hls_out_aac_encoder, Membrane.AAC.FDK.Encoder)
+            |> child(:mp4_audio_transcoder, %Transcoders.Audio{
+              output_stream_format_module: Membrane.AAC
+            })
             |> via_in(Pad.ref(:input, :audio),
               options: [encoding: :AAC, segment_duration: Time.milliseconds(2000)]
             )
             |> get_child(:hls_sink_bin)
 
           {:video, builder} ->
             builder
+            |> child(:hls_video_transcoder, %Transcoders.Video{
+              output_stream_format: %H264{alignment: :au, stream_structure: :avc3}
+            })
             |> via_in(Pad.ref(:input, :video),
               options: [encoding: :H264, segment_duration: Time.milliseconds(2000)]
             )

lib/boombox/mp4.ex

@@ -4,6 +4,11 @@ defmodule Boombox.MP4 do
   import Membrane.ChildrenSpec
   require Membrane.Pad, as: Pad
   alias Boombox.Pipeline.{Ready, Wait}
+  alias Boombox.Transcoders
+  alias Membrane.H264
+  alias Membrane.H265
+
+  defguardp is_h264_or_h265(format) when is_struct(format) and format.__struct__ in [H264, H265]
 
   @spec create_input(String.t(), transport: :file | :http) :: Wait.t()
   def create_input(location, opts) do
@@ -33,11 +38,10 @@ defmodule Boombox.MP4 do
             get_child(:mp4_demuxer)
             |> via_out(Pad.ref(:output, id))
             |> child(:mp4_in_aac_parser, Membrane.AAC.Parser)
-            |> child(:mp4_in_aac_decoder, Membrane.AAC.FDK.Decoder)
 
           {:audio, spec}
 
-        {id, %Membrane.H264{}} ->
+        {id, video_format} when is_h264_or_h265(video_format) ->
           spec =
             get_child(:mp4_demuxer)
             |> via_out(Pad.ref(:output, id))
@@ -62,7 +66,9 @@ defmodule Boombox.MP4 do
         Enum.map(track_builders, fn
           {:audio, builder} ->
             builder
-            |> child(:mp4_out_aac_encoder, Membrane.AAC.FDK.Encoder)
+            |> child(:mp4_audio_transcoder, %Transcoders.Audio{
+              output_stream_format_module: Membrane.AAC
+            })
             |> child(:mp4_out_aac_parser, %Membrane.AAC.Parser{
               out_encapsulation: :none,
               output_config: :esds
@@ -72,7 +78,18 @@ defmodule Boombox.MP4 do
 
           {:video, builder} ->
             builder
-            |> child(:mp4_out_h264_parser, %Membrane.H264.Parser{output_stream_structure: :avc3})
+            |> child(:mp4_video_transcoder, %Transcoders.Video{
+              output_stream_format: fn
+                %H264{stream_structure: :annexb} = h264 ->
+                  %{h264 | stream_structure: :avc3, alignment: :au}
+
+                %H264{} = h264 ->
+                  %{h264 | alignment: :au}
+
+                _not_h264 ->
+                  %H264{stream_structure: :avc3, alignment: :au}
+              end
+            })
             |> via_in(Pad.ref(:input, :video))
             |> get_child(:mp4_muxer)
         end)

lib/boombox/rtmp.ex

@@ -47,10 +47,9 @@ defmodule Boombox.RTMP do
       child(:rtmp_source, %RTMP.SourceBin{client_ref: client_ref})
       |> via_out(:audio)
       |> child(:rtmp_in_aac_parser, Membrane.AAC.Parser)
-      |> child(:rtmp_in_aac_decoder, Membrane.AAC.FDK.Decoder)
 
     track_builders = %{
-      audio: get_child(:rtmp_in_aac_decoder),
+      audio: get_child(:rtmp_in_aac_parser),
       video: get_child(:rtmp_source) |> via_out(:video)
     }
 

lib/boombox/rtsp.ex

@@ -59,7 +59,6 @@ defmodule Boombox.RTSP do
           spec =
             get_child(:rtsp_source)
             |> via_out(Membrane.Pad.ref(:output, ssrc))
-            |> child(Membrane.Debug.Sink)
 
           {spec, track_builders}
 
@@ -82,7 +81,6 @@ defmodule Boombox.RTSP do
             get_child(:rtsp_source)
             |> via_out(Membrane.Pad.ref(:output, ssrc))
             |> child(:rtsp_in_aac_parser, Membrane.AAC.Parser)
-            |> child(:rtsp_in_aac_decoder, Membrane.AAC.FDK.Decoder)
 
           {[], Map.put(track_builders, :audio, audio_spec)}
 

lib/boombox/transcoders/audio.ex

@@ -0,0 +1,139 @@
+defmodule Boombox.Transcoders.Audio do
+  @moduledoc false
+  use Membrane.Bin
+
+  require Membrane.Logger
+
+  alias Boombox.Transcoders.Helpers.ForwardingFilter
+  alias Membrane.{AAC, Funnel, Opus, RawAudio, RemoteStream}
+
+  @type stream_format :: AAC.t() | Opus.t() | RemoteStream.t() | RawAudio.t()
+  @type stream_format_module :: AAC | Opus | RawAudio
+
+  @opus_sample_rate 48_000
+
+  def_input_pad :input,
+    accepted_format:
+      any_of(AAC, Opus, %RemoteStream{content_format: Opus, type: :packetized}, RawAudio)
+
+  def_output_pad :output, accepted_format: any_of(AAC, Opus, RawAudio)
+
+  def_options input_stream_format: [
+                spec: stream_format(),
+                default: nil
+              ],
+              output_stream_format_module: [
+                spec: stream_format_module()
+              ]
+
+  @impl true
+  def handle_init(_ctx, opts) do
+    spec = [
+      bin_input()
+      |> child(:forwarding_filter, ForwardingFilter),
+      child(:output_funnel, Funnel)
+      |> bin_output()
+    ]
+
+    state =
+      Map.from_struct(opts)
+      |> Map.put(:input_linked_with_output?, false)
+
+    {link_actions, state} = maybe_link_input_with_output(state)
+    {[spec: spec] ++ link_actions, state}
+  end
+
+  @impl true
+  def handle_child_notification({:stream_format, stream_format}, :forwarding_filter, _ctx, state) do
+    %{state | input_stream_format: stream_format}
+    |> maybe_link_input_with_output()
+  end
+
+  @impl true
+  def handle_child_notification(_notification, _element, _ctx, state) do
+    {[], state}
+  end
+
+  defp maybe_link_input_with_output(state)
+       when state.input_linked_with_output? or state.input_stream_format == nil do
+    {[], state}
+  end
+
+  defp maybe_link_input_with_output(state) do
+    spec =
+      link_input_with_output(
+        state.input_stream_format,
+        state.output_stream_format_module
+      )
+
+    state = %{state | input_linked_with_output?: true}
+    {[spec: spec], state}
+  end
+
+  defp link_input_with_output(%format_module{}, format_module) do
+    Membrane.Logger.debug("""
+    This bin will only forward buffers, as the input stream format is the same as the output stream format.
+    """)
+
+    get_child(:forwarding_filter)
+    |> get_child(:output_funnel)
+  end
+
+  defp link_input_with_output(%RemoteStream{content_format: Opus}, Opus) do
+    get_child(:forwarding_filter)
+    |> child(:opus_parser, Opus.Parser)
+    |> get_child(:output_funnel)
+  end
+
+  defp link_input_with_output(input_format, output_format_module) do
+    get_child(:forwarding_filter)
+    |> maybe_plug_decoder(input_format)
+    |> maybe_plug_resampler(input_format, output_format_module)
+    |> maybe_plug_encoder(output_format_module)
+    |> get_child(:output_funnel)
+  end
+
+  defp maybe_plug_decoder(builder, %Opus{}) do
+    builder |> child(:opus_decoder, Opus.Decoder)
+  end
+
+  defp maybe_plug_decoder(builder, %RemoteStream{content_format: Opus, type: :packetized}) do
+    builder |> child(:opus_decoder, Opus.Decoder)
+  end
+
+  defp maybe_plug_decoder(builder, %AAC{}) do
+    builder |> child(:aac_decoder, AAC.FDK.Decoder)
+  end
+
+  defp maybe_plug_decoder(builder, %RawAudio{}) do
+    builder
+  end
+
+  defp maybe_plug_resampler(builder, %{sample_rate: sample_rate} = input_format, Opus)
+       when sample_rate != @opus_sample_rate do
+    builder
+    |> child(:resampler, %Membrane.FFmpeg.SWResample.Converter{
+      output_stream_format: %RawAudio{
+        sample_format: :s16le,
+        sample_rate: @opus_sample_rate,
+        channels: input_format.channels
+      }
+    })
+  end
+
+  defp maybe_plug_resampler(builder, _input_format, _output_format_module) do
+    builder
+  end
+
+  defp maybe_plug_encoder(builder, Opus) do
+    builder |> child(:opus_encoder, Opus.Encoder)
+  end
+
+  defp maybe_plug_encoder(builder, AAC) do
+    builder |> child(:aac_encoder, AAC.FDK.Encoder)
+  end
+
+  defp maybe_plug_encoder(builder, RawAudio) do
+    builder
+  end
+end