Fix rtsp track match, use membrane_simple_rtsp_server

examples.livemd

@@ -9,7 +9,7 @@ System.put_env("PATH", "/opt/homebrew/bin:#{System.get_env("PATH")}")
 # In case of problems installing Nx/EXLA/Bumblebee,
 # you can remove them and the Nx backend config below.
 # Examples that don't mention them should still work.
-Mix.install([:boombox, :kino, :nx, :exla, :bumblebee, :websockex])
+Mix.install([:boombox, :kino, :nx, :exla, :bumblebee, :websockex, :membrane_simple_rtsp_server])
 
 Nx.global_default_backend(EXLA.Backend)
 ```
@@ -264,7 +264,7 @@ Task.start_link(fn ->
     input: {:webrtc, "ws://localhost:8829"},
     output: {
       :stream,
-      # Audio format that the OpenAI API expects 
+      # Audio format that the OpenAI API expects
       video: false, audio: :binary, audio_format: :s16le, audio_channels: 1, audio_rate: 24_000
     }
   )
@@ -516,6 +516,18 @@ System.shell("ffplay #{out_dir}/mp4_webrtc_mp4.mp4")
 
 <!-- livebook:{"branch_parent_index":0} -->
 
+## Stream MP4 to HLS via RTSP
+
+To receive the stream, visit http://localhost:1234/hls.html after running the cell below
+
+```elixir
+rtsp_port = 8554
+Membrane.SimpleRTSPServer.start_link("#{input_dir}/bun.mp4", port: rtsp_port)
+Boombox.run(input: "rtsp://localhost:#{rtsp_port}/", output: "#{out_dir}/index.m3u8")
+```
+
+<!-- livebook:{"branch_parent_index":0} -->
+
 ## Stream MP4 via WebRTC
 
 To receive the stream, visit http://localhost:1234/webrtc_to_browser.html after running the cell below.

lib/boombox/rtmp.ex

@@ -19,17 +19,17 @@ defmodule Boombox.RTMP do
     handle_new_client = fn client_ref, app, stream_key ->
       if app == target_app and stream_key == target_stream_key do
         send(boombox, {:rtmp_client_ref, client_ref})
+        Membrane.RTMP.Source.ClientHandlerImpl
       else
         Membrane.Logger.warning("Unexpected client connected on /#{app}/#{stream_key}")
       end
     end
 
     server_options = %{
-      handler: %RTMP.Source.ClientHandlerImpl{controlling_process: self()},
       port: port,
       use_ssl?: use_ssl?,
       handle_new_client: handle_new_client,
-      client_timeout: 60_000
+      client_timeout: Membrane.Time.seconds(60)
     }
 
     {:ok, _server} =

lib/boombox/rtsp.ex

@@ -77,7 +77,7 @@ defmodule Boombox.RTSP do
 
           {[], Map.put(track_builders, :video, video_spec)}
 
-        %{rtpmap: %{encoding: "mpeg4-generic", type: :audio}} ->
+        %{rtpmap: %{encoding: "mpeg4-generic"}, type: :audio} ->
           audio_spec =
             get_child(:rtsp_source)
             |> via_out(Membrane.Pad.ref(:output, ssrc))

mix.exs

@@ -48,19 +48,21 @@ defmodule Boombox.Mixfile do
       {:membrane_core, "~> 1.1"},
       {:membrane_webrtc_plugin, "~> 0.22.0"},
       {:membrane_opus_plugin, "~> 0.20.3"},
-      {:membrane_aac_plugin, "~> 0.18.0"},
+      {:membrane_aac_plugin, "~> 0.19.0"},
       {:membrane_aac_fdk_plugin, "~> 0.18.0"},
       {:membrane_h26x_plugin, "~> 0.10.0"},
       {:membrane_h264_ffmpeg_plugin, "~> 0.32.0"},
       {:membrane_mp4_plugin, "~> 0.35.2"},
       {:membrane_realtimer_plugin, "~> 0.9.0"},
       {:membrane_http_adaptive_stream_plugin, "~> 0.18.5"},
-      {:membrane_rtmp_plugin, "~> 0.25.0"},
-      {:membrane_rtsp_plugin, "~> 0.3.0"},
+      {:membrane_rtmp_plugin, "~> 0.27.2"},
+      {:membrane_rtsp_plugin, "~> 0.5.0"},
       {:membrane_rtp_plugin, "~> 0.29.0"},
       {:membrane_ffmpeg_swresample_plugin, "~> 0.20.0"},
       {:membrane_hackney_plugin, "~> 0.11.0"},
       {:membrane_ffmpeg_swscale_plugin, "~> 0.16.0"},
+      {:ex_sdp, "~> 1.1"},
+      {:membrane_simple_rtsp_server, "~> 0.1.0", only: :test},
       {:image, "~> 0.54.0"},
       {:burrito, "~> 1.0", runtime: burrito?(), optional: true},
       {:ex_doc, ">= 0.0.0", only: :dev, runtime: false},

test/boombox_test.exs

@@ -143,17 +143,17 @@ defmodule BoomboxTest do
 
     parent_process_pid = self()
 
-    new_client_callback = fn client_ref, app, stream_key ->
+    handle_new_client = fn client_ref, app, stream_key ->
       send(parent_process_pid, {:client_ref, client_ref, app, stream_key})
+      Membrane.RTMP.Source.ClientHandlerImpl
     end
 
     {:ok, server} =
       Membrane.RTMPServer.start_link(
-        handler: %Membrane.RTMP.Source.ClientHandlerImpl{controlling_process: self()},
         port: port,
         use_ssl?: use_ssl?,
-        new_client_callback: new_client_callback,
-        client_timeout: 1_000
+        handle_new_client: handle_new_client,
+        client_timeout: Membrane.Time.seconds(5)
       )
 
     p = send_rtmp(url)
@@ -232,75 +232,46 @@ defmodule BoomboxTest do
     end)
   end
 
-  @tag :rtsp_mp4_video
-  async_test "rtsp video -> mp4", %{tmp_dir: tmp} do
-    rtp_server_port = 30_003
+  @tag :rtsp_mp4
+  async_test "rtsp -> mp4", %{tmp_dir: tmp} do
     rtsp_port = 8554
     output = Path.join(tmp, "output.mp4")
 
-    {:ok, _server} =
-      Membrane.RTSP.Server.start_link(
-        handler: Membrane.Support.RTSP.Server.Handler,
-        handler_config: %{fixture_path: @bbb_mp4},
-        address: {127, 0, 0, 1},
-        port: rtsp_port,
-        udp_rtp_port: rtp_server_port,
-        udp_rtcp_port: rtp_server_port + 1
-      )
+    Membrane.SimpleRTSPServer.start_link(@bbb_mp4, port: rtsp_port)
 
-    Boombox.run(input: "rtsp://localhost:#{rtsp_port}/livestream", output: output)
-    Compare.compare(output, "test/fixtures/ref_bun10s_rtsp.mp4", kinds: [:video])
+    Boombox.run(input: "rtsp://localhost:#{rtsp_port}/", output: output)
+    Compare.compare(output, "test/fixtures/ref_bun10s_rtsp_aac.mp4")
   end
 
-  @tag :rtsp_hls_video
-  async_test "rtsp video -> hls", %{tmp_dir: tmp} do
-    rtp_server_port = 30_005
+  @tag :rtsp_hls
+  async_test "rtsp -> hls", %{tmp_dir: tmp} do
     rtsp_port = 8555
-
-    {:ok, _server} =
-      Membrane.RTSP.Server.start_link(
-        handler: Membrane.Support.RTSP.Server.Handler,
-        handler_config: %{fixture_path: @bbb_mp4},
-        address: {127, 0, 0, 1},
-        port: rtsp_port,
-        udp_rtp_port: rtp_server_port,
-        udp_rtcp_port: rtp_server_port + 1
-      )
-
+    Membrane.SimpleRTSPServer.start_link(@bbb_mp4, port: rtsp_port)
     manifest_filename = Path.join(tmp, "index.m3u8")
-    Boombox.run(input: "rtsp://localhost:#{rtsp_port}/livestream", output: manifest_filename)
-    ref_path = "test/fixtures/ref_bun10s_aac_hls"
-    Compare.compare(tmp, ref_path, kinds: [:video], format: :hls)
+    Boombox.run(input: "rtsp://localhost:#{rtsp_port}/", output: manifest_filename)
+    ref_path = "test/fixtures/ref_bun10s_rtsp_aac_hls"
+    Compare.compare(tmp, ref_path, format: :hls)
   end
 
-  @tag :rtsp_webrtc_mp4_video
-  async_test "rtsp video -> webrtc -> mp4", %{tmp_dir: tmp} do
-    rtp_server_port = 30_007
+  @tag :rtsp_webrtc_mp4
+  async_test "rtsp -> webrtc -> mp4", %{tmp_dir: tmp} do
     rtsp_port = 8556
     output = Path.join(tmp, "output.mp4")
     signaling = Membrane.WebRTC.SignalingChannel.new()
 
-    {:ok, _server} =
-      Membrane.RTSP.Server.start_link(
-        handler: Membrane.Support.RTSP.Server.Handler,
-        handler_config: %{fixture_path: @bbb_mp4_v},
-        address: {127, 0, 0, 1},
-        port: rtsp_port,
-        udp_rtp_port: rtp_server_port,
-        udp_rtcp_port: rtp_server_port + 1
-      )
+    Membrane.SimpleRTSPServer.start_link(@bbb_mp4, port: rtsp_port)
 
     t =
       Task.async(fn ->
         Boombox.run(
-          input: "rtsp://localhost:#{rtsp_port}/livestream",
+          input: "rtsp://localhost:#{rtsp_port}/",
           output: {:webrtc, signaling}
         )
       end)
 
     Boombox.run(input: {:webrtc, signaling}, output: output)
     Task.await(t)
-    Compare.compare(output, "test/fixtures/ref_bun10s_rtsp.mp4", kinds: [:video])
+    Compare.compare(output, "test/fixtures/ref_bun10s_rtsp_opus_aac.mp4")
   end
 
   @tag :mp4_elixir_rotate_mp4
@@ -403,7 +374,8 @@ defmodule BoomboxTest do
         |> child(Membrane.Realtimer)
         |> child(:audio_parser, %Membrane.AAC.Parser{
           out_encapsulation: :none,
-          output_config: :esds
+          output_config: :esds,
+          audio_specific_config: nil
         })
         |> via_in(Pad.ref(:audio, 0))
         |> get_child(:rtmp_sink),

test/fixtures/ref_bun10s_rtsp_aac_hls/g3cFdmlkZW8.m3u8

@@ -0,0 +1,13 @@
+#EXTM3U
+#EXT-X-VERSION:7
+#EXT-X-TARGETDURATION:5
+#EXT-X-MEDIA-SEQUENCE:0
+#EXT-X-DISCONTINUITY-SEQUENCE:0
+#EXT-X-MAP:URI="muxed_header_g3cFdmlkZW8_part_0.mp4"
+#EXTINF:4.803654194,
+muxed_segment_0_g3cFdmlkZW8.m4s
+#EXTINF:4.288195805,
+muxed_segment_1_g3cFdmlkZW8.m4s
+#EXTINF:0.912866893,
+muxed_segment_2_g3cFdmlkZW8.m4s
+#EXT-X-ENDLIST

test/fixtures/ref_bun10s_rtsp_aac_hls/index.m3u8

@@ -0,0 +1,5 @@
+#EXTM3U
+#EXT-X-VERSION:7
+#EXT-X-INDEPENDENT-SEGMENTS
+#EXT-X-STREAM-INF:BANDWIDTH=1176669,AVERAGE-BANDWIDTH=972665,RESOLUTION=480x270,CODECS=",mp4a.40.2"
+g3cFdmlkZW8.m3u8

test/support/compare.ex

@@ -139,7 +139,7 @@ defmodule Support.Compare do
     Testing.Pipeline.terminate(p)
   end
 
-  @spec samples_min_square_error(binary, binary, pos_integer) :: non_neg_integer()
+  @spec samples_min_square_error(binary, binary, pos_integer) :: float()
   def samples_min_square_error(bin1, bin2, sample_size) do
     assert byte_size(bin1) == byte_size(bin2)