Support for streams with fragmented MP4 segments

[samrat]

Hello,

Thanks for this plugin!

It looks like the plugin currently only supports MPEG-TS segments.

Would it be possible to add support for streams with fMP4 segments?

[dmorn]

Hi @samrat! I don't think there is anything that could stop you from doing it. This plugin does not decode nor parse the renditions you select, it just forwards the contents of the files.

Using the test as an example, when you receive the master playlist from the source notification just select the fMP4 rendition.

The Source then delivers the stream format message, which is indeed going to be incorrect (it will report MPEGTS with the fMP4 codecs), but you can easily put a filter after that and take care of correcting the format before handing the bytestream over to a decoder.

If you don't find a Membrane decoder for your format (check membrane's dash plugin I think that one uses this format), you can "easily" craft one using Exile and ffmpeg, something like this

defmodule Decoder do
  use Membrane.Filter

  defmodule FFmpegError do
    defexception [:message]

    @impl true
    def exception(value) do
      %FFmpegError{message: "FFmpeg error: #{inspect(value)}"}
    end
  end

  def_input_pad(:input,
    accepted_format: Membrane.RemoteStream,
    availability: :always,
    flow_control: :auto
  )

  def_output_pad(:output,
    accepted_format: Membrane.RawAudio,
    availability: :always,
    flow_control: :auto
  )

  @impl true
  def handle_init(_ctx, _opts) do
    {:ok, p} =
      Exile.Process.start_link(
        ~w(ffmpeg -hide_banner -loglevel error -i - -ac 1 -ar 48000 -f s16le -)
      )

    parent = self()

    read_loop_task =
      Task.Supervisor.async(Speech.Core.Task.Supervisor, fn ->
        :ok = Exile.Process.change_pipe_owner(p, :stdout, self())
        read_loop(p, parent)
      end)

    {[], %{ffmpeg: p, read_loop_task: read_loop_task}}
  end

  @impl true
  def handle_stream_format(:input, _input_format, _ctx, state) do
    {[
       stream_format:
         {:output,
          %Membrane.RawAudio{
            channels: 1,
            sample_format: :s16le,
            sample_rate: 48000
          }}
     ], state}
  end

  @impl true
  def handle_buffer(:input, buffer, _ctx, state) do
    :ok = Exile.Process.write(state.ffmpeg, buffer.payload)
    {[], state}
  end

  @impl true
  def handle_end_of_stream(:input, _ctx, state) do
    Exile.Process.close_stdin(state.ffmpeg)
    {[], state}
  end

  @impl true
  def handle_info({:exile, {:data, payload}}, _ctx, state) do
    {[buffer: {:output, %Membrane.Buffer{payload: payload}}], state}
  end

  def handle_info({ref, :eof}, _ctx, state = %{read_loop_task: %Task{ref: ref}}) do
    # Avoid receiving the DOWN message.
    Process.demonitor(ref, [:flush])
    {:ok, 0} = Exile.Process.await_exit(state.ffmpeg)
    {[end_of_stream: :output], state}
  end

  def handle_info({ref, {:error, any}}, _ctx, %{read_loop_task: %Task{ref: ref}}) do
    raise FFmpegError, any
  end

  defp read_loop(p, parent) do
    case Exile.Process.read(p) do
      {:ok, data} ->
        send(parent, {:exile, {:data, data}})
        read_loop(p, parent)

      :eof ->
        :eof

      {:error, any} ->
        {:error, any}
    end
  end
end

This one will turn an audio stream into raw audio, depending on your use case change the ffmpeg command. Afterwards, you'll for sure find a Membrane parser that can produce well-behaving buffers for your pipeline.

Let me know if you manage to do it!