HLS HTTP output

.circleci/config.yml

@@ -7,20 +7,16 @@ workflows:
   build:
     jobs:
       - elixir/build_test:
-          cache-version: 7
           filters: &filters
             tags:
               only: /v.*/
       - elixir/test:
-          cache-version: 7
           filters:
             <<: *filters
       - elixir/lint:
-          cache-version: 7
           filters:
             <<: *filters
       - elixir/hex_publish:
-          cache-version: 7
           requires:
             - elixir/build_test
             - elixir/test

examples.livemd

@@ -6,7 +6,7 @@ Logger.configure(level: :info)
 # For ffmpeg and ffplay commands to work on Mac Livebook Desktop
 System.put_env("PATH", "/opt/homebrew/bin:#{System.get_env("PATH")}")
 
-Mix.install([{:boombox, path: __DIR__}, :kino, :nx, :exla, :bumblebee])
+Mix.install([{:boombox, path: __DIR__, env: :test}, :kino, :nx, :exla, :bumblebee])
 
 Nx.global_default_backend(EXLA.Backend)
 ```
@@ -87,10 +87,40 @@ Boombox.run(input: {:webrtc, "ws://localhost:8829"}, output: {:webrtc, "ws://loc
 To receive the stream, visit http://localhost:1234/hls/stream.html after running the cell below
 
 ```elixir
-Boombox.run(
-  input: "#{input_dir}/bun10s.mp4",
-  output: {:hls, "#{__DIR__}/examples_assets/hls/hls_output"}
-)
+hls_out_dir = "#{__DIR__}/examples_assets/hls/hls_output"
+
+"#{hls_out_dir}/*" |> Path.wildcard() |> Enum.each(&File.rm!/1)
+
+Boombox.run(input: "#{input_dir}/bun10s.mp4", output: {:hls, hls_out_dir})
+```
+
+<!-- livebook:{"branch_parent_index":0} -->
+
+## RTSP to HLS
+
+To receive the stream, visit http://localhost:1234/hls/stream.html after running the cell below
+
+```elixir
+hls_out_dir = "#{__DIR__}/examples_assets/hls/hls_output"
+
+"#{hls_out_dir}/*" |> Path.wildcard() |> Enum.each(&File.rm!/1)
+
+rtp_server_port = 30_003
+rtsp_port = 8554
+
+{: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
+  )
+
+Boombox.run(input: "rtsp://localhost:#{rtsp_port}/livestream", output: "#{hls_out_dir}/index.m3u8")
+
+Membrane.RTSP.Server.stop(server)
 ```
 
 <!-- livebook:{"branch_parent_index":0} -->

lib/boombox.ex

@@ -22,13 +22,14 @@ defmodule Boombox do
           | {:mp4, location :: String.t(), transport: :file | :http}
           | {:webrtc, webrtc_signaling()}
           | {:rtmp, (uri :: String.t()) | (client_handler :: pid)}
+          | {:rtsp, url :: String.t()}
           | {:stream, in_stream_opts()}
 
   @type output ::
           (path_or_uri :: String.t())
           | {:mp4, location :: String.t()}
           | {:webrtc, webrtc_signaling()}
-          | {:hls, location :: String.t()}
+          | {:hls, location :: String.t(), transport: :file | :http}
           | {:stream, out_stream_opts()}
 
   @typep procs :: %{pipeline: pid(), supervisor: pid()}
@@ -84,7 +85,8 @@ defmodule Boombox do
       {scheme, ".mp4", :input} when scheme in [nil, "http", "https"] -> {:mp4, value}
       {nil, ".mp4", :output} -> {:mp4, value}
       {scheme, _ext, :input} when scheme in ["rtmp", "rtmps"] -> {:rtmp, value}
-      {nil, ".m3u8", :output} -> {:hls, value}
+      {"rtsp", _ext, :input} -> {:rtsp, value}
+      {scheme, ".m3u8", :output} when scheme in [nil, "http", "https"] -> {:hls, value}
       _other -> raise ArgumentError, "Unsupported URI: #{value} for direction: #{direction}"
     end
     |> then(&parse_opt!(direction, &1))
@@ -113,6 +115,13 @@ defmodule Boombox do
         value
 
       {:hls, location} when direction == :output and is_binary(location) ->
+        parse_opt!(:output, {:hls, location, []})
+
+      {:hls, location, opts} when direction == :output and is_binary(location) ->
+        if Keyword.keyword?(opts),
+          do: {:hls, location, transport: resolve_transport(location, opts)}
+
+      {:rtsp, location} when direction == :input and is_binary(location) ->
         value
 
       {:stream, opts} ->

lib/boombox/hls.ex

@@ -4,22 +4,112 @@ defmodule Boombox.HLS do
   import Membrane.ChildrenSpec
 
   require Membrane.Pad, as: Pad
+
   alias Boombox.Pipeline.Ready
-  alias Membrane.Time
+  alias Membrane.{HTTPAdaptiveStream, Time, UtilitySupervisor}
+
+  defmodule HTTPUploader do
+    @moduledoc false
+    use GenServer
+
+    require Logger
+
+    alias Membrane.HTTPAdaptiveStream.Storages.GenServerStorage
+
+    @spec start_link(directory: String.t()) :: GenServer.on_start()
+    def start_link(config) do
+      GenServer.start_link(__MODULE__, config)
+    end
+
+    @impl true
+    def init(config) do
+      {:ok, %{directory: config[:directory]}}
+    end
+
+    @impl true
+    def handle_call(
+          {GenServerStorage, :store, %{context: %{type: :partial_segment}}},
+          _from,
+          state
+        ) do
+      Logger.warning("LL-HLS is not supported. The partial segment is omitted.")
+      {:reply, :ok, state}
+    end
+
+    @impl true
+    def handle_call({GenServerStorage, :store, params}, _from, state) do
+      location = Path.join(state.directory, params.name)
+
+      reply =
+        :hackney.request(:post, location, [], params.contents, follow_redirect: true)
+        |> handle_request_result()
+
+      {:reply, reply, state}
+    end
+
+    @impl true
+    def handle_call({GenServerStorage, :remove, params}, _from, state) do
+      location = Path.join(state.directory, params.name)
+
+      reply =
+        :hackney.request(:delete, location, [], <<>>, follow_redirect: true)
+        |> handle_request_result()
+
+      {:reply, reply, state}
+    end
+
+    @spec handle_request_result(
+            {:ok, pos_integer(), list(), :hackney.client_ref()}
+            | {:error, term()}
+          ) :: :ok | {:error, term()}
+    defp handle_request_result(result) do
+      case result do
+        {:ok, status, _headers, _ref} when status in 200..299 ->
+          :ok
+
+        {:ok, status, _headers, ref} ->
+          {:ok, body} = :hackney.body(ref)
+          {:error, "Request failed with status code #{status}: #{body}"}
+
+        error ->
+          error
+      end
+    end
+  end
 
   @spec link_output(
           Path.t(),
           Boombox.Pipeline.track_builders(),
-          Membrane.ChildrenSpec.t()
+          Membrane.ChildrenSpec.t(),
+          UtilitySupervisor.t(),
+          transport: :file | :http
         ) :: Ready.t()
-  def link_output(location, track_builders, spec_builder) do
+  def link_output(location, track_builders, spec_builder, utility_supervisor, opts) do
     {directory, manifest_name} =
       if Path.extname(location) == ".m3u8" do
         {Path.dirname(location), Path.basename(location, ".m3u8")}
       else
         {location, "index"}
       end
 
+    storage =
+      case opts[:transport] do
+        :file ->
+          %HTTPAdaptiveStream.Storages.FileStorage{directory: directory}
+
+        :http ->
+          {:ok, uploader} =
+            UtilitySupervisor.start_link_child(
+              utility_supervisor,
+              {HTTPUploader, directory: directory}
+            )
+
+          %HTTPAdaptiveStream.Storages.GenServerStorage{destination: uploader}
+      end
+
+    hls_mode =
+      if Map.keys(track_builders) == [:video], do: :separate_av, else: :muxed_av
+
     spec =
       [
         spec_builder,
@@ -28,10 +118,8 @@ defmodule Boombox.HLS do
           %Membrane.HTTPAdaptiveStream.SinkBin{
             manifest_name: manifest_name,
             manifest_module: Membrane.HTTPAdaptiveStream.HLS,
-            storage: %Membrane.HTTPAdaptiveStream.Storages.FileStorage{
-              directory: directory
-            },
-            hls_mode: :muxed_av
+            storage: storage,
+            hls_mode: hls_mode
           }
         ),
         Enum.map(track_builders, fn

lib/boombox/mp4.ex

@@ -38,7 +38,10 @@ defmodule Boombox.MP4 do
           {:audio, spec}
 
         {id, %Membrane.H264{}} ->
-          spec = get_child(:mp4_demuxer) |> via_out(Pad.ref(:output, id))
+          spec =
+            get_child(:mp4_demuxer)
+            |> via_out(Pad.ref(:output, id))
+
           {:video, spec}
       end)
 

lib/boombox/pipeline.ex

@@ -61,7 +61,8 @@ defmodule Boombox.Pipeline do
                   spec_builder: [],
                   track_builders: nil,
                   last_result: nil,
-                  eos_info: nil
+                  eos_info: nil,
+                  rtsp_state: nil
                 ]
 
     @typedoc """
@@ -90,6 +91,7 @@ defmodule Boombox.Pipeline do
             track_builders: Boombox.Pipeline.track_builders() | nil,
             last_result: Boombox.Pipeline.Ready.t() | Boombox.Pipeline.Wait.t() | nil,
             eos_info: term(),
+            rtsp_state: Boombox.RTSP.rtsp_state() | nil,
             parent: pid()
           }
   end
@@ -112,6 +114,18 @@ defmodule Boombox.Pipeline do
     |> proceed_result(ctx, state)
   end
 
+  @impl true
+  def handle_child_notification({:set_up_tracks, tracks}, :rtsp_source, ctx, state) do
+    {result, state} = Boombox.RTSP.handle_set_up_tracks(tracks, state)
+    proceed_result(result, ctx, state)
+  end
+
+  @impl true
+  def handle_child_notification({:new_track, ssrc, track}, :rtsp_source, ctx, state) do
+    {result, state} = Boombox.RTSP.handle_input_track(ssrc, track, state)
+    proceed_result(result, ctx, state)
+  end
+
   @impl true
   def handle_child_notification({:new_tracks, tracks}, :webrtc_input, ctx, state) do
     Boombox.WebRTC.handle_input_tracks(tracks)
@@ -284,6 +298,10 @@ defmodule Boombox.Pipeline do
     Boombox.RTMP.create_input(src, ctx.utility_supervisor)
   end
 
+  defp create_input({:rtsp, uri}, _ctx, _state) do
+    Boombox.RTSP.create_input(uri)
+  end
+
   defp create_input({:stream, params}, _ctx, state) do
     Boombox.ElixirStream.create_input(state.parent, params)
   end
@@ -314,8 +332,8 @@ defmodule Boombox.Pipeline do
     Boombox.MP4.link_output(location, track_builders, spec_builder)
   end
 
-  defp link_output({:hls, location}, track_builders, spec_builder, _ctx, _state) do
-    Boombox.HLS.link_output(location, track_builders, spec_builder)
+  defp link_output({:hls, location, opts}, track_builders, spec_builder, ctx, _state) do
+    Boombox.HLS.link_output(location, track_builders, spec_builder, ctx.utility_supervisor, opts)
   end
 
   defp link_output({:stream, opts}, track_builders, spec_builder, _ctx, state) do

lib/boombox/rtmp.ex

@@ -4,9 +4,9 @@ defmodule Boombox.RTMP do
   import Membrane.ChildrenSpec
   require Membrane.Logger
   alias Boombox.Pipeline.{Ready, Wait}
-  alias Membrane.{RTMP, RTMPServer}
+  alias Membrane.{RTMP, RTMPServer, UtilitySupervisor}
 
-  @spec create_input(String.t() | pid(), pid()) :: Wait.t()
+  @spec create_input(String.t() | pid(), UtilitySupervisor.t()) :: Wait.t()
   def create_input(client_ref, _utility_supervisor) when is_pid(client_ref) do
     handle_connection(client_ref)
   end
@@ -43,12 +43,11 @@ defmodule Boombox.RTMP do
 
   @spec handle_connection(pid()) :: Ready.t()
   def handle_connection(client_ref) do
-    spec = [
+    spec =
       child(:rtmp_source, %RTMP.SourceBin{client_ref: client_ref})
       |> via_out(:audio)
       |> child(:rtmp_in_aac_parser, Membrane.AAC.Parser)
       |> child(:rtmp_in_aac_decoder, Membrane.AAC.FDK.Decoder)
-    ]
 
     track_builders = %{
       audio: get_child(:rtmp_in_aac_decoder),