membrane_hls_plugin

defmodule Membrane.HLS.SourceBin do
  use Membrane.Bin

  alias HLS.Playlist
  alias HLS.Playlist.Master
  alias HLS.Storage
  alias Membrane.HLS.Format
  alias Membrane.HLS.Source

  @master_check_retry_interval_ms 1_000
  require Membrane.Logger

  def_output_pad(:output,
    availability: :on_request,
    accepted_format:
      %Membrane.RemoteStream{content_format: %format{}}
      when format in [Format.PackedAudio, Format.WebVTT, Format.MPEG]
  )

  def_options(
    storage: [
      spec: Storage.t(),
      description: "HLS.Storage implementation used to obtain playlist contents"
    ],
    master_playlist_uri: [
      spec: URI.t(),
      description: "URI of the master playlist"
    ]
  )

  @impl true
  def handle_init(_ctx, options) do
    {[],
     %{
       storage: options.storage,
       master_playlist_uri: options.master_playlist_uri,
       pad_to_child: %{}
     }}
  end

  @impl true
  def handle_playing(_ctx, state) do
    send(self(), :check_master_playlist)
    {[], state}
  end

  @impl true
  def handle_pad_added(pad = {Membrane.Pad, :output, {:rendition, rendition}}, _ctx, state) do
    media_uri = Playlist.build_absolute_uri(state.master_playlist_uri, extract_uri(rendition))
    stream_format = build_stream_format(rendition)
    child_name = {:source, pad}

    spec = [
      child(child_name, %Source{
        storage: state.storage,
        media_playlist_uri: media_uri,
        stream_format: stream_format
      })
      |> via_out(:output)
      |> bin_output(pad)
    ]

    {[
       {:spec, spec}
     ], put_in(state, [:pad_to_child, pad], child_name)}
  end

  @impl true
  def handle_pad_removed(pad, _ctx, state) do
    case Map.fetch(state.pad_to_child, pad) do
      {:ok, child_name} ->
        {[remove_child: child_name], update_in(state, [:pad_to_child], &Map.delete(&1, pad))}

      :error ->
        {[], state}
    end
  end

  @impl true
  def handle_info(
        :check_master_playlist,
        _ctx,
        state = %{storage: storage, master_playlist_uri: uri}
      ) do
    case Storage.get(storage, uri) do
      {:ok, data} ->
        playlist = Playlist.unmarshal(data, %Master{uri: uri})
        {[{:notify_parent, {:hls_master_playlist, playlist}}], state}

      {:error, reason} ->
        Membrane.Logger.warning("Master playlist check failed: #{inspect(reason)}")

        Membrane.Logger.warning(
          "Master playlist check attempt scheduled in #{@master_check_retry_interval_ms}ms"
        )

        Process.send_after(self(), :check_master_playlist, @master_check_retry_interval_ms)
        {[], state}
    end
  end

  defp extract_uri(%HLS.AlternativeRendition{uri: uri}), do: uri
  defp extract_uri(%HLS.VariantStream{uri: uri}), do: uri

  defp build_stream_format(%HLS.VariantStream{codecs: codecs}), do: %Format.MPEG{codecs: codecs}

  defp build_stream_format(%HLS.AlternativeRendition{type: :subtitles, language: lang}),
    do: %Format.WebVTT{language: lang}

  defp build_stream_format(%HLS.AlternativeRendition{type: :audio}), do: %Format.PackedAudio{}

  defp build_stream_format(rendition),
    do: raise(ArgumentError, "Unable to provide a proper cap for rendition #{inspect(rendition)}")
end