Update the rtsp_to_hls demo

rtsp_to_hls/lib/pipeline.ex

@@ -1,4 +1,4 @@
-defmodule Membrane.Demo.RtspToHls.Pipeline do
+defmodule Membrane.Demo.RTSPToHLS.Pipeline do
   @moduledoc """
   The pipeline, which converts the RTP stream to HLS.
   """
@@ -14,7 +14,8 @@ defmodule Membrane.Demo.RtspToHls.Pipeline do
       child(:source, %Membrane.RTSP.Source{
         transport: {:udp, options.port, options.port + 5},
         allowed_media_types: [:video],
-        stream_uri: options.stream_url
+        stream_uri: options.stream_url,
+        on_connection_closed: :send_eos
       }),
       child(
         :hls,
@@ -32,27 +33,47 @@ defmodule Membrane.Demo.RtspToHls.Pipeline do
      %{
        video: nil,
        output_path: options.output_path,
-       parent_pid: options.parent_pid,
-       rtp_started: false
+       parent_pid: options.parent_pid
      }}
   end
 
   @impl true
-  def handle_child_notification(
-        {:new_track, ssrc, %{type: :video} = track},
-        :source,
-        _ctx,
-        %{rtp_started: false} = state
-      ) do
+  def handle_child_notification({:new_tracks, tracks}, :source, _ctx, state) do
     Logger.debug(":new_rtp_stream")
 
+    {spec, rtp_playing} =
+      Enum.map_reduce(tracks, false, fn {ssrc, track}, rtp_playing ->
+        create_spec_for_track(ssrc, track, rtp_playing)
+      end)
+
+    if not rtp_playing, do: raise("No video tracks received")
+
+    {[spec: spec], state}
+  end
+
+  @impl true
+  def handle_child_notification({:track_playable, _ref}, :hls, _ctx, state) do
+    send(state.parent_pid, :track_playable)
+    {[], state}
+  end
+
+  @impl true
+  def handle_child_notification(notification, element, _ctx, state) do
+    Logger.warning("Unknown notification: #{inspect(notification)}, el: #{inspect(element)}")
+
+    {[], state}
+  end
+
+  @spec create_spec_for_track(pos_integer(), map(), boolean()) ::
+          {Membrane.ChildrenSpec.t(), boolean()}
+  defp create_spec_for_track(ssrc, %{type: :video} = track, false) do
     {spss, ppss} =
       case track.fmtp.sprop_parameter_sets do
         nil -> {[], []}
         parameter_sets -> {parameter_sets.sps, parameter_sets.pps}
       end
 
-    structure =
+    spec =
       get_child(:source)
       |> via_out(Pad.ref(:output, ssrc))
       |> child(
@@ -63,34 +84,22 @@ defmodule Membrane.Demo.RtspToHls.Pipeline do
           generate_best_effort_timestamps: %{framerate: {30, 1}}
         }
       )
-      |> via_in(:input, options: [encoding: :H264, segment_duration: Membrane.Time.seconds(4)])
+      |> via_in(:input,
+        options: [encoding: :H264, segment_duration: Membrane.Time.seconds(4)]
+      )
       |> get_child(:hls)
 
-    {[spec: structure], %{state | rtp_started: true}}
+    {spec, true}
   end
 
-  @impl true
-  def handle_child_notification({:new_track, ssrc, _track}, :source, _ctx, state) do
+  defp create_spec_for_track(ssrc, _track, rtp_playing) do
     Logger.warning("new_rtp_stream Unsupported stream connected")
 
-    structure =
+    spec =
       get_child(:rtp)
       |> via_out(Pad.ref(:output, ssrc))
       |> child({:fake_sink, ssrc}, Membrane.Element.Fake.Sink.Buffers)
 
-    {[spec: structure], state}
-  end
-
-  @impl true
-  def handle_child_notification({:track_playable, _ref}, :hls, _ctx, state) do
-    send(state.parent_pid, :track_playable)
-    {[], state}
-  end
-
-  @impl true
-  def handle_child_notification(notification, element, _ctx, state) do
-    Logger.warning("Unknown notification: #{inspect(notification)}, el: #{inspect(element)}")
-
-    {[], state}
+    {spec, rtp_playing}
   end
 end

rtsp_to_hls/lib/server/handler.ex

@@ -0,0 +1,82 @@
+defmodule Membrane.Demo.RTSPToHLS.Server.Handler do
+  @moduledoc false
+
+  use Membrane.RTSP.Server.Handler
+
+  require Membrane.Logger
+
+  alias Membrane.RTSP.Response
+  @test_pt 96
+  @test_clock_rate 90_000
+
+  @impl true
+  def init(config) do
+    config
+    |> Map.put(:pipeline_pid, nil)
+    |> Map.put(:socket, nil)
+  end
+
+  @impl true
+  def handle_open_connection(conn, state) do
+    %{state | socket: conn}
+  end
+
+  @impl true
+  def handle_describe(_req, state) do
+    sdp = """
+    v=0
+    m=video 0 RTP/AVP 96
+    a=control:/control
+    a=rtpmap:#{@test_pt} H264/#{@test_clock_rate}
+    a=fmtp:#{@test_pt} packetization-mode=1
+    """
+
+    response =
+      Response.new(200)
+      |> Response.with_header("Content-Type", "application/sdp")
+      |> Response.with_body(sdp)
+
+    {response, state}
+  end
+
+  @impl true
+  def handle_setup(_req, state) do
+    {Response.new(200), state}
+  end
+
+  @impl true
+  def handle_play(configured_media_context, state) do
+    media_context = configured_media_context |> Map.values() |> List.first()
+
+    {client_rtp_port, _client_rtcp_port} = media_context.client_port
+
+    arg = %{
+      socket: state.socket,
+      ssrc: media_context.ssrc,
+      pt: @test_pt,
+      clock_rate: @test_clock_rate,
+      client_port: client_rtp_port,
+      client_ip: media_context.address,
+      server_rtp_socket: media_context.rtp_socket,
+      fixture_path: state.fixture_path
+    }
+
+    {:ok, _sup_pid, pipeline_pid} =
+      Membrane.Demo.RTSPToHLS.Server.Pipeline.start_link(arg)
+
+    {Response.new(200), %{state | pipeline_pid: pipeline_pid}}
+  end
+
+  @impl true
+  def handle_pause(state) do
+    {Response.new(501), state}
+  end
+
+  @impl true
+  def handle_teardown(state) do
+    {Response.new(200), state}
+  end
+
+  @impl true
+  def handle_closed_connection(_state), do: :ok
+end

rtsp_to_hls/lib/server/pipeline.ex

@@ -0,0 +1,78 @@
+defmodule Membrane.Demo.RTSPToHLS.Server.Pipeline do
+  @moduledoc false
+
+  use Membrane.Pipeline
+
+  @spec start_link(map()) :: Membrane.Pipeline.on_start()
+  def start_link(config) do
+    Membrane.Pipeline.start_link(__MODULE__, config)
+  end
+
+  @impl true
+  def handle_init(_ctx, opts) do
+    spec =
+      child(:mp4_in_file_source, %Membrane.File.Source{
+        location: opts.fixture_path,
+        seekable?: true
+      })
+      |> child(:mp4_demuxer, %Membrane.MP4.Demuxer.ISOM{optimize_for_non_fast_start?: true})
+
+    {[spec: spec], opts}
+  end
+
+  @impl true
+  def handle_child_notification({:new_tracks, tracks}, :mp4_demuxer, _ctx, state) do
+    spec =
+      Enum.map(tracks, fn
+        {id, %Membrane.AAC{}} ->
+          get_child(:mp4_demuxer)
+          |> via_out(Pad.ref(:output, id))
+          |> child(Membrane.Debug.Sink)
+
+        {id, %Membrane.H264{}} ->
+          get_child(:mp4_demuxer)
+          |> via_out(Pad.ref(:output, id))
+          |> child(:parser, %Membrane.H264.Parser{
+            output_alignment: :nalu,
+            repeat_parameter_sets: true,
+            skip_until_keyframe: true,
+            output_stream_structure: :annexb
+          })
+          |> via_in(Pad.ref(:input, state.ssrc),
+            options: [payloader: Membrane.RTP.H264.Payloader]
+          )
+          |> child(:rtp, Membrane.RTP.SessionBin)
+          |> via_out(Pad.ref(:rtp_output, state.ssrc),
+            options: [
+              payload_type: state.pt,
+              clock_rate: state.clock_rate
+            ]
+          )
+          |> child(:realtimer, Membrane.Realtimer)
+          |> child(:udp_sink, %Membrane.UDP.Sink{
+            destination_address: state.client_ip,
+            destination_port_no: state.client_port,
+            local_socket: state.server_rtp_socket
+          })
+      end)
+
+    {[spec: spec], state}
+  end
+
+  @impl true
+  def handle_child_notification(_notification, _element, _ctx, state) do
+    {[], state}
+  end
+
+  @impl true
+  def handle_element_end_of_stream(:udp_sink, :input, _ctx, state) do
+    Process.sleep(50)
+    :gen_tcp.close(state.socket)
+    {[terminate: :normal], state}
+  end
+
+  @impl true
+  def handle_element_end_of_stream(_child, _pad, _ctx, state) do
+    {[], state}
+  end
+end

rtsp_to_hls/mix.exs

@@ -1,9 +1,9 @@
-defmodule Membrane.Demo.RtspToHls.MixProject do
+defmodule Membrane.Demo.RTSPToHLS.MixProject do
   use Mix.Project
 
   def project do
     [
-      app: :hls_proxy_api,
+      app: :rtsp_to_hls_demo,
       version: "0.1.0",
       elixir: "~> 1.13",
       elixirc_paths: elixirc_paths(Mix.env()),
@@ -33,7 +33,14 @@ defmodule Membrane.Demo.RtspToHls.MixProject do
       {:connection, "~> 1.1"},
       {:membrane_http_adaptive_stream_plugin, "~> 0.18.0"},
       {:membrane_realtimer_plugin, "~> 0.9.0"},
-      {:membrane_rtsp_plugin, "~> 0.2.0"}
+      {:membrane_rtsp_plugin,
+       github: "membraneframework-labs/membrane_rtsp_plugin",
+       branch: "allow-for-eos",
+       override: true},
+      {:membrane_rtsp, "~> 0.8.0"},
+      {:membrane_mp4_plugin, "~> 0.35.2", override: true},
+      {:membrane_udp_plugin, "~> 0.14.0"}
+      # {:membrane_rtsp_plugin, "~> 0.2.0"}
     ]
   end