Fix the oban example

Author: varsill

guides/useful_concepts/running_membrane_in_elixir_application.md

@@ -2,8 +2,6 @@
 
 This guide outlines best practices for integrating Membrane Pipelines into your Elixir application, specifically focusing on how to attach pipelines to your application's supervision tree.
 
-
-
 In most cases, pipelines are used in one of the following scenarios:
 
 - Static Orchestration: Maintaining a single, permanent pipeline that always runs with your application (e.g. mixing output from multiple IP cameras into a single HLS stream).
@@ -17,7 +15,7 @@ In most cases, pipelines are used in one of the following scenarios:
 This approach is ideal when your pipeline architecture is fixed and needs to run continuously from the moment your application starts.
 Imagine an application that monitors a security camera feed to detect people or vehicles in real-time. Such an application could consist of two components:
 
-- The Membrane Pipeline which connects to an RTSP stream, decodes the video, and extracts raw frames. It sends these frames to the external process (via a Unix socket or standard input), receives the transformed video back, re-encodes it, and broadcasts the final stream using HLS.
+- The Membrane Pipeline which connects to an SRT stream, decodes the video, and extracts raw frames. It sends these frames to the external process (via a Unix socket or standard input), receives the transformed video back, re-encodes it, and broadcasts the final stream using HLS.
 
 - The OS Process being a Python script running a machine learning model like [RF-DETR](https://github.com/roboflow/rf-detr). It reads raw video frames and performs object segmentation, coloring the pixels that correspond to detected objects.
 
@@ -28,21 +26,7 @@ defmodule MyProject.Pipeline do
   use Membrane.Pipeline
 
   @impl true
-  def handle_init(_ctx, rtsp_url) do
-    spec = [
-      child(:source, %Membrane.RTSP.Source{
-        transport: :tcp,
-        allowed_media_types: [:video],
-        stream_uri: rtsp_url,
-        on_connection_closed: :send_eos
-      })
-    ]
-
-    {[spec: spec], %{}}
-  end
-
-  @impl true
-  def handle_child_pad_added(:source, :output, _ctx, state) do
+  def handle_init(_ctx, srt_url) do
     hls_config = %Membrane.HTTPAdaptiveStream.SinkBin{
       manifest_module: Membrane.HTTPAdaptiveStream.HLS,
       target_window_duration: Membrane.Time.seconds(120),
@@ -52,7 +36,12 @@ defmodule MyProject.Pipeline do
     }
 
     spec = [
-      get_child(:source)
+      child(:source, %Membrane.SRT.Source{
+        transport: :tcp,
+        allowed_media_types: [:video],
+        stream_uri: srt_url,
+        on_connection_closed: :send_eos
+      })
       |> child(:depayloader, Membrane.H264.RTP.Depayloader)
       |> child(:parser, Membrane.H264.Parser)
       |> child(:decoder, Membrane.H264.FFmpeg.Decoder)
@@ -62,11 +51,8 @@ defmodule MyProject.Pipeline do
       |> child(:hls, hls_config)
     ]
 
-    {[spec: structure], state}
+    {[spec: spec], %{}}
   end
-
-  @impl true
-  def handle_child_pad_added(_child, _pad, _ctx, state), do: {:ok, state}
 end
 ```
 
@@ -82,10 +68,10 @@ defmodule MyProject.InfrastructureSupervisor do
   end
 
   @impl true
-  def init(rtsp_url) do
+  def init(srt_url) do
     children = [
       {MuonTrap.Daemon, ["python", ["run_model.py", "rf-detr-large-2026.pth"], [log_output: :debug]], restart: :transient]}
-      {MyProject.Pipeline, [input_url: rtsp_url], restart: :transient}
+      {MyProject.Pipeline, [input_url: srt_url], restart: :transient}
     ]
 
     Supervisor.init(children, strategy: :one_for_all)
@@ -106,7 +92,7 @@ defmodule MyProject.Application do
   @impl true
   def start(_type, _args) do
     children = [
-      {MyProject.InfrastructureSupervisor, [rtsp_url: "rtsp://user:password@127.0.0.1:554"]}
+      {MyProject.InfrastructureSupervisor, [srt_url: "srt://127.0.0.1:9710"]}
       # ... other children required by your application
     ]
 
@@ -119,7 +105,7 @@ end
 ## Dynamically spawning the pipelines under the DynamicSupervisor
 
 While the static approach works perfectly for fixed infrastructure, many applications require more flexibility.
-Consider a scenario where you need to spawn a new pipeline on demand to ingest an RTSP stream from a camera provided by a user.
+Consider a scenario where you need to spawn a new pipeline on demand to ingest an SRT stream from a camera provided by a user.
 
 Since we have already defined the `InfrastructureSupervisor`, scaling to multiple dynamic pipelines is straightforward.
 
@@ -151,7 +137,7 @@ def mount(params, _session, socket) do
   if connected?(socket) do
     {:ok, infrastructure_supervisor_pid} = DynamicSupervisor.start_child(
       MyProject.PipelineDynamicSupervisor,
-      {MyProject.InfrastructureSupervisor, params["rtsp_url"]}
+      {MyProject.InfrastructureSupervisor, params["srt_url"]}
     )
     {:ok, assign(socket, :infrastructure_supervisor_pid, infrastructure_supervisor_pid)}
   else
@@ -177,73 +163,86 @@ end
 
 Batch processing is ideal for "offline tasks" where you need to ensure the job completes successfully.
 
-For example, let's assume we want to transcode video from H.264 (in MP4) to VP8 (in Matroska). To achieve this, we build the pipeline as follows:
+For example, let's assume we want to rescale H.264 video in MP4 container. To achieve this, we build the pipeline as follows:
 
 ```elixir
 defmodule TranscodePipeline do
   use Membrane.Pipeline
 
   alias Membrane.File.{Source, Sink}
+  alias Membrane.H264.FFmpeg.{Decoder, Encoder}
   alias Membrane.H264.Parser
-  alias Membrane.FFmpeg.SWScale.PixelFormatConverter
+  alias Membrane.FFmpeg.SWScale.Converter
 
   @impl true
   def handle_init(_ctx, opts) do
-    structure = [
+    spec = [
       child(:source, %Source{location: opts[:input_path]})
       |> child(:demuxer, Membrane.MP4.Demuxer.ISOM)
-      |> via_out(:video)
-      |> child(:parser, Parser)
-      |> child(:decoder, Membrane.H264.FFmpeg.Decoder)
-      |> child(:converter, %PixelFormatConverter{format: :I420})
-      |> child(:encoder, Membrane.VP8.Encoder)
-      |> child(:muxer, Membrane.Matroska.Muxer)
+      |> via_out(:output, options: [kind: :video])
+      |> child(:in_parser, %Parser{output_stream_structure: :annexb})
+      |> child(:decoder, Decoder)
+      |> child(:converter, %Converter{output_width: 1080, output_height: 720})
+      |> child(:encoder, Encoder)
+      |> child(:out_parser, %Parser{output_stream_structure: :avc1})
       |> child(:sink, %Sink{location: opts[:output_path]})
     ]
 
-    {[spec: structure], %{}}
+    {[spec: spec], %{}}
   end
+
+  @impl true
+  def handle_element_end_of_stream(:sink, :input, _ctx, state), do: {[terminate: :normal], state}
+
+  @impl true
+  def handle_element_end_of_stream(_child, _pad, _ctx, state), do: {[], state}
 end
 ```
 
-To ensure reliability, we might wrap this execution in an [Oban](https://hexdocs.pm/oban/Oban.html) worker.
+To ensure reliability, we might wrap execution of the pipeline in an [Oban](https://hexdocs.pm/oban/Oban.html) worker.
 
 Oban is the standard library for background job processing in Elixir. It persists jobs to your database, ensuring that your long-running transcoding tasks are fault-tolerant.
 If the pipeline crashes or the server restarts, Oban will automatically retry the job until it succeeds.
 To install Oban in your project, you can follow this [installation guide](https://hexdocs.pm/oban/installation.html).
 
-Assuming that Oban is installed in your project we can define an Oban worker:
+Assuming that Oban is installed in your project and the `:default` queue is configured we can define the Oban worker:
 
 ```elixir
+
 defmodule VideoTranscoderWorker do
-  @timeout_sec :time
-    
-  use Oban.Worker, queue: :media, unique: [period: 60], timeout: :timer.seconds(10)
+  use Oban.Worker, queue: :default, unique: [period: 60]
 
   @impl true
   def perform(%Oban.Job{args: %{"input_path" => input, "output_path" => output}}) do
-    {:ok, _supervisor_pid, pipeline_pid} = Membrane.Pipeline.start_link(
-      TranscodePipeline, 
-      [input_path: input, output_path: output]
-    )
-
-    case Membrane.Pipeline.wait_for_termination(pipeline_pid, 100_000) do
-      :ok -> :ok
-      {:error, reason} -> {:error, reason}
+    {:ok, _supervisor_pid, pipeline_pid} =
+      Membrane.Pipeline.start_link(
+        TranscodePipeline,
+        input_path: input,
+        output_path: output
+      )
+
+    ref = Process.monitor(pipeline_pid)
+
+    receive do
+      {:DOWN, ^ref, :process, ^pipeline_pid, :normal} ->
+        :ok
     end
   end
+
+  @impl true
+  def timeout(_job), do: :timer.minutes(5)
 end
 ```
 
 Now we can run the Oban worker:
 
 ```elixir
 %{
-  "input_path" => "uploads/input_video.mp4",
-  "output_path" => "processed/transcoded_video.mkv"
+  "input_path" => "input.mp4",
+  "output_path" => "output.mp4"
 }
 |> VideoTranscoderWorker.new()
 |> Oban.insert()
 ```
 
-When the job terminates successfully, you will see the output `.mkv` file with transcoded video.
+When the job terminates successfully, you will see the output `output.mp4` file with transcoded video.