membrane_core

defmodule Membrane.Testing.Pipeline do
  @moduledoc """
  This Pipeline was created to reduce testing boilerplate and ease communication with it's elements.
  It also provides utility for receiving messages when `Pipeline` playback state changes
  and notifications it receives.

  ## Starting Pipeline

  When you want a build Pipeline to test your elements you need three things:
   - Pipeline Module
   - List of elements
   - Links between those elements

  When creating pipelines for tests the only essential part is the list of elements.
  In most cases during the tests elements are linked in a way that
  `:output` pad is linked to `:input` pad of subsequent element.
  So we only need to pass a list of elements and links can be generated automatically.

  To start a testing pipeline you need to build `Membrane.Testing.Pipeline.Options` struct
  and pass to to `Membrane.Testing.Pipeline.start_link/2`.

  ```
  options = %Membrane.Testing.Pipeline.Options {
    elements: [
      el1: MembraneElement1,
      el2: MembraneElement2,
      ...
    ]
  }
  {:ok, pipeline} = Membrane.Testing.Pipeline.start_link(options)
  ```

  See `Membrane.Testing.Pipeline.Options` for available options.
  Links are generated by `populate_links/1`.

  ## Receiving notifications about callbacks invocation

  In some cases, you want to get a notification when `Pipeline` reaches a certain state.
  You can achieve that by monitoring pipeline callbacks.

  ```
  options = %Membrane.Testing.Pipeline.Options {
    monitored_callbacks: [:handle_prepared_to_playing],
    test_process: pid
    ...
  }
  ```

  First, you need to configure which callbacks are to be monitored by putting their names
  into `monitored_callbacks` field of `Options` struct. Check `t:Membrane.Testing.Pipeline.Options.pipeline_callback/0`
  for list of available callback names.
  You also need to pass a `PID` of the process that will receive messages.

  ```
  import Membrane.Testing.Pipeline.Assertions
  assert_receive_message :handle_prepared_to_playing
  ```

  Then it is a matter of waiting for the right message to come.
  You can do that easily by using `Membrane.Testing.Pipeline.Assertions.assert_receive_message/3`.

  ## Messaging children

  You can send messages to children using their names specified in the elements list.
  Please check `message_child/3` for more details.

  """

  use Membrane.Pipeline

  alias Membrane.Element
  alias Membrane.Pipeline.Spec

  defmodule Options do
    @moduledoc """
    Structure representing `options` passed to testing pipeline.

    ## Monitored Callbacks
    A message will be sent to `test process` if the invoked callback is in the list of monitored callbacks.
    See `t:pipeline_callback/0` for available callbacks.

    ##  Test Process
    `pid` of process that shall receive messages when Pipeline invokes playback state change callback
    and receives notification.

    ## Elements
    List of element specs.

    ## Links
    Map describing links between elements.

    If links are not present or set to nil they will be populated automatically based on elements order
    using default pad names.
    """

    @enforce_keys [:elements]
    defstruct @enforce_keys ++ [:monitored_callbacks, :links, :test_process]

    @typedoc """
    Defines supported callback names.
    """
    @type pipeline_callback ::
            :handle_notification
            | :handle_playing_to_prepared
            | :handle_prepared_to_playing
            | :handle_prepared_to_stopped
            | :handle_stopped_to_prepared

    @type t :: %__MODULE__{
            monitored_callbacks: pipeline_callback() | nil,
            test_process: pid() | nil,
            elements: Spec.children_spec_t(),
            links: Spec.links_spec_t() | nil
          }
  end

  @doc """
  Links subsequent elements using default pads (linking `:input` to `:output` of previous element).

  ## Examples

      iex> Pipeline.populate_links([el1: MembraneElement1, el2: MembraneElement2])
      %{{:el1, :output} => {:el2, :input}}
  """
  @spec populate_links(elements :: Spec.children_spec_t()) :: Spec.links_spec_t()
  def populate_links(elements) do
    Enum.chunk_every(elements, 2, 1, :discard)
    |> Enum.map(fn [{output_name, _}, {input_name, _}] ->
      {{output_name, :output}, {input_name, :input}}
    end)
    |> Enum.into(%{})
  end

  @doc """
  Sends message to a child by Element name.

  ## Examples

  Knowing that `pipeline` has child named `sink`, message can be sent as follows:

      message_child(pipeline, :sink, {:message, "to handle"})
  """
  @spec message_child(pid(), Element.name_t(), any()) :: :ok
  def message_child(pipeline, child, message) do
    send(pipeline, {:for_element, child, message})
    :ok
  end

  @doc """
  Starts the Testing Pipeline and links it to the current process.

  Proxy for `Membrane.Pipeline.start_link/3`.
  """
  @spec start_link(pipeline_options :: Options.t(), process_options :: GenServer.options()) ::
          GenServer.on_start()
  def start_link(%Options{} = options, process_options \\ []) do
    Membrane.Pipeline.start_link(__MODULE__, options, process_options)
  end

  @impl true
  def handle_init(options)

  def handle_init(%Options{test_process: nil, monitored_callbacks: list})
      when is_list(list) and length(list) > 0,
      do: {:error, :no_pid_when_monitoring}

  def handle_init(%Options{monitored_callbacks: nil} = options),
    do: handle_init(%Options{options | monitored_callbacks: []})

  def handle_init(%Options{links: nil, elements: elements} = options) do
    new_links = populate_links(elements)
    handle_init(%Options{options | links: new_links})
  end

  def handle_init(args) do
    %Options{elements: elements, links: links} = args

    spec = %Membrane.Pipeline.Spec{
      children: elements,
      links: links
    }

    new_state = Map.take(args, [:monitored_callbacks, :test_process])
    {{:ok, spec}, new_state}
  end

  @impl true
  def handle_stopped_to_prepared(state),
    do: notify_parent(:handle_stopped_to_prepared, state)

  @impl true
  def handle_playing_to_prepared(state),
    do: notify_parent(:handle_playing_to_prepared, state)

  @impl true
  def handle_prepared_to_playing(state),
    do: notify_parent(:handle_prepared_to_playing, state)

  @impl true
  def handle_prepared_to_stopped(state),
    do: notify_parent(:handle_prepared_to_stopped, state)

  @impl true
  def handle_notification(notification, from, state),
    do: notify_parent({:handle_notification, {notification, from}}, state)

  @impl true
  def handle_other({:for_element, element, message}, state) do
    {{:ok, forward: {element, message}}, state}
  end

  def handle_other(message, state),
    do: notify_parent({:handle_other, message}, state)

  defp get_callback_name(name) when is_atom(name), do: name
  defp get_callback_name({name, _}), do: name

  defp notify_parent(message, state) do
    %{test_process: parent, monitored_callbacks: monitored_callbacks} = state

    if get_callback_name(message) in monitored_callbacks do
      send(parent, {__MODULE__, message})
    end

    {:ok, state}
  end
end