Hi there, unfortunately, due to how the framework is designed, it’s not straightforward. The best way IMO is to use the handle_element_end_of_stream callback to know when the stream finishes and restart the pipeline or a part of it. If you want to restart only a part of the pipeline, you need a tee between the part you want to restart and the rest. The stream needs to be at least demuxed (extracted from a container) at that point.

Another approach is to use a looping filter that will buffer the entire stream, as described here. The problem with this approach is that it, well, buffers the entire stream

Hello.

I got the idea from the handle_element_end_of_stream callback method you suggested and the link you shared.

Instead of removing the pipeline to stop the media streaming with RTMP later,

I chose to use a custom Membrane.Filter to loop the file again.

Here is a simple example I made:

defmodule MyApp.Application do
  use Application

  @impl true
  def start(_type, _args) do
    children = [
      MyApp.Pipeline
    ]

    opts = [strategy: :one_for_one, name: MyApp.Supervisor]
    Supervisor.start_link(children, opts)
  end
end

defmodule MyApp.Pipeline do
  use Membrane.Pipeline

  @location :code.priv_dir(:my_app) |> Path.join("bgm_24000hz_1ch_s16le.wav")
  @stream_format %Membrane.RawAudio{sample_rate: 24_000, channels: 1, sample_format: :s16le}

  def start_link(_opts) do
    Membrane.Pipeline.start_link(__MODULE__, [], name: __MODULE__)
  end

  @impl true
  def handle_init(_ctx, _opts) do
    spec = [
      child(:source, %Membrane.File.Source{location: @location, seekable?: true})
      |> child(:loop_filter, MyApp.LoopFilter)
      |> child(:raw_parser, %Membrane.RawAudioParser{
        stream_format: @stream_format,
        overwrite_pts?: true
      })
      |> child(:sink, Membrane.PortAudio.Sink)
    ]

    {[spec: spec], %{}}
  end
end

defmodule MyApp.LoopFilter do
  use Membrane.Filter

  def_input_pad(:input, accepted_format: _any)
  def_output_pad(:output, accepted_format: _any)

  @impl true
  def handle_init(_ctx, _opts) do
    {[], %{}}
  end

  @impl true
  def handle_playing(_ctx, state) do
    {[event: {:input, %Membrane.File.SeekSourceEvent{start: :bof, size_to_read: :infinity}}],
     state}
  end

  @impl true
  def handle_event(:input, %Membrane.File.EndOfSeekEvent{}, _ctx, state) do
    {[event: {:input, %Membrane.File.SeekSourceEvent{start: :bof, size_to_read: :infinity}}],
     state}
  end

  @impl true
  def handle_event(pad, event, ctx, state) do
    super(pad, event, ctx, state)
  end

  @impl true
  def handle_buffer(:input, buffer, _ctx, state) do
    {[buffer: {:output, buffer}], state}
  end
end

There’s no problem with Membrane.PortAudio.Sink, but later, when I add an audio mixer and send it via RTMP, additional pipelines may be created.

I’ll ask if I have more questions later.

It was very helpful, Thank you!