Additional features you might find usefull while setting up your pipelines


In this section, we gathered some topics which should give you better insight into Membrane. We will base on the ["All you need to know about pipelines pt 1" tutorial](../basic_pipeline/01_Introduction.md), so make sure you have gone through it first.
The chapters are loosely related, so feel free to read them in any order you want.

**Table of contents**

- [Bin](../basic_pipeline_extension/02_Bin.md) - how to create a reusable group of [elements](../glossary/glossary.md#element)
- [Dynamic Pads](../basic_pipeline_extension/03_DynamicPads.md) - a more flexible way to define element's [pads](../glossary/glossary.md#pad)
- [Tests](../basic_pipeline_extension/04_Tests.md) - creating unit tests for the elements


Introductions

# Bin

A Membrane's bin is a container for [elements](../glossary/glossary.md#element), which allows for creating reusable groups of elements.
Bin is similar to a [pipeline](../glossary/glossary.md#pipeline) in that it consists of linked elements. Such bin can then be placed inside a pipeline and linked with other entities - elements or bins. Bins can also be nested within one another.
Bin also has another advantage - it manages its children, for instance by dynamically spawning or replacing them as the stream changes.

## Enclosing pipeline elements inside a bin

As you can see, we have [`Source`](../glossary/glossary.md#source) -> [`Ordering Buffer`](../glossary/glossary.md#jitter-buffer--ordering-buffer) -> [`Depayloader`](../glossary/glossary.md#payloader-and-depayloader) chain, which is duplicated.
![Pipeline scheme](assets/images/basic_pipeline.png) <br>

We can encapsulate these elements inside `Bin`.
![Pipeline scheme using bin](assets/images/basic_pipeline_bin.png) <br>

Notice that there is no direct connection between `Depayloader` and [`Mixer`](../glossary/glossary.md#mixer). We have to explicitly link the `Depayloader` with `Bin`'s output [pads](../glossary/glossary.md#pad) and then we will connect the output pads to `Mixer`'s input pads.

Let's define the bin's output pads and its elements.

**_`lib/Bin.ex`_**

```elixir
defmodule Basic.Bin do
  use Membrane.Bin

  def_output_pad :output, accepted_format: %Basic.Formats.Frame{encoding: :utf8}

  def_options input_filename: [
              type: :string,
              description: "Input file for conversation."
            ]

  @impl true
  def handle_init(_ctx, options) do
    spec = [
      child(:input, %Basic.Elements.Source{location: options.input_filename})
      |> child(:ordering_buffer, Basic.Elements.OrderingBuffer)
      |> to(:depayloader, %Basic.Elements.Depayloader{packets_per_frame: 4}) 
      |> bin_output(:output)
    ]

    {[spec: spec] %{} }
  end
end
```

The output pads of the bin are matching the ones we [defined for depayloader](/basic_pipeline/08_Depayloader.md#libelementsdepayloaderex-2).
Notice that the last link is between `depayloader` and the bin's output pads. In general, if we wanted to receive data in a bin we would need to link the first processing component in the bin with the `bin_input(<bin's input pad name>)`.

## Modifying pipeline using bin

Using the bin we created, we can replace the elements in the pipeline.

**_`lib/Pipeline.ex`_**


```elixir
defmodule Basic.Pipeline do
  @moduledoc """
  A module providing the pipeline, which aggregates and links the elements.
  """
  use Membrane.Pipeline

  @impl true
  def handle_init(_ctx, _opts) do
    spec = [
      child(:bin1, %Basic.Bin{input_filename: "input.A.txt"}) 
      |> via_in(:first_input)
      |> child(:mixer, Basic.Elements.Mixer),
      child(:bin2, %Basic.Bin{input_filename: "input.B.txt"}) 
      |> via_in(:second_input) 
      |> get_child(:mixer),
      get_child(:mixer) 
      |> child(:output, %Basic.Elements.Sink{location: "output.txt"})
    ]

    {[spec: spec], %{}}
  end
end
```

Combining the usage of the bin and [dynamic pads](03_DynamicPads.md) will result in an even cleaner and more scalable solution.


# Dynamic Pads

The solution we have implemented along the tutorial has at least one downside - it is definitely not easily extendable.
What if we needed to support mixing streams coming from three different speakers in the conversation?
Well, we would need to add another input [pad](../glossary/glossary.md#pad) in the [Mixer](../glossary/glossary.md#mixer), for instance - `:third_input` pad, and then update our [pipeline](../glossary/glossary.md#pipeline) definition:

**_`lib/Pipeline.ex`_**

```elixir
@impl true
def handle_init(_ctx, _opts) do
 spec = [
   ...
   child(:input3, %Basic.Elements.Source{location: "input.C.txt"}) 
   |> child(:ordering_buffer3, Basic.Elements.OrderingBuffer) 
   |> child(:depayloader3, %Basic.Elements.Depayloader{packets_per_frame: 4}),
   ...
   get_child(:depayloader3) 
   |> via_in(:third_input) 
   |> get_child(:mixer),
 ]
 ...
end
```

But what if there were 4 people in the conversation? Adding another pad to the Mixer would solve the problem, but this solution does not scale well.
And what if the number of speakers was unknown at the moment of the compilation? Then we wouldn't be able to predefine the pads in the Mixer.
The Membrane Framework comes with a solution for this sort of problem - and the solution is called: **dynamic pads**.

## What is the idea behind the dynamic pads?

Well, the idea is quite simple! Instead of specifying a single pad with a predefined name (*static pad*) as we did in all the modules before, we specify, that we want **a set of pads** of a given type. Initially, that set will be empty, but with each link created in the parent's child specification, the [element](../glossary/glossary.md#element) will be informed, that the pad of a given type was added - and therefore it will be able to invoke the `handle_pad_added/3` callback.

## The Mixer revisited

Let's try to use dynamic pads for the input in the Mixer!
The very first thing we need to do is to use the `def_input_pads` appropriately.

**_`lib/elements/Mixer.ex`_**

```elixir
...
def_input_pad :input, 
   demand_unit: :buffers, 
   flow_control: :pull, 
   availability: :on_request, 
   accepted_format: %Basic.Formats.Frame{encoding: :utf8}
...
```

We have added the [`availability: :on_request` option](https://hexdocs.pm/membrane_core/Membrane.Pad.html#t:availability/0), which allows us to define the set of dynamic pads, identified as `:input`.

No more do we have the `:first_input` and the `:second_input` pads defined, so we do not have the [tracks](../glossary/glossary.md#track) corresponding to them either! Let's update the `handle_init/2` callback:

**_`lib/elements/Mixer.ex`_**

```elixir
...
@impl true
def handle_init(_ctx, _options) do
    {[], %{ tracks: %{} }}
end
...
```

Tracks map is initially empty since there are no corresponding pads.
The next thing we need to do is to implement the `handle_pad_added/3` callback, which will be called once the pipeline starts, with some links pointing to dynamic pads:

**_`lib/elements/Mixer.ex`_**

```elixir
...
@impl true
def handle_pad_added(pad, _context, state) do
 state = %{state | tracks: Map.put(state.tracks, pad, %Track{})}
 {[], state}
end
...
```

Once a pad is created, we add a new `Track` to the tracks map, with the pad being its key.
That's it! Since we have already designed the Mixer in a way it is capable of serving more tracks, there is nothing else to do.

## Updated pipeline

Below you can find the updated version of the pipeline's `handle_init/2` callback:

**_`lib/Pipeline.ex`_**

```elixir
...
@impl true
def handle_init(_ctx, _opts) do
 spec = [
    child(:input1, %Basic.Elements.Source{location: "input.A.txt"}) 
    |> child(:ordering_buffer1, Basic.Elements.OrderingBuffer) 
    |> child(:depayloader1, %Basic.Elements.Depayloader{packets_per_frame: 4}),
    child(:input2, %Basic.Elements.Source{location: "input.B.txt"}) 
    |> child(:ordering_buffer2, Basic.Elements.OrderingBuffer) 
    |> child(:depayloader2, %Basic.Elements.Depayloader{packets_per_frame: 4}),
    get_child(:depayloader1) 
    |> via_in(Pad.ref(:input, :first))
    |> child(:mixer, Basic.Elements.Mixer),
    get_child(:depayloader2) 
    |> via_in(Pad.ref(:input, :second)) 
    |> get_child(:mixer),
    get_child(:mixer) 
    |> child(:output, %Basic.Elements.Sink{location: "output.txt"})
 ]

 {[spec: spec], %{}}
end
...
```

The crucial thing was to change the plain atom name identifying the pad (like `:first_input`) into the [Membrane.Pad.ref/2](https://hexdocs.pm/membrane_core/Membrane.Pad.html#ref/2).
The first argument passed to that function is the name of the dynamic pad's set (in our case: `:input`, as we have defined the `:input` dynamic pads set in the Mixer element), and the second argument is a particular pad identifier.
As you can see, we have created two `:input` pads: `:first` and `:second`. While starting the pipeline, the `handle_pad_added/3` callback will be called twice, once per each dynamic pad created.

## Further actions

As an exercise, you can try to modify the `lib/pipeline.ex` file and define a pipeline consisting of three parallel branches, being mixed in a single Mixer. Later on, you can check if the pipeline works as expected, by generating the input files out of the conversation in which participate three speakers.

If you combine the approach taken in the chapter about [Bin](02_Bin.md) you can simplify this solution by reducing the size of the link defintions inside the pipeline.


Dynamic Pads

# Tests

When creating elements for the basic pipeline you used the provided tests.
In this chapter we will explain in detail how they work, and give you with some good practices which will allow you to write reliable tests for your Membrane system.

## Do we need testing?

Testing in terms of software engineering is no less important than functionalities programming.
We are sure we do not need to persuade you that our [pipeline](../glossary/glossary.md#pipeline), and especially its elements need testing. In fact - who wouldn't like to be sure, that changes just made in the functionalities code do not break the desired element's behavior? And all that in the matter of typing simple `mix test` command?

In the scope of this chapter we will implement some unit tests for the elements of our pipeline. They will check the behavior of the elements in isolation.
We will immediately jump to the code and try to experience the Membrane tests on our own.

## Our first test

Elixir comes with a great tool for testing - [ExUnit](https://hexdocs.pm/ex_unit/ExUnit.html).
If you have never written tests in ExUnit, feel free to stop for a moment and read about how the tests are constructed - however, ExUnit deals with tests in such a clear way that probably you will be able to see what's going on by just looking at the code snippets in the further part of this tutorial.
We need to specify, that we will be writing only unit tests - that means, we will write tests checking the behavior of a single [element](../glossary/glossary.md#element), isolated from the other elements in the pipeline.
Let's create a `test/elements/depayloader_test.exs` file and put the following code inside it:

**_`test/elements/depayloader_test.exs`_**

```elixir
defmodule DepayloaderTest do
 use ExUnit.Case 
 doctest Basic.Elements.Depayloader

 test "Depayloader should assemble the packets and form a frame" do
  ...
 end
end
```

This way we are defining a testing module with the use of the ExUnit, as well as a single test (we will put the test logic inside the inner `do...end` scope).
The `doctest` macro checks if the given module has proper documentation (typespecs, module description etc.).

We have decided to show the process of writing tests based on the depayloader - since the behavior of this element is well described and can be easily checked. Let's recall what is the responsibility of the depayloader - this element is receiving ordered [packets](../glossary/glossary.md#packet) and is about to form [frames](../glossary/glossary.md#frame) out of them. Let's check if it is doing it properly!

**_`test/elements/depayloader_test.exs`_**

```elixir
defmodule DepayloaderTest do
 ...
 alias Basic.Elements.Depayloader
 alias Membrane.Buffer
 
 test "Depayloader should assemble the packets and form a frame" do
  {[], state} = Depayloader.handle_init(nil, %Depayloader{packets_per_frame: 5})

  {[], state} =
    Depayloader.handle_buffer(
      :input,
      %Buffer{payload: "[frameid:1s][timestamp:1]Hello! "},
      nil,
      state
    )

  {[], state} =
    Depayloader.handle_buffer(
      :input,
      %Buffer{payload: "[frameid:1][timestamp:1]How are"},
      nil,
      state
    )

  {actions, _state} =
    Depayloader.handle_buffer(
      :input,
      %Buffer{payload: "[frameid:1e][timestamp:1] you?"},
      nil,
      state
    )

  [buffer: {:output, buffer}] = actions
  assert buffer.payload == "Hello! How are you?"
 end
end
```

We have been explicitly calling the callbacks defined in the `Depayloader` module, with the appropriate arguments.
First, we have initialized the state by calling the `handle_init/2` callback, later on, we have made the depayloader `handle_buffer/4` some [buffers](../glossary/glossary.md#buffer) (note that we had to explicitly pass the `%Membrane.Buffer{}` structure as the argument of the function invocation as well as keep track of the state, which gets updated after each call).
Finally, after the last buffer is processed (the last buffer is the buffer whose `frameid` should contain the `e` letter meaning that that is the packet that is ending a given frame), we are expecting the action to be returned - and this action should be a `:buffer` actions, transmitting the buffer with the complete frame through the `:output` pad. We also assert (using the ExUnit's [`assert`](https://hexdocs.pm/ex_unit/ExUnit.Assertions.html#assert/1) macro) the value hold in the buffer's payload (It should be a complete sentence).
If no action was returned from the last `handle_buffer/4`, the pattern wouldn't match to the `{ {actions, _state}`, and the test would fail.
If the assertion on the output buffer's payload wouldn't be true - the test would also fail.

## The Membrane's support for tests

The test written above is quite simple. Probably you have noticed that what we did there was "simulating" the behavior of the Membrane's Core, in a limited, but satisfying our needs, way. During the pipeline run, it is Membrane's responsibility to invoke the callbacks and pass the updated version of the state as the argument.
However, Membrane's Core behavior is much more complicated - if we were using some more complex mechanism (i.e. [redemands](../glossary/glossary.md#redemands)), possibly we would need to simulate that behavior in a more detailed way - finally ending with the Membrane's Core in our test module.
Such an approach scales terribly - and that is why we want to avoid it. Membrane Core's developers have given us support for testing the elements which allow us to have a simple pipeline consisting of a generic [source](../glossary/glossary.md#source) and [sink](../glossary/glossary.md#sink), as well as our element.
Such a pipeline behaves just like any other pipeline in a regular working Membrane system - however, we are also given a bunch of helpful tools (like assertion macros) to check if our element has a desired business logic implemented in its behavior.
Below we will rewrite the test we have just written, but with the support from the Membrane Framework:

**_`test/elements/depayloader_test.exs`_**

```elixir
defmodule DepayloaderTest do
  ...
  alias Membrane.Buffer

  alias Membrane.Testing.{Source, Sink}

  import Membrane.Testing.Assertions
  alias Membrane.Testing.{Source, Sink, Pipeline}
  alias Basic.Formats.Packet

  test "Depayloader should assemble the packets and form a frame (with membrane's testing framework)" do
    inputs = [
      "[frameid:1s][timestamp:1]Hello! ",
      "[frameid:1][timestamp:1]How are",
      "[frameid:1e][timestamp:1] you?"
    ]

    generator = fn state, size ->
      if state == [] do
        {[end_of_stream: :output], state}
      else
        [payload | new_state] = state

        if size > 1 do
          {[buffer: {:output, %Buffer{payload: payload}}, redemand: :output], new_state}
        else
          {[buffer: {:output, %Buffer{payload: payload}}], new_state}
        end
      end
    end

    spec =
      child(:source, %Source{
        output: {initial_state, generator},
        stream_format: %Packet{type: :custom_packets}
      })
      |> child(:depayloader, %Depayloader{packets_per_frame: 5})
      |> child(:sink, Sink)

    pipeline = Pipeline.start_link_supervised!(spec: spec)
    assert_start_of_stream(pipeline, :sink)

    assert_sink_buffer(pipeline, :sink, %Buffer{payload: "Hello! How are you?"})

    assert_end_of_stream(pipeline, :sink)
    refute_sink_buffer(pipeline, :sink, _, 2000)
    Pipeline.terminate(pipeline)
 end
end
```

First, we have defined a `:inputs` list, consisting of the messages which will be wrapped by the `Membrane.Buffer` and used to "feed" our element.
Later on, we have specified the testing pipeline.
Our testing pipeline consists only of three elements - the source, the sink, and the element we are about to test.
We are specifying these elements by passing the list of link as a `:spec` option to the `Membrane.Testing.Pipeline`, in a similar manner as we do when we return the list of links with the `:spec` action in a regular pipeline.
The generic [`Membrane.Testing.Source`](https://hexdocs.pm/membrane_core/Membrane.Testing.Source.html) accepts `:output` field as one of its options - we can pass the list of payloads which will be sent through the `:output` pad of the testing or a ("generator")[https://hexdocs.pm/membrane_core/Membrane.Testing.Source.html#t:generator/0]. In our case we use the "generator" which allows us to pass a function describing the source behaviour.
Once the pipeline structure is defined, we can start the pipeline process.
And here comes the assertions section - we are taking advantage of some [Membrane specific assertions](https://hexdocs.pm/membrane_core/Membrane.Testing.Assertions.html):

- first, we are asserting that the stream has started, with the `assert_start_of_stream/2`
- then we are asserting that the sink has received a buffer of a given form (in our case - we want the sink to receive the buffer with a frame assembled out of the input packets) - with help of `assert_sink_buffer/3`
- then we are asserting that `:end_of_stream` has reached the `:sink` - with `assert_end_of_stream/2`
- the last assertion we made is that the `:sink` hasn't received any buffer within 2000 milliseconds - and `refute_sink_buffer/4` helps us do it

Finally, we need to stop and terminate our pipeline. It is a good practice to do it in a blocking manner so that the test returns after the pipeline is terminated (the `Membrane.Pipeline.terminate` works in a blocking manner by default).

At the first glance, this might look like a little bit of overkill to use the Membrane's testing framework - the amount of code in this particular test has swollen enormously!
But that is just because the functionality we are testing is quite simple.
Keep in mind that in the second test we are making some additional, more complicated assertions - just imagine if you were about to check if no buffer has reached the `:sink` after the `:end_of_stream` was sent - it wouldn't be that easy, would it?
With the Membrane's testing framework you can do it in one line only!

Now we can run the tests with a simple [mix](../glossary/glossary.md#mix) task, by typing:

```console
mix test
```

If everything works (both the tests and the functionality's code itself), you should see a notification that the test has passed successfully, which we hope you do see!

## Some special types of tests

As you remember, Source and Sink elements act specifically different than the [Filter elements](../glossary/glossary.md#filter) - that is why they are communicating with the 'outer world', i.e. by reading the data from a file or saving the result to the file. In order to check if their behavior is desired, we cannot create a testing pipeline with generic Source and Sink, since it is a Source/Sink that we want to test.
We will need to somehow mock the `outer environment` - let's see how this can be done, based on the example of the Source test:

**_`test/elements/source_test.exs`_**

```elixir
defmodule SourceTest do
 use ExUnit.Case, async: false
  import Mock
  alias Basic.Elements.Source
  alias Membrane.Buffer

  doctest Basic.Elements.Source

  @exemplary_content ["First Line", "Second Line"]
  @exemplary_location "path/to/file"
  @options %Source{location: @exemplary_location}
 ...

  test "reads the input file correctly" do
    with_mock File, read!: fn _ -> "First Line\nSecond Line" end do
      {_actions, state} =
        Source.handle_setup(nil, %{location: @exemplary_location, content: nil})

      assert state.content == @exemplary_content
    end
 end
```

We take advantage of the [Mock](https://hexdocs.pm/mock/Mock.html) library which is designed to help us substitute the function invocations.
As you can see in the code snippet above, we have mocked the invocations of the `File.read!` function - inside the scope of `with_mock/2` they are always returning `"First Line\nSecondLine"`.
That makes our test so much easier since we do not need to create a mock file with some content meant just for testing - the whole test is defined inside one file.

Would you be able to write some more tests for the other pipeline's elements yourself? Or perhaps you could try to extend the tests for the element we have just checked in action?
Give it a try, as it is a great exercise that will not only examine if you are familiar with testing the Membrane's system but also - do you understand functionalities of the elements and what is the element generic behavior!

When writing your tests for other elements feel free to take inspiration from the ones we provided in the `test/` directory. You can find them [here](https://github.com/membraneframework/membrane_basic_pipeline_tutorial/tree/template/end/test).
