Initial implementation

lib/transcoder.ex

@@ -0,0 +1,99 @@
+defmodule Membrane.Transcoder do
+  @moduledoc false
+  use Membrane.Bin
+
+  require __MODULE__.Audio
+  require __MODULE__.Video
+
+  alias __MODULE__.{Audio, ForwardingFilter, Video}
+  alias Membrane.{AAC, Funnel, H264, H265, Opus, RawAudio, RawVideo, RemoteStream, VP8}
+
+  @type stream_format ::
+          H264.t()
+          | H265.t()
+          | VP8.t()
+          | RawVideo.t()
+          | AAC.t()
+          | Opus.t()
+          | RemoteStream.t()
+          | RawAudio.t()
+
+  @type stream_format_module :: H264 | H265 | VP8 | RawVideo | AAC | Opus | RawAudio
+
+  @type stream_format_resolver :: (stream_format() -> stream_format() | stream_format_module())
+
+  def_input_pad :input,
+    accepted_format: format when Audio.is_audio_format(format) or Video.is_video_format(format)
+
+  def_output_pad :output,
+    accepted_format: format when Audio.is_audio_format(format) or Video.is_video_format(format)
+
+  def_options output_stream_format: [
+                spec: stream_format() | stream_format_module() | stream_format_resolver()
+              ]
+
+  @impl true
+  def handle_init(_ctx, opts) do
+    spec = [
+      bin_input()
+      |> child(:forwarding_filter, ForwardingFilter),
+      child(:output_funnel, Funnel)
+      |> bin_output()
+    ]
+
+    state =
+      opts
+      |> Map.from_struct()
+      |> Map.put(:input_stream_format, nil)
+
+    {[spec: spec], state}
+  end
+
+  @impl true
+  def handle_child_notification(
+        {:stream_format, format},
+        :forwarding_filter,
+        _ctx,
+        %{input_stream_format: nil} = state
+      ) do
+    state =
+      %{state | input_stream_format: format}
+      |> resolve_output_stream_format()
+
+    spec =
+      get_child(:forwarding_filter)
+      |> plug_transcoding(format, state.output_stream_format)
+      |> get_child(:output_funnel)
+
+    {[spec: spec], state}
+  end
+
+  @impl true
+  def handle_child_notification(_notification, _element, _ctx, state) do
+    {[], state}
+  end
+
+  defp resolve_output_stream_format(state) do
+    case state.output_stream_format do
+      format when is_struct(format) ->
+        state
+
+      module when is_atom(module) ->
+        %{state | output_stream_format: struct(module)}
+
+      resolver when is_function(resolver) ->
+        %{state | output_stream_format: resolver.(state.input_stream_format)}
+        |> resolve_output_stream_format()
+    end
+  end
+
+  defp plug_transcoding(builder, input_format, output_format)
+       when Audio.is_audio_format(input_format) do
+    builder |> Audio.plug_audio_transcoding(input_format, output_format)
+  end
+
+  defp plug_transcoding(builder, input_format, output_format)
+       when Video.is_video_format(input_format) do
+    builder |> Video.plug_video_transcoding(input_format, output_format)
+  end
+end

lib/transcoder/audio.ex

@@ -0,0 +1,120 @@
+defmodule Membrane.Transcoder.Audio do
+  @moduledoc false
+
+  import Membrane.ChildrenSpec
+  require Membrane.Logger
+  alias Membrane.{AAC, ChildrenSpec, Opus, RawAudio, RemoteStream}
+
+  @opus_sample_rate 48_000
+  @aac_sample_rates [
+    96_000,
+    88_200,
+    64_000,
+    48_000,
+    44_100,
+    32_000,
+    24_000,
+    22_050,
+    16_000,
+    12_000,
+    11_025,
+    8000
+  ]
+
+  @type audio_stream_format :: AAC.t() | Opus.t() | RawAudio.t()
+
+  defguard is_audio_format(format)
+           when is_struct(format) and
+                  (format.__struct__ in [AAC, Opus, RawAudio] or
+                     (format.__struct__ == RemoteStream and format.content_format in [Opus, AAC] and
+                        format.type == :packetized))
+
+  @spec plug_audio_transcoding(
+          ChildrenSpec.builder(),
+          audio_stream_format() | RemoteStream.t(),
+          audio_stream_format()
+        ) :: ChildrenSpec.builder()
+  def plug_audio_transcoding(builder, input_format, output_format)
+      when is_audio_format(input_format) and is_audio_format(output_format) do
+    do_plug_audio_transcoding(builder, input_format, output_format)
+  end
+
+  defp do_plug_audio_transcoding(builder, %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.
+    """)
+
+    builder
+  end
+
+  defp do_plug_audio_transcoding(builder, %RemoteStream{content_format: Opus}, %Opus{}) do
+    builder
+  end
+
+  defp do_plug_audio_transcoding(builder, %RemoteStream{content_format: AAC}, %AAC{}) do
+    builder
+  end
+
+  defp do_plug_audio_transcoding(builder, input_format, output_format) do
+    builder
+    |> maybe_plug_decoder(input_format)
+    |> maybe_plug_resampler(input_format, output_format)
+    |> maybe_plug_encoder(output_format)
+  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, %{sample_rate: sample_rate} = input_format, %AAC{})
+       when sample_rate not in @aac_sample_rates do
+    builder
+    |> child(:resampler, %Membrane.FFmpeg.SWResample.Converter{
+      output_stream_format: %RawAudio{
+        sample_format: :s16le,
+        sample_rate: 44_100,
+        channels: input_format.channels
+      }
+    })
+  end
+
+  defp maybe_plug_resampler(builder, _input_format, _output_format) 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

lib/transcoder/forwarding_filter.ex

@@ -0,0 +1,117 @@
+defmodule Membrane.Transcoder.ForwardingFilter do
+  @moduledoc false
+  use Membrane.Filter
+
+  alias Membrane.TimestampQueue
+
+  def_input_pad :input,
+    accepted_format: _any,
+    availability: :on_request
+
+  def_output_pad :output,
+    accepted_format: _any,
+    availability: :on_request
+
+  defguardp is_input_linked(state) when state.input_pad_ref != nil
+  defguardp is_output_linked(state) when state.output_pad_ref != nil
+
+  @impl true
+  def handle_init(_ctx, _opts) do
+    state = %{queue: TimestampQueue.new(), output_pad_ref: nil, input_pad_ref: nil}
+    {[], state}
+  end
+
+  @impl true
+  def handle_playing(ctx, state), do: maybe_flush_queue(ctx, state)
+
+  @impl true
+  def handle_pad_added(Pad.ref(direction, _id) = pad_ref, ctx, state) do
+    same_direction_pads_number =
+      ctx.pads
+      |> Enum.count(fn {_pad_ref, pad_data} -> pad_data.direction == direction end)
+
+    if same_direction_pads_number > 1 do
+      raise """
+      #{inspect(__MODULE__)} can have only one #{inspect(direction)} pad, but it has \
+      #{same_direction_pads_number}
+      """
+    end
+
+    state =
+      case direction do
+        :input -> %{state | input_pad_ref: pad_ref}
+        :output -> %{state | output_pad_ref: pad_ref}
+      end
+
+    maybe_flush_queue(ctx, state)
+  end
+
+  @impl true
+  def handle_stream_format(_input_pad_ref, stream_format, _ctx, state)
+      when is_output_linked(state) do
+    {[
+       stream_format: {state.output_pad_ref, stream_format},
+       notify_parent: {:stream_format, stream_format}
+     ], state}
+  end
+
+  @impl true
+  def handle_stream_format(input_pad_ref, stream_format, _ctx, state) do
+    queue = TimestampQueue.push_stream_format(state.queue, input_pad_ref, stream_format)
+    {[notify_parent: {:stream_format, stream_format}], %{state | queue: queue}}
+  end
+
+  @impl true
+  def handle_buffer(_input_pad_ref, buffer, _ctx, state) when is_output_linked(state) do
+    {[buffer: {state.output_pad_ref, buffer}], state}
+  end
+
+  @impl true
+  def handle_buffer(input_pad_ref, buffer, _ctx, state) do
+    {_suggested_actions, queue} = TimestampQueue.push_buffer(state.queue, input_pad_ref, buffer)
+    {[], %{state | queue: queue}}
+  end
+
+  @impl true
+  def handle_event(Pad.ref(:input, _id), event, _ctx, state) when is_output_linked(state) do
+    {[forward: event], state}
+  end
+
+  @impl true
+  def handle_event(Pad.ref(:output, _id), event, _ctx, state) when is_input_linked(state) do
+    {[forward: event], state}
+  end
+
+  @impl true
+  def handle_event(pad_ref, event, _ctx, state) do
+    queue = TimestampQueue.push_event(state.queue, pad_ref, event)
+    {[], %{state | queue: queue}}
+  end
+
+  @impl true
+  def handle_end_of_stream(_input_pad_ref, _ctx, state) when is_output_linked(state) do
+    {[end_of_stream: state.output_pad_ref], state}
+  end
+
+  @impl true
+  def handle_end_of_stream(input_pad_ref, _ctx, state) do
+    queue = TimestampQueue.push_end_of_stream(state.queue, input_pad_ref)
+    {[], %{state | queue: queue}}
+  end
+
+  defp maybe_flush_queue(ctx, state)
+       when ctx.playback == :playing and is_input_linked(state) and is_output_linked(state) do
+    {_suggested_actions, items, queue} = TimestampQueue.flush_and_close(state.queue)
+
+    actions =
+      Enum.map(items, fn
+        {Pad.ref(:input, _id), {item_type, item}} -> {item_type, {state.output_pad_ref, item}}
+        {Pad.ref(:input, _id), :end_of_stream} -> {:end_of_stream, state.output_pad_ref}
+        {Pad.ref(:output, _id), {:event, item}} -> {:event, {state.input_pad_ref, item}}
+      end)
+
+    {actions, %{state | queue: queue}}
+  end
+
+  defp maybe_flush_queue(_ctx, state), do: {[], state}
+end