# Sink

The sink is the last [element](../glossary/glossary.md#element) in our [pipeline](../glossary/glossary.md#pipeline), designed to store the data processed by the pipeline.
In contrast to the [filter elements](../glossary/glossary.md#filter), it won't have any output [pad](../glossary/glossary.md#pad) - that is why we need to make our element `use Membrane.Sink` and define the input pad only.
Since we want to parameterize the usage of that element, it will be good to define the options structure, so that we can specify the path to the file where the output should be saved. This stuff is done in the code snippet below:

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

```elixir
defmodule Basic.Elements.Sink do
 use Membrane.Sink

 def_options location: [
 spec: String.t(),
 description: "Path to the file"
 ]

 def_input_pad :input,
 flow_control: :manual,
 demand_unit: :buffers,
 accepted_format: _any
 ...
end
```

No surprises there - now we need to specify the element's behavior by defining the relevant callbacks!

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

```elixir

defmodule Basic.Elements.Sink do
 ...
 @impl true
 def handle_init(_context, options) do
 {[], %{location: options.location}}
 end
 ...
end
```

We have started with `handle_init/2`, where we are initializing the state of the element (we need to store the path to the output files).

Later on, we can specify the `handle_playing/2` callback - this callback gets called once the pipeline's setup finishes - that is a moment when we can demand the [buffers](../glossary/glossary.md#buffer) for the first time (since the pipeline is already prepared to work):

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

```elixir

defmodule Basic.Elements.Sink do
 ...
 @impl true
 def handle_playing(_context, state) do
 {[demand: {:input, 10}], state}
 end
 ...
end
```

There is only one more callback that needs to be specified - `handle_buffer/4`, which get's called once there are some buffers that can be processed (which means, that there are buffers to be written since there are no output pads through which we could be transmitting these buffers):

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

```elixir

defmodule Basic.Elements.Sink do
 ...
 @impl true
 def handle_buffer(:input, buffer, _context, state) do
 File.write!(state.location, buffer.payload <> "\n", [:append])
 {[demand: {:input, 10}], state}
 end
end
```

Note that after the successful writing, we are taking the `:demand` action and we ask for some more buffer.

With the `Sink` completed, we have implemented all elements of our pipeline. Now let's move to the very last step - defining the actual pipeline using the elements we have created.

Sink

See how pipeline works in Membrane and learn how to set it up for you

In this tutorial, we will create our very own Membrane Framework consisting of our custom which will fulfill the multimedia processing task.
Despite the fact that the multimedia processing task we will be facing will be really simple, we will need to deal with some problems occurring in real-life scenarios.
At the same time, we will make ourselves comfortable with some concepts of multimedia streaming as well as get in touch with the nomenclature used in this field.
Prepare for an adventure!

## Environment preparation

In order to be able to proceed with the tutorial, you need to have Elixir installed: [How to install Elixir](https://elixir-lang.org/install.html).
We assume, that you are at least slightly familiar with that language - if that is not true, we would like to strongly encourage you to take a look at the [official Elixir tutorial](https://elixir-lang.org/getting-started/introduction.html).
Once you are ready with the Elixir, you can get the project template we have prepared for you:

```console
git clone https://github.com/membraneframework/membrane_basic_pipeline_tutorial
cd membrane_basic_pipeline_tutorial
git checkout template/start
mix deps.get
```

As you can see, the template's code is put on the `template/start` branch of the repository.
In the repository, there is also a `template/end` branch, where you can find the completed template.
If you find yourself lost during the tutorial feel free to check the implementation proposed by us, put on this branch.

## Use Case

Imagine that there is a conversation occurring between two peers and the chat is based on a question-answer scheme.
Each of the peers sends its part of the conversation (that means - the first peer sends questions and the second peers sends the answers to them). We refer to each sentence sent by the peers as a [**frame**](../glossary/glossary.md#frame).
However, due to network limitations each frame sent by the peers is fragmented since only a few chars can be sent at the same time - that means, a single frame is sent as a sequence of [**packets**](../glossary/glossary.md#packet).
In a real network there is a parameter called MTU (*Maximum Transmission Unit*) which is the maximum number of bytes which can be send in a single packet. For the Ethernet II MTU is equal to 1500 bytes. We wanted to simulate the situation,
in which the data needs to be fragmented in order to be transmitted - but we decided to choose a small MTU so that this fragmentation is easy to be spotted.
Each packet consists of the *header* (describing where the particular bunch of characters it transports should be put) and the *body* - aforementioned bunch of characters.
Below you can see an exemplary frame sent by one peer to the other. It gets fragmented into multiple packets, which later on are sent via the network (during that process their order will probably get messed). On the second peer's side, the packets get assembled into the original frame.
![Example Chat](assets/images/example_chat.drawio.png)

### Packet format

Here is how each packet looks like:

```elixir
[seq:<sequence_id>][frameid:<frame_id>][timestamp:<timestamp>]<text>
```

where:

- sequence_id - the ordering number of the packet (relative to each of the peers). It reflects the position of a particular packet in the packets' stream.
- frame_id - the identifier which consists of the number of the frame to which the body of a given packet belongs, optionally followed by a single **'e'** character (meaning that the packet is the **e**nding packet of the frame). Note that frames are numbered relatively to each peer in that conversation and that frame_id does not describe the global order of the frames in the final file.
- timestamp - a number indicating a time at which a given sentence was said. Timestamp describes the order of the frames from both peers.
- text - the proper body of the packet, in our case - a bunch of characters which could be sent in a single packet.

### Packets generator

We have equipped you with the tool which produces the packets in the format described previously, based on the input conversation. You can use it as a [mix](../glossary/glossary.md#mix) task, by typing:

```console
mix generate_input --packetsPerFrame <packets_per_frame> <input_file_path>
```

where:

- *input_file_path* is a path to the file which contains the text of the conversation. The input file consists of multiple lines of the following format:

```
<speaker id>: <text>
```
 

- *packets_per_frame* is a number of packets on which each frame will be dismantled

Based on the input file content, that command will create multiple files, `<input_file_path>.<speaker id>.<input file format>` . For instance: `input.txt` with input from speakers `A` and `B` will be split into `input.A.txt` and `input.B.txt` files.
The first file will contain the shuffled list of packets made from the `A` speaker's lines from the input file and the second file will contain a shuffled list of packets made out of the `B` speaker's lines of the input file. That `shuffle` is a way to simulate the imperfectness of the network - in real-life scenario, the order in which the packets are received is not always the same as the order in which they were sent. [Here](https://www.openmymind.net/How-Unreliable-Is-UDP/) you can read about this phenomenon occurring while using [UDP](../glossary/glossary.md#udp).
Below you can see the steps which are taken during the input files generation: 
![Example Input](assets/images/example_input.drawio.png)

Create a file with your own input conversation occurring between two speakers or use the `input.txt` file where we have provided you such a conversation. Generate the files containing packets, with the use of `mix generate_input` task.

## Task description

By the end of this chapter you should have generated the files containing packets (i.e. `input.A.txt` and `input.B.txt`).

Your task is to assemble the conversation together from those packets. In the first step you will reproduce the sentences sent by each of the peers by gathering the words sent in packets in the appropriate order. In multimedia processing, we would say that you need to reproduce the *frame* out of the *transport layer packets*. Later on you need to put those *frames* in the correct order, producing the *raw data stream*.
The resulting file should be the same as the input file from which you have created two packets list with the use of the `InputFilesGenerator`.

Introduction

# System Architecture

Once we know what should be done, let's start thinking how should our system look like!

First, let's get familiar with some terms. In the Membrane Framework, there is a concept of the [`pipeline`](../glossary/glossary.md#pipeline) which consists of multiple [`elements`](../glossary/glossary.md#element). Elements are linked with the use of the [`pads`](../glossary/glossary.md#pad), which can be input pads or output pads.
Depending on what type of pads the particular element is equipped with, we distinguish the following types of elements:

- Source - element with only output pads, the first element of each pipeline. It is responsible for fetching the data and transmitting it through the output pad.
- Filter - element with both the input pads and the output pads. Its purpose is to get the data via the input pad, transform it and finally transmit the processed data via the output pad.
- Sink - element with only input pads, the last element of the pipeline. It can be responsible for storing or playing the data received via the input pad.
 Between the pads, the [`buffers`](../glossary/glossary.md#buffer) are being sent.
 Buffers sent between the pads should be of the same type. In the Membrane Framework nomenclature, we say, that the pads should have corresponding capabilities - [`caps`](../glossary/glossary.md#caps).
 One might wonder when does the event of the buffers being sent occurs - well, it depends on the pad's mode!
 If the pad works in the *pull* mode, the buffer is sent when it is demanded by the succeeding element in the pipeline.
 Otherwise, the pad works in the *push* mode, which means that it is pushing the buffers once they are ready, independently from the fact if the following elements want (and are capable of processing) it.
 In our pipeline, all the pads will work in the *pull* mode, which inducts a flow control by a [backpressure](https://medium.com/@jayphelps/backpressure-explained-the-flow-of-data-through-software-2350b3e77ce7) mechanism.

## Scheme

![Pipeline scheme](assets/images/basic_pipeline.png)

As you can see, our pipeline will consist of two twin branches, one per each of the peers. The branches will be merged with the `Mixer` element and the result produced by this element will be put in the file with the `Sink` elements.
Here you can find the description of the particular elements of the system.

## Elements description

- **Source** - that element is responsible for reading the list of [packets](../glossary/glossary.md#packet) from the file.
- **Ordering Buffer** - this element is responsible for reordering the packets based on their sequence id. The most important part of this element is the buffer, within which there is a place for all the packets, identified by the packet's sequence id. Once the new packet is received, it is placed in the proper place in the buffer, depending on the sequence id. If there is a consistent part of packets "at the bottom of the buffer" (which means - packets with the subsequent sequence ids, which haven't already been sent yet), then this part is sent via the output pad. That is how we make sure that the next element will receive elements sorted by the sequence id.
 Below you can see how the Ordering Buffer is expected to work, once the message "How are you doing?" will be sent in the four packets, each with one word, and received in the following order: (are, you, how, doing?) 
 ![Ordering Buffer](assets/images/ordering_buffer.drawio.png) 
 
- **Depayloader** - responsible for assembling the packets into the [frames](../glossary/glossary.md#frame). Since the depayloader receives the packets in the order of their sequence id (which means - the order in which they were 'sent'), it is capable of separating out the particular frames, basing on the **e** (ending packet of the frame) characters at the end of the frame id. This element also reads the timestamp from the packets' headers and puts it into buffer's 'pts' (*Presentation timestamp*) field.
- **Mixer** - responsible for mixing the frames received via two input pads. The order in which the frames are mixed is based on their 'pts' (*Presentation timestamp*).
- **Sink** - this element is responsible for writing the received frames to the file. Since the Mixer takes care of sorting the frames based on the timestamp, Sink can take advantage of the fact that the frames will be ordered and write them to the file in the order they are received.

Now, that you have some high-level understanding of the system we can get down to implementation.

System Architecture

# Source

Let's get to the code!
We will start where all the [pipelines](../glossary/glossary.md#pipeline) start - with the `Source` [element](../glossary/glossary.md#element).
Since this will be the first element we implement, we need to find out something more about how the Membrane Framework's elements should be implemented and some concepts associated with them.
The first thing you need to be aware of is that `Membrane. Element` describes a specific behavior, based on the OTP [GenServer's](https://elixir-lang.org/getting-started/mix-otp/genserver.html) behavior.
Our process keeps a state which is updated in callbacks.
We only need to provide an implementation of some callbacks in order to make our element act in the desired way.
The set of callbacks that can be implemented depends on the type of the elements and we will get familiar with them during the implementation of these elements.
However, each callback is required to return a tuple of a [specific form](https://hexdocs.pm/membrane_core/Membrane.Pipeline.html#t:callback_return/0).
As you can see, we are returning an optional list of [actions](https://hexdocs.pm/membrane_core/Membrane.Pipeline.Action.html#t:t/0) to be performed, and
the updated state (which later on will be passed to the next invoked callback).
Take your time and read about the possible actions which can be requested to be performed while returning from the callback. Their usage is crucial for the pipeline to work.

As you can judge based on the structure of the project, all the elements will be put in the `lib/elements` directory. Therefore there is a place where `source.ex` with the `Basic.Elements.Source` module's definition should be placed.

## What makes our module a Membrane Framework's element?

Let's start with specifying that our module will implement the `Membrane.Source` behavior as well as alias the modules which will be used later in the module's code:

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

```elixir
defmodule Basic.Elements.Source do
 use Membrane.Source
 alias Membrane.Buffer
 alias Basic.Formats.Packet
 ...
end
```

## Pads and options

Later on, we will make use of [macros](https://elixir-lang.org/getting-started/meta/macros.html) defined in the `Membrane.Source` module:

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

```elixir
defmodule Basic.Elements.Source do
 ...
  def_options location: [
                spec: String.t(),
                description: "Path to the file"
              ]

  def_output_pad :output,
    accepted_format: %Packet{type: :custom_packets},
    flow_control: :manual
 ...
end
```

The first macro, `def_options` allows us to define the parameters which are expected to be passed while instantiating the element. The parameters will be passed as an automatically generated structure `%Basic.Elements.Source{}`. In our case, we will have a `:location` field inside of that structure. This parameter is about to be a path to the files which will contain input [packets](../glossary/glossary.md#packet).
Later on, while instantiating the Source element, we will be able to write:

```elixir
%Basic.Elements.Source{location: "input.A.txt"}
```

and the `:location` option will be passed during the construction of the element.

The second macro, `def_output_pad`, lets us define the output pad. The pad name will be `:output` (which is a default name for the output pad). The second argument of the macro describes the `:accepted_format` - which is the type of data sent through the pad. As the code states, we want to send data in `Basic.Formats.Packet` format.
What's more, we have specified that the `:output` pad will work in `:manual` mode.
You can read more on pad specification [here](https://hexdocs.pm/membrane_core/Membrane.Pad.html#types).

## Initialization of the element

Let's define our first callback! Why not start with [`handle_init/2`](https://hexdocs.pm/membrane_core/Membrane.Element.Base.html#c:handle_init/2), which gets called once the element is created?

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

```elixir
defmodule Basic.Elements.Source do
 ...
  @impl true
  def handle_init(_context, options) do
    {[],
     %{
       location: options.location,
       content: nil
     }}
  end
 ...
end
```

As said before, `handle_init/2` expects a structure with the previously defined parameters to be passed as an argument.
All we need to do there is to initialize the state - our state will be in a form of a map, and for now on we will put there a `location` (a path to the input file) and the `content`, where we will be holding packets read from the file, which haven't been sent yet. The recommended approach is to keep `handle_init/2` lean and defer any heavier initialization to `handle_setup/2` which runs automatically afterwards, if defined. In the next section we'll talk about more involved initialization and resources managed by our element.

> ### TIP
>
> You might also wonder what is the purpose of the `@impl true` specifier, put just above the function signature - this is simply a way to tell the compiler
> that the function defined below is about to be a callback. If we have misspelled the function name (or provided a wrong arguments list), we will be informed in the compilation time.

## Preparing our element

In the `handle_setup/2` callback, we'd like to open, read and save the contents of the input file. We then save it in our state as `content`.

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

```elixir
defmodule Basic.Elements.Source do
 ...
  @impl true
  def handle_setup(_context, state) do
    content =
      File.read!(state.location)
      |> String.split("\n")

    new_state = %{state | content: content}
    {[], new_state}
  end
 ...
end
```

When the setup is complete, the element goes into `:playing` state. It can then demand buffers from previous elements and send its `:stream_format` to subsequent elements. Since we are implementing a source we do not have any previous element to demand from, but we can specify the format. We can do this, for example, in `handle_playing/2`:

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

```elixir
defmodule Basic.Elements.Source do
 ...
  @impl true
  def handle_playing(_context, state) do
    {[stream_format: {:output, %Packet{type: :custom_packets}}], state}
  end
 ...
end
```

The `:stream_format` action means that we want to transmit the information about the supported [formats](../glossary/glossary.md#stream-format-formerly-caps) through the `output` pad, to the next element in the pipeline. In [chapter 4](../basic_pipeline/04_StreamFormat.md) you will find out more about stream formats and learn why it is required to do so.

## Demands

Before going any further let's stop for a moment and talk about the demands. Do you remember, that the `:output` pad is working in `:manual` mode? That means that the succeeding element has to ask the Source element for the data to be sent and our element has to take care of keeping that data in some kind of buffer until it is requested.
Once the succeeding element requests for the data, the `handle_demand/4` callback will be invoked - therefore it would be good for us to define it:

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

```elixir
defmodule Basic.Elements.Source do
 ...

  @impl true
  def handle_demand(:output, _size, :buffers, _context, state) do
    if state.content == [] do
      {[end_of_stream: :output], state}
    else
      [first_packet | rest] = state.content
      new_state = %{state | content: rest}

      actions = [
        buffer: {:output, %Buffer{payload: first_packet}},
        redemand: :output
      ]

      {actions, new_state}
    end
  end
 ...
end
```

The callback's body describes the situation in which some buffers were requested. Then we are checking if we have any packets left in the list persisting in the state of the element. If that list is empty, we are sending an `end_of_stream` action, indicating that there will be no more buffers sent through the `:output` pad and that is why there is no point in requesting more buffers. 

However, in case of the `content` list of packets being non-empty, we are taking the head of that list, and storing the remaining tail of the list in the state of the element. Later on, we are defining the actions we want to take - that is, we want to return a buffer with the head packet from the original list. We make use of the [`buffer:` action](https://hexdocs.pm/membrane_core/Membrane.Element.Action.html#t:buffer/0), and specify that we want to transmit the [`%Buffer`](https://hexdocs.pm/membrane_core/Membrane.Buffer.html#t:t/0) structure through the `:output` pad. Note the fields available in the `%Buffer` structure - in our case, we make use of only the `:payload` field, which, according to the documentation, can be of `any` type - however, in almost all cases you will need to send binary data within this field. Any structured data (just like timestamps etc.) should be passed in the other fields available in the `%Buffer`, designed especially for that cases. 

Then there's the other action that is taken - the `:redemand` action, queued to take place on the `:output` pad. This action will simply invoke the `handle_demand/4` callback once again, which is helpful when the whole demand cannot be completely fulfilled in the single `handle_demand` invocation we are just processing. The great thing here is that the `size` of the demand will be automatically determined by the element and we do not need to specify it anyhow. Redemanding, in the context of sources, helps us simplify the logic of the `handle_demand` callback since all we need to do in that callback is to supply a single piece of data and in case this is not enough, take a [`:redemand`](https://hexdocs.pm/membrane_core/Membrane.Element.Action.html#t:redemand/0) action and invoke that callback once again. As you will see later, the process of redemanding is even more powerful in the context of [filter elements](../glossary/glossary.md#filter). 

> ### TIP
>
> Membrane also supports `:auto` demand mode, which should cover 90% of use cases. 
> 
> You can learn more about demand modes [here](https://membrane.stream/learn/get_started_with_membrane/6).

In the next chapter we will explore what stream formats are in Membrane.


Source

# Stream format

We owe you something...and we would like to pay it back as soon as possible!
As promised in the [3rd chapter](03_Source.md), we will talk more about the concept of stream formats - which in fact we have used in the previous chapter, but which weren't described sufficiently.

## Why formats are important

Specifying stream formats allows us to define what kind of data is flowing through the [pad](../glossary/glossary.md#pad). 
This isn't necessarily limited to data formats, as you'll see below.
In the Membrane Framework's nomenclature, we say, that we define a stream format specification for a given [element](../glossary/glossary.md#element).

We believe that an example might speak here louder than a plain definition, so we will try to describe it with a real-life scenario example.
Let's say that we are connecting two elements that process the video multimedia.
The link is made between the pads which are working on raw video data.
Here is where stream formats come up - they can be defined with the following constraints:

- data format - in our case, we are having a raw video format
- some additional constraints - i.e. [frame](../glossary/glossary.md#frame) resolution (480p) , framerate (30 fps) etc.

The `:stream_format` action helps us find out if the given elements are capable to communicate with each other. Not only can we not send the data between the pads if the format they are expecting is different - we need to take into consideration some other constraints! We can think of a situation in which the _data format_ would be the same (i.e. raw video data), but the element which receives the data performs a much more complex computation on that data than the sender, and therefore cannot digest such a great amount of data as the sender is capable of transmitting. Then their stream formats wouldn't be compatible, which could be expressed by adding some constraint, i.e. framerate.

Stream formats help us define a contract between elements and prevent us from connecting incompatible elements. That is why it is always better to define precise constraints rather than using `stream_format: :any`.

## When are stream formats compatible?

A comparison between formats is made when an input pad receives the `stream_format` action, and checks whether it matches its [accepted format](https://hexdocs.pm/membrane_core/Membrane.Pad.html#t:accepted_format/0).
Here is how you define a stream format specification:

1. First you need to specify the format module

```elixir
defmodule Formats.Raw do
 defstruct [:pixel_format, :framerate, :width, :height]
end
```

Module name defines the type of the format, however it is possible to pass some other options in a form of a struct. That is why we have defined a structure with the use of `defstruct`. Our format will be described with the following options:

- :pixel_format - pixel format, i.e. [I420](https://en.wikipedia.org/wiki/Chroma_subsampling) ([YUV](https://en.wikipedia.org/wiki/YUV)) or RGB888
- :framerate - number of frames per second, i.e. 30 (FPS)
- :width - width of the picture in pixels, i.e. 480 (px)
- :height - height of the picture in pixels, i.e. 300 (px)

2. We specify the pad of the element with the format we have just defined, using the `:accepted_format` option. For the purpose of an example, let it be the `:input` pad:

```elixir
 def_input_pad :input,
 demand_unit: :buffers,
 accepted_format:
 %Format.Raw{pixel_format: pixel_format, framerate: framerate, width: 480, height: 300}
 when pixel_format in [:I420, :I422] and framerate >= 30 and framerate <= 60
```
As you can see, the argument of that option is simply a match pattern. The incoming stream format is later confronted against that match pattern. If it does not match, an exception is thrown at the runtime. 

To simplify the pattern definition, there is `any_of/1` helper function that allows to define a alternative of match patterns - the matching will succeed if the stream format received on the pad matches any of the patterns listed as `any_of/1` argument. Below you can see an example of defining alternative of match patterns:

```elixir
def_input_pad :input,
 demand_unit: :buffers,
 accepted_format:
 any_of([
 %Format.Raw{pixel_format: pixel_format, framerate: framerate, width: 480, height: 300}
 when pixel_format in [:I420, :I422] and framerate >= 30 and framerate <= 60,
 %Format.Raw{pixel_format: pixel_format, framerate: framerate, width: 720, height: 480}
 when pixel_format in [:I420, :I422] and framerate >= 30 and framerate <= 60
 ])
```

Our journey with stream formats does not end here. We know how to describe their specification...but we also need to make our elements send the `:stream_format` events so that the following elements will be aware of what type of data our element is producing!

An element can send a stream format as one of the [actions](https://hexdocs.pm/membrane_core/Membrane.Element.Action.html) it can take - the [`:stream_format` action](https://hexdocs.pm/membrane_core/Membrane.Element.Action.html#t:stream_format/0).

Another thing is that we can specify the behavior of an element when it receives the stream format with the use of [`handle_stream_format/4` callback](https://hexdocs.pm/membrane_core/Membrane.Element.WithInputPads.html#c:handle_stream_format/4).

For all the [filter elements](../glossary/glossary.md#filter), `handle_stream_format/4` has a default implementation, which is relaying the received format on all the output pads of that filter.
However, if your filter is changing the format of data being sent, it should override the implementation of that callback to prevent formats flying through it, and send the proper spec via the output pads.

For the [source element](../glossary/glossary.md#source), it is necessary to send the format as in each [pipeline](../glossary/glossary.md#pipeline) the source is the first element - formats wouldn't flow through the pipeline if the source element wouldn't have sent them. Sending can be done in the `handle_playing/2` callback.

## Example

Imagine a pipeline, which starts with the source producing a video, which is then passed to the filter, responsible for reducing the quality of that video if it is too high.
For the source element, we should have the `:output` pads format which would allow us to send video in the higher and in the lower quality. The same format should be specified on the input of the filter element. However, the stream format on the output of the filter should accept only video in the lower quality.
Here is the definition of the source element:

```elixir
# Source element

defmodule Source do
 def_output_pad(:output,
 demand_unit: :buffers,
 stream_format: any_of([
 %Format.Raw{pixel_format: pixel_format, framerate: framerate, width: 480, height: 300}
 when pixel_format in [:I420, :I422] and framerate >= 30 and framerate <= 60,
 %Format.Raw{pixel_format: pixel_format, framerate: framerate, width: 720, height: 480}
 when pixel_format in [:I420, :I422] and framerate >= 30 and framerate <= 60
 ])
 )
 ...
 def handle_playing(_context, state) do
 ...
 { {[stream_format: {:output, %Formats.Raw{pixel_format: I420, framerate: 45, width: 720, height: 300} }]}, state}
 end
```

While returning from the `handle_playing/2` callback, the element will send the format described by the `Formats.Raw` structure, through the `:output` pad.
Will this format meet the accepted specification provided by us? Think about it!
In fact, it will, as the `Formats.Raw` structure sent with `:stream_format` action matches the pattern - the value of `:pixel_format` field is one of `:I420` and `:I422`, and the `:framerate` is in the range between 30 and 60. In case the structure didn't match the pattern, a runtime exception would be thrown.

Below there is the draft of the filter implementation:

```elixir
# Filter

defmodule Filter do
 def_input_pad:input,
 demand_unit: :buffers,
 accepted_format: any_of([
 %Format.Raw{pixel_format: pixel_format, framerate: framerate, width: 480, height: 300}
 when pixel_format in [:I420, :I422] and framerate >= 30 and framerate <= 60,
 %Format.Raw{pixel_format: pixel_format, framerate: framerate, width: 720, height: 480}
 when pixel_format in [:I420, :I422] and framerate >= 30 and framerate <= 60
 ])

 def_output_pad :output,
 demand_unit: :buffers,
 accepted_format: %Format.Raw{pixel_format: pixel_format, framerate: framerate, width: 480,height: 300} when pixel_format in [:I420, :I422] and framerate >= 30 and framerate <= 60

 ...

 def handle_stream_format(_pad, _stream_format, _context, state) do
 ...
 { {[stream_format: {:output, %Formats.Raw{pixel_format: I420, framerate: 60, width: 480, height: 300} }]}, state}
 end

end
```

When we receive the spec on the input pad, we do not propagate it to our `:output` pad - instead, we send a different format, with reduced quality (values of the `width` and `height` fields might be lower).

We hope by now you have a better understanding of what stream formats are. This knowledge will be helpful in the following chapters.

Stream format

# Stream formats in action

Since we have already discussed what stream formats are and how to use them, let's make use of them in our project!
The first thing to do is to define modules responsible for describing the formats used in our pipeline.
We will put them in a separate directory - `lib/formats`. Let's start with the format describing the [packets](../glossary/glossary.md#packet):

**_`lib/formats/packet_format.ex`_**

```elixir
defmodule Basic.Formats.Packet do
 @moduledoc """
 A module describing the format of the packet.
 """
 defstruct type: :custom_packets
end
```

The definition of the module is not complicated, as you can see in the code snippet above - we are only defining a structure within that module, with a `:type` parameter, whose default value is `:custom_packets`.

In our [pipeline](../glossary/glossary.md#pipeline) we will also send another type of data - [frames](../glossary/glossary.md#frame). Let's define a format for them:

**_`lib/formats/frame_format.ex`_**

```elixir
defmodule Basic.Formats.Frame do
 @moduledoc """
 A module describing the format of the frame.
 """
 defstruct encoding: :utf8
end
```

Same as in the case of the previous format - we are defining a structure with a single field, called `:encoding`, and the default value of that field - `:utf8`.

That's it! Format modules are really simple, and using them is not that hard either, as you will see in the subsequent chapters - defining the accepted stream formats and sending stream format actions!

Before advancing you can test the `Source` [element](../glossary/glossary.md/#source), using the tests provided in `/test` directory.

```console
mix test test/elements/source_test.exs
```

In case of errors, you may go back to the [Source chapter](/basic_pipeline/03_Source.md) or take a look how [Source](https://github.com/membraneframework/membrane_basic_pipeline_tutorial/blob/template/end/lib/elements/Source.ex) and [formats](https://github.com/membraneframework/membrane_basic_pipeline_tutorial/tree/template/end/lib/formats) should look like.


Formats

# Ordering Buffer

In this chapter we will deal with the next [element](../glossary/glossary.md#element) in our [pipeline](../glossary/glossary.md#pipeline) - the [Ordering Buffer](../glossary/glossary.md#jitter-buffer--ordering-buffer).
As stated in the [chapter about the system architecture](02_SystemArchitecture.md), this element is responsible for ordering the incoming [packets](../glossary/glossary.md#packet), based on their sequence id.
Because Ordering Buffer is a filtering element, we need to specify both the input and the output [pads](../glossary/glossary.md#pad):

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

```elixir
defmodule Basic.Elements.OrderingBuffer do
 use Membrane.Filter
 alias Basic.Formats.Packet

 def_input_pad :input,
 flow_control: :manual,
 demand_unit: :buffers,
 accepted_format: %Packet{type: :custom_packets}

 def_output_pad :output,
 flow_control: :manual,
 accepted_format: %Packet{type: :custom_packets}
 ...
end
```

Note the format specification definition there - we expect `Basic.Formats.Packet` of type `:custom_packets` to be sent on the input pad, and the same type of packets to be sent through the output pad.
In the next step let's specify how we want the state of our element to look like:

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

```elixir
defmodule Basic.Elements.OrderingBuffer do
 ...
 @impl true
 def handle_init(_context, _options) do
 {[],
 %{
 ordered_packets: [],
 last_sent_seq_id: 0
 }}
 end
 ...
end
```

If you don't remember what is the purpose of the Ordering Buffer, please refer to the [2nd chapter](02_SystemArchitecture.md).
We will need to hold a list of ordered packets, as well as a sequence id of the packet, which most recently was sent through the output pad (we need to know if there are some packets missing between the last sent packet and the first packet in our ordered list).

Handling demands is quite straightforward:

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

```elixir
defmodule Basic.Elements.OrderingBuffer do
 ...
 @impl true
 def handle_demand(:output, size, _unit, _context, state) do
 {[demand: {:input, size}], state}
 end
 ...
end
```

We simply send the `:demand` on the `:input` pad once we receive a demand on the `:output` pad. One packet on input corresponds to one packet on output so for each 1 unit of demand we send 1 unit of demand to the `:input` pad.

Now we can go to the main part of the Ordering Buffer implementation - the `handle_buffer/4` callback.
The purpose of this callback is to process the incoming buffer. It gets called once a new buffer is available and waiting to be processed.

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

```elixir
defmodule Basic.Elements.OrderingBuffer do
 ...
 @impl true
 def handle_buffer(:input, buffer, _context, state) do
 packet = unzip_packet(buffer.payload)
 ordered_packets = [packet | state.ordered_packets] |> Enum.sort()
 state = %{state | ordered_packets: ordered_packets}
 [{last_seq_id, _} | _] = ordered_packets
 ...
 end

 defp unzip_packet(packet) do
 regex = ~r/^\[seq\:(?<seq_id>\d+)\](?<data>.*)$/
 %{"data" => data, "seq_id" => seq_id} = Regex.named_captures(regex, packet)
 {String.to_integer(seq_id), %Membrane.Buffer{payload: data} }
 end
 ...
end
```

First, we are taking advantage of the [Regex module](https://hexdocs.pm/elixir/1.13/Regex.html) available in the Elixir.
With the `Regex.named_captures` we can access the values of the fields defined within the regex description.
Do you remember what our packet looks like?

```
[seq:7][frameid:2][timestamp:3]data
```

Above you can see an exemplary packet. We need to fetch the value of the sequence id (in our case it is equal to 7) and get the rest of the packet.
Therefore we have defined the regex description as:

```elixir
~r/^\[seq\:(?<seq_id>\d+)\](?<data>.*)$/
```

> ### TIP - How to read the regex?
>
> - `~r/.../` stands for the `sigil_r/1` [sigil](https://elixir-lang.org/getting-started/sigils.html)
> - `^` describes the beginning of the input
> - `\[` stands for the opening square bracket ('\[') at the beginning is required to escape the char since the plain '\[' has a special meaning in the regex syntax
> - `seq` is a sequence of 's', 'e', 'q' characters (we need to adjust our regex description to match the header of the packet)
> - `\:` stands for the ':' character (we also need to escape that character since it is meaningful in the regex's syntax)
> - `(?<seq_id>\d+)` allows us to define a named capture - later one, once we use the `Regex.named_captures/2`, we will retrieve the map with 'seq_id' key and the corresponding value equal to the string described by the `\d+` 'partial' regex (which means - one or more occurrences of a decimal). Generally speaking, a named capture can be specified with the following structure: `(?<key>regex)` where `regex` is a regex description.
> - `\]` is the escaped closing square bracket character
> - `(?<data>.*)` is a named capture description that allows us to get the value of a `.*` regex (any character no or any number of times) under a `data` key.
> - `$` stands for the end of the input

The result of `Regex.named_captures/2` applied to that regex description and the exemplary packet should be following:

```elixir
{"seq_id"=>7, "data"=>"[frameid:2][timestamp:3]data"}
```

Once we unzip the header of the packet in the `handle_buffer/4` callback, we can put the incoming packet in the `ordered_packets` list and sort that list. Due to the fact, that elements of this list are tuples, whose first element is a sequence id (a value that is unique), the list will be sorted based on the sequence id.
We also get the sequence id of the first element in the updated `ordered_packets` list.

Here comes the rest of the `handle_buffer/4` definition:

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

```elixir
defmodule Basic.Elements.OrderingBuffer do
 ...
 def handle_buffer(:input, buffer, _context, state) do
 ...
 if state.last_sent_seq_id + 1 == last_seq_id do
 {ready_packets_sequence, ordered_packets_left} =
 get_ready_packets_sequence(ordered_packets, [])

 {last_sent_seq_id, _} = List.last(ready_packets_sequence)

 state = %{
 state
 | ordered_packets: ordered_packets_left,
 last_sent_seq_id: last_sent_seq_id
 }

 ready_buffers = Enum.map(ready_packets_sequence, &elem(&1, 1))

 {[buffer: {:output, ready_buffers}], state}
 else
 {[redemand: :output], state}
 end
 end
 ...
end
```

We need to distinguish between two situations: the currently processed packet can have a sequence id which is subsequent to the sequence id of the last sent packet or there might be some packets not yet delivered to us, with sequence ids in between the last sent sequence id and the sequence id of a currently processed packet. In the second case, we should store the packet and wait for the next packets to arrive. We will accomplish that using [`redemands` mechanism](../glossary/glossary.md#redemands), which will be explained in detail in the next chapter.
However, in the first situation, we need to get the ready packet's sequence - that means, a consistent batch of packets from the `:ordered_packets`. This can be done in the following way:

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

```elixir
defmodule Basic.Elements.OrderingBuffer do
 ...
 defp get_ready_packets_sequence([first_packet | ordered_rest], []) do
 get_ready_packets_sequence(ordered_rest, [first_packet])
 end

 defp get_ready_packets_sequence(
 [next_seq = {next_id, _} | ordered_rest],
 [{last_id, _} | _] = ready_sequence
 )
 when next_id == last_id + 1 do
 get_ready_packets_sequence(ordered_rest, [next_seq | ready_sequence])
 end

 defp get_ready_packets_sequence(ordered_packets, ready_sequence) do
 {Enum.reverse(ready_sequence), ordered_packets}
 end
 ...
end
```

Note the order of the definitions, since we are taking advantage of the pattern matching mechanism!
The algorithm implemented in the snippet above is really simple - we are recursively taking the next packet out of the `:ordered_packets` buffer until it becomes empty or there is a missing packet (`next_id == last_id + 1`) between the last taken packet and the next packet in the buffer.
Once we have a consistent batch of packets, we can update the state (both the`:ordered_packets` and the `:last_sent_seq_id` need to be updated) and output the ready packets by defining the `:buffer` action.

Test the `OrderingBuffer`:

```console
mix test test/elements/ordering_buffer_test.exs
```

Now the `OrderingBuffer` element is ready. Before we implement the next element let us introduce you to the concept of redemands.

Ordering Buffer

# Redemands

Redemanding is a very convenient mechanism that helps you, the Membrane Developer, stick to the DRY (Don't Repeat Yourself) rule.
Generally speaking, it can be used in two situations:

- in the [source elements](../glossary/glossary.md#source)
- in the [filter elements](../glossary/glossary.md#filter)

To comprehensively understand the concept behind redemanding, you need to be aware of the typical control flow which occurs in Membrane's [elements](../glossary/glossary.md#element) - which you could have seen in the elements we have already defined.

## In Source elements

In the [source elements](../glossary/glossary.md#source), there is a "side-channel", from which we can receive data. That "side-channel" can be, as in the exemplary [pipeline](../glossary/glossary.md#pipeline) we are working on, in form of a file, from which we are reading the data. In real-life scenarios it could be also, i.e. an [RTP](../glossary/glossary.md#rtp) stream received via the network. Since we have that "side-channel", there is no need to receive data via the input [pad](../glossary/glossary.md#pad) (that is why we don't have it in the source element, do we?).
The whole logic of fetching the data can be put inside the `handle_demand/5` callback - once we are asked to provide the [buffers](../glossary/glossary.md#buffer), the `handle_demand/5` callback gets called and we can provide the desired number of buffers from the "side-channel", inside the body of that callback. No processing occurs here - we get asked for the buffer and we provide the buffer, simple as that.
The redemand mechanism here lets you focus on providing a single buffer in the `handle_demand/5` body - later on, you can simply return the `:redemand` action and that action will invoke the `handle_demand/5` once again, with the updated number of buffers which are expected to be provided. Let's see it in an example - we could have such a `handle_demand/5` definition (and it wouldn't be a mistake!):

```elixir
@impl true
def handle_demand(:output, size, _unit, _context, state) do
 actions = for x <- 1..size do
 payload = Input.get_next() #Input.get_next() is an exemplary function which could be providing data
 {:buffer, %Membrane.Buffer(payload: payload)}
 end
 { {:ok, actions}, state}
end
```

As you can see in the snippet above, we need to generate the required `size` of buffers in the single `handle_demand/5` run. The logic of supplying the demand there is quite easy - but what if you would also need to check if there is enough data to provide a sufficient number of buffer? You would need to check it in advance (or try to read as much data as possible before supplying the desired number of buffers). And what if an exception occurs during the generation, before supplying all the buffers?
You would need to take under the consideration all these situations and your code would become larger and larger.

Wouldn't it be better to focus on a single buffer in each `handle_demand/5` call - and let the Membrane Framework automatically update the demand's size? This can be done in the following way:

```elixir
@impl true
def handle_demand(:output, _size, unit, context, state) do
 payload = Input.get_next() #Input.get_next() is an exemplary function which could be providing data
 actions = [buffer: %Membrane.Buffer(payload: payload), redemand: :output]
 { {:ok, actions}, state}
end
```

## In Filter elements
In the filter element, the situation is quite different.
Since the filter's responsibility is to process the data sent via the input pads and transmit it through the output pads, there is no 'side-channel' from which we could take data. That is why in normal circumstances you would transmit the buffer through the output pad in the `handle_buffer/4` callback (which means - once your element receives a buffer, you process it, and then you 'mark' it as ready to be output with the `:buffer` action). When it comes to the `handle_demand/5` action on the output pad, all you need to do is to demand the appropriate number of buffers on the element's input pad. 

That behavior is easy to specify when we exactly know how many input buffers correspond to the one output buffer (recall the situation in the [Depayloader](../glossary/glossary.md#payloader-and-depayloader) of our pipeline, where we *a priori* knew, that each output buffer ([frame](../glossary/glossary.md#frame)) consists of a given number of input buffers ([packets](../glossary/glossary.md#packet))). However it becomes impossible to define if the output buffer might be a combination of a discretionary set number of input buffers. At the same time, we have dealt with an unknown number of required buffers in the OrderingBuffer implementation, where we didn't know how many input buffers do we need to demand to fulfill the missing spaces between the packets ordered in the list. How did we manage to do it? 

We simply used the `:redemand` action! In case there was a missing space between the packets, we returned the `:redemand` action, which immediately called the `handle_demand/5` callback (implemented in a way to request for a buffer on the input pad). The fact, that that callback invocation was immediate, which means - the callback was called synchronously, right after returning from the `handle_buffer/4` callback, before processing any other message from the element's mailbox - might be crucial in some situations, since it guarantees that the demand will be done before handling any other event.
Recall the situation in the [Mixer](../glossary/glossary.md#mixer), where we were producing the output buffers right in the `handle_demand/5` callback. We needed to attempt to create the output buffer after:

- updating the buffers' list in `handle_buffer/4`
- updating the status of the [track](../glossary/glossary.md#track) in `handle_end_of_stream/3`
 Therefore, we were simply returning the `:redemand` action, and the `handle_demand/5` was called sequentially after on, trying to produce the output buffer.

As you can see, redemand mechanism in filters helps us deal with situations, where we do not know how many input buffers to demand in order to be able to produce an output buffer/buffers.
In case we don't provide enough buffers in the `handle_demand/5` callback (or we are not sure that we do provide), we should call `:redemand` somewhere else (usually in the `handle_buffer/4`) to make sure that the demand is not lost.

With that knowledge let's carry on with the next element in our pipeline - `Depayloader`.

Redemands

# Depayloader
Since we have [packets](../glossary/glossary.md#packet) put in order by the [Ordering Buffer](../basic_pipeline/06_OrderingBuffer.md), we can assemble them into the original [frames](../glossary/glossary.md#frame).
The Depayloader is an element responsible for this task. Specifically speaking, it unpacks the payload from the packets -
and that is why it's called 'depayloader'.
Let's create a new module in the `lib/elements/depayloader.ex` file:

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

```elixir
defmodule Basic.Elements.Depayloader do
 use Membrane.Filter

 alias Basic.Formats.{Packet, Frame}
 ...
end
```

What input data do we expect? Of course in `Basic.Format.Packet` format!

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

```elixir
defmodule Basic.Elements.Depayloader do
 ...
 def_input_pad :input,
 flow_control: :manual,
 demand_unit: :buffers,
 accepted_format: %Packet{type: :custom_packets}
 ...
end
```

However, our element will process that input data in a way that will change the format - on output, there will be frames instead of packets!
We need to specify it while defining the `:output` pad:

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

```elixir
defmodule Basic.Elements.Depayloader do
 ...
 def_output_pad :output,
 flow_control: :manual,
 accepted_format: %Frame{encoding: :utf8}
 ...
end
```

We will also need a parameter describing how many packets should we request once we receive a demand for a frame:

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

```elixir
defmodule Basic.Elements.Depayloader do
 ...
 def_options packets_per_frame: [
 spec: pos_integer,
 description:
 "Positive integer, describing how many packets form a single frame. Used to demand the proper number of packets while assembling the frame."
 ]
 ...
end
```

In the `handle_init/2` callback we are simply saving the value of that parameter in the state of our element:

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

```elixir
defmodule Basic.Elements.Depayloader do
 ...
 @impl true
 def handle_init(_context, options) do
 {[],
 %{
 frame: [],
 packets_per_frame: options.packets_per_frame
 }}
 end
 ...
end
```

Within the state, we will also hold a (potentially not complete) `:frame` - a list of packets, which form a particular frame. We will aggregate the packets in the `:frame` until the moment the frame is complete.

As noted in the [chapter dedicated to stream formats](04_StreamFormat.md), since we are changing the type of data within the element, we cannot rely on the default implementation of the `handle_stream_format/4` callback. We need to explicitly send the updated version of the format:

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

```elixir
defmodule Basic.Elements.Depayloader do
 ...
 @impl true
 def handle_stream_format(:input, _stream_format, _context, state) do
 {[stream_format: {:output, %Frame{encoding: :utf8}}], state}
 end
 ...
end
```

As in most elements, the `handle_demand/5` implementation is quite easy - what we do is simply make a demand on our `:input` pad once we receive a demand on the `:output` pad. However, since we are expected to produce a frame (which is formed from a particular number of packets) on the `:output` pad, we need to request a particular number of packets on the `:input` pad - that is why we have defined the `:packets_per_frame` option and now we will be making use of it. In case we would have been asked to produce 10 frames, and each frame would have been made out of 5 packets, then we would need to ask for 10\*5 = 50 packets on the `:input`.

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

```elixir
defmodule Basic.Elements.Depayloader do
 ...
 @impl true
 def handle_demand(:output, size, :buffers, _context, state) do
 {[demand: {:input, size * state.packets_per_frame}], state}
 end
 ...
end
```

There is nothing left apart from processing the input data - that is - the packets. Since the packets are coming in order, we can simply hold them in the `:frame` list until all the packets forming that frame will be there. As you might remember, each packet has a frame id in its header, which can be followed by a 'b' or 'e' character, indicating the type of the packet (the one beginning a frame or the one ending the frame). We will use information about the type to find a moment in which we should produce a frame out of the packets list.

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

```elixir
defmodule Basic.Elements.Depayloader do
 ...
 @impl true
 def handle_buffer(:input, buffer, _context, state) do
 packet = buffer.payload

 regex =
 ~r/^\[frameid\:(?<frame_id>\d+(?<type>[s|e]*))\]\[timestamp\:(?<timestamp>\d+)\](?<data>.*)$/

 %{"data" => data, "frame_id" => _frame_id, "type" => type, "timestamp" => timestamp} =
 Regex.named_captures(regex, packet)

 frame = [data | state.frame]
 ...
end
```

Once again we are taking advantage of the `Regex.named_captures`.
Once we fetch the interesting values of the header's parameters, we can update the `:frame`.

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

```elixir
defmodule Basic.Elements.Depayloader do
 ...
 @impl true
 def handle_buffer(:input, buffer, _context, state) do
 ...
 if type == "e" do
 buffer = %Membrane.Buffer{
 payload: prepare_frame(frame),
 pts: String.to_integer(timestamp)
 }

 {[buffer: {:output, buffer}], %{state | frame: []}}
 else
 {[], %{state | frame: frame}}
 end
 ...
end
```

Now, depending on the type of frame, we perform different actions.
If we have the 'ending' packet, we are making the `:buffer` action with the frame made out of the packets (that's where `prepare_frame/1` function comes in handy), and clear the `:frame` buffer. Here is how the `prepare_frame/1` function can be implemented:

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

```elixir
defmodule Basic.Elements.Depayloader do
 ...
 defp prepare_frame(frame) do
 frame |> Enum.reverse() |> Enum.join("")
 end
 ...
end
```

Otherwise, if the packet is not of the 'ending' type (that is - it can be both the 'beginning' frame or some packet in the middle), we are simply updating the state with the processed packet added to the `:frame` buffer and redemanding packet.

Test the `Depayloader`:

```console
mix test test/elements/depayloader_test.exs
```

With the [`Source`](../glossary/glossary.md#source), [`OrderingBuffer`](../glossary/glossary.md#jitter-buffer--ordering-buffer) and [`Depayloader`](../glossary/glossary.md#payloader-and-depayloader) elements ready we are able to read packets from file, order them based on their sequence ID and assemble them back into frames.
In the next chapter we will be dealing with the [`Mixer`](../glossary/glossary.md#mixer) which will merge two message streams in order to create complete conversation.

Depayloader

# Mixer

Here comes the mixer - an [element](../glossary/glossary.md#element) responsible for mixing two streams of [frames](../glossary/glossary.md#frame), coming from two different sources.
Once again we start with defining the initialization options and the pads of both types:

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

```elixir
defmodule Basic.Elements.Mixer do
  @moduledoc """
  Element responsible for mixing the frames coming from two sources, basing on their timestamps.
  """
  use Membrane.Filter
  alias Basic.Formats.Frame

  def_input_pad :first_input,
    flow_control: :manual,
    demand_unit: :buffers,
    accepted_format: %Frame{encoding: :utf8}

  def_input_pad :second_input,
    flow_control: :manual,
    demand_unit: :buffers,
    accepted_format: %Frame{encoding: :utf8}

  def_output_pad :output,
    flow_control: :manual,
    accepted_format: %Frame{encoding: :utf8}
 ...
end
```

Note, that we have defined two input [pads](../glossary/glossary.md#pad): `:first_input` and the `:second_input`.
Each of these input pads will have a corresponding incoming [track](../glossary/glossary.md#track) in form of a [buffers](../glossary/glossary.md#buffer) stream. We need a structure that will hold the state of the track. Let's create it by defining a `Track` inside the mixer module:

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

```elixir
defmodule Basic.Elements.Mixer do
 ...
  defmodule Track do
   @type t :: %__MODULE__{
     buffer: Membrane.Buffer.t(),
     status: :started | :finished
   }
   defstruct buffer: nil, status: :started
  end
 ...
end
```

As you can see in the code snippet above, the `Track` will consist of the `:buffer` field, holding the very last buffer received on the corresponding input pad, and the `:status` fields, indicating the status of the track - `:started`, in case we are still expecting some buffers to come (that means - in case `:end_of_stream` event hasn't been received yet) and `:finished` otherwise. 

It's a good practice to provide a type specification for such a custom struct since it makes the code easier to reuse and lets the compiler warn us about some misspellings (for instance in the status field atoms), which cause some hard to spot errors. 

A careful reader might notice, that we are holding only one buffer for each track, instead of a list of all the potentially unprocessed buffers - does it mean that we are losing some of them? Not at all, since we are taking advantage of the elements which have appeared earlier in the [pipeline](../glossary/glossary.md#pipeline) and which provide us with an ordered list of frames on each of the inputs - however, we will need to process each buffer just at the moment it comes on the pad.

The logic we're going to implement can be described in the following three steps:

- If all the tracks are in the 'active' state ('active' means - the ones in the `:started` state or the ones in the `:finished` state but with an unprocessed buffer in the `Track` structure) - output the one with the lower timestamp. Otherwise do nothing.
- If all the tracks are in the `:finished` state and their `:buffer` is empty - send the `:end_of_stream` event.
- For all the tracks which are in the `:started` state and their buffer is empty - demand on the pad corresponding to that track.

The next step in our element implementation is quite an obvious one:

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

```elixir
defmodule Basic.Elements.Mixer do
 ...
  @impl true
  def handle_init(_context, _options) do
    {[],
     %{
       tracks: %{first_input: %Track{}, second_input: %Track{}}
     }}
  end
 ...
end
```

We have provided a `handle_init/2` callback, which does not expect any options to be passed. We are simply setting up the structure of the element state.
As mentioned previously, we will have a `Track` structure for each of the input pads. 

What's interesting is this is where the mixer having exactly two inputs stops being important. The missing functionality can be defined generically without much hassle. 

Following on the callbacks implementation, let's continue with the `handle_buffer/4` implementation:

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

```elixir
defmodule Basic.Elements.Mixer do
 ...
  @impl true
  def handle_buffer(pad, buffer, _context, state) do
    tracks = Map.update!(state.tracks, pad, &%Track{&1 | buffer: buffer})
    {tracks, buffer_actions} = get_output_buffers_actions(tracks)
    state = %{state | tracks: tracks}

    {buffer_actions ++ [redemand: :output], state}
  end
 ...
end
```

In this callback we update the mixer's state by assigning the incoming buffer to its track. We can be sure no overwriting of an existing buffer happens, which will become more apparent as we delve further into the logic's implementation. 

Once the state is updated we gather all buffers that can be sent (might be none) in `get_output_buffers_actions/1` and return the coresponding `buffer` actions. In case any demands should be sent afterwards we also tell the output pad to redemand.

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

```elixir
defmodule Basic.Elements.Mixer do
 ...
  @impl true
  def handle_end_of_stream(pad, _context, state) do
    tracks = Map.update!(state.tracks, pad, &%Track{&1 | status: :finished})
    {tracks, buffer_actions} = get_output_buffers_actions(tracks)
    state = %{state | tracks: tracks}

    if Enum.all?(tracks, fn {track_id, track} ->
         track.status == :finished and not has_buffer?({track_id, track})
       end) do
      {buffer_actions ++ [end_of_stream: :output], state}
    else
      {buffer_actions ++ [redemand: :output], state}
    end
  end
 ...
end
```

What we did here was similar to the logic defined in `handle_buffer/4` - we have just updated the state of the track (in that case - by setting its status to `:finished`), gather the buffers and send them. The important difference is that in case all inputs have closed, we should forward an `end_of_stream` action instead of a `redemand`, signaling the mixer has finished its processing. 

Let's now implement gathering ready buffers:

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

```elixir
defmodule Basic.Elements.Mixer do
 ...
  defp has_buffer?({_track_id, track}),
    do: track.buffer != nil

  defp can_send_buffer?(tracks) do
    started_tracks =
      Enum.filter(
        tracks,
        fn {_track_id, track} -> track.status != :finished end
      )

    (started_tracks == [] and Enum.any?(tracks, &has_buffer?/1)) or
      (started_tracks != [] and Enum.all?(started_tracks, &has_buffer?/1))
  end

  defp get_output_buffers_actions(tracks) do
    {buffers, tracks} = prepare_buffers(tracks)
    buffer_actions = Enum.map(buffers, fn buffer -> {:buffer, {:output, buffer}} end)
    {tracks, buffer_actions}
  end

  defp prepare_buffers(tracks) do
    if can_send_buffer?(tracks) do
      {next_track_id, next_track} =
        tracks
        |> Enum.filter(&has_buffer?/1)
        |> Enum.min_by(fn {_track_id, track} -> track.buffer.pts end)

      tracks = Map.put(tracks, next_track_id, %Track{next_track | buffer: nil})
      {buffers, tracks} = prepare_buffers(tracks)
      {[next_track.buffer | buffers], tracks}
    else
      {[], tracks}
    end
  end
end
```

The `prepare_buffers/1` function is the most involved here, so let's start with that. We first check whether we can send a buffer at all. The next buffer to send in order will of course be one with lowest `.pts`. We then empty the corresponding track's buffer. There might be more than one buffer ready to send and so we iterate the gathering recursively. 

We define `can_send_buffer?` as follows. If there's any `:started` track still waiting on a buffer we cannot send more, since whatever buffers the mixer's currently holding might come after the one that's yet to be received on this track. 

Otherwise, if all tracks have finished it can still be the case that some have non-empty buffers. We can happily send all of these in order since it is guaranteed there is no buffer preceding them that we would have to wait for.

All that's left now is to handle redemands.

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

```elixir
defmodule Basic.Elements.Mixer do
 ...
  def handle_demand(:output, _size, _unit, context, state) do
    demand_actions =
      state.tracks
      |> Enum.reject(&has_buffer?/1)
      |> Enum.filter(fn {track_id, track} ->
        track.status != :finished and context.pads[track_id].demand == 0
      end)
      |> Enum.map(fn {track_id, _track} -> {:demand, {track_id, 1}} end)

    {demand_actions, state}
  end
 ...
end
```

Since it should be responsible for producing and sending `demand` actions to the corresponding input pads, we accordingly filter tracks for ones that are empty, started, and with no demands pending. 
It should also become clearer why in `handle_buffer/4` the receiving track is sure to have an empty buffer ready to be overwritten, since we only send demands to input pads of empty tracks.

And that's all! the mixer's ready to mix, and ready to be tested:

```console
mix test test/elements/mixer_test.exs
```

Now all that's left to do is to save our stream to file using [`Sink`](../glossary/glossary.md#sink).


Mixer

# Pipeline

The time has come to assemble all the bricks together and create the pipeline!
This task is really easy since the Membrane Framework provides a sort of DSL (*Domain Specific Language*) which allows you to link the prefabricated components together.
In many real-life scenarios, this part would be the only thing you would need to do since you can take advantage of plenty of ready components (in form of [elements](../glossary/glossary.md#element) and [bins](../glossary/glossary.md#bin)) which are available as a part of the Membrane Framework. For now, we will create the pipeline out of the elements we have created during that tutorial!

## Defining the pipeline

The pipeline is another behavior introduced by the Membrane Framework. To make the module a pipeline, we need to make it `use Membrane.Pipeline`. That is how we will start our implementation of the pipeline module, in the `lib/pipeline.ex` file:

**_`lib/pipeline.ex`_**

```elixir

defmodule Basic.Pipeline do

  use Membrane.Pipeline
 ...
end
```

You could have guessed - all we need to do now is to describe the desired behavior by implementing the callbacks! In fact, the only callback we want to implement if the pipeline is the `handle_init/2` callback, called once the pipeline is initialized - of course, there are plenty of other callbacks which you might find useful while dealing with a more complex task. You can read about them [here](https://hexdocs.pm/membrane_core/Membrane.Pipeline.html#callbacks).
Please add the following callback signature to our `Basic.Pipeline` module:

**_`lib/pipeline.ex`_**

```elixir

defmodule Basic.Pipeline do
 ...
 @impl true
 def handle_init(_context, _options) do

 end
end
```

As you can see, we can initialize the pipeline with some options, but in our case, we do not need them.

## Supervising and children-parent relationship

In Elixir's actor model, derived from the Erlang programming language (as well as in many other implementations of the actor system) there is a concept of the actors supervising each other.
In case of the actor failing due to an error it is its supervisor's responsibility to deal with that fact - either by stopping that actor, restarting it, or performing some other action.
With such a concept in mind, it's possible to create reliable and fault-tolerant actor systems.
[Here](https://blog.appsignal.com/2021/08/23/using-supervisors-to-organize-your-elixir-application.html) there is a really nice article describing that concept and providing an example of the actor system. Feel free to stop here and read about the supervision mechanism in Elixir if you have never met with that concept before.
Our pipeline will also be an actor system - with the `Basic.Pipeline` module being the supervisor of all its elements.
As you have heard before - it is the supervisor's responsibility to launch its children's processes.
In the Membrane Framework's pipeline there is a special action designed for that purpose - [`:spec`](https://hexdocs.pm/membrane_core/Membrane.Pipeline.Action.html#t:spec/0).
As you can see, you need to specify the `Membrane.ChildrenSpec` for that purpose.

Please stop for a moment and read about the [`Membrane.ChildrenSpec`](https://hexdocs.pm/membrane_core/Membrane.ChildrenSpec.html).
We will wait for you and once you are ready, we will define our own children and links ;)

Let's start with defining what children we need inside the `handle_init/2` callback! If you have forgotten what structure we want to achieve please refer to the [2nd chapter](02_SystemArchitecture.md) and recall what elements we need inside of our pipeline.

**_`lib/pipeline.ex`_**

```elixir
defmodule Basic.Pipeline do
 ...
  @impl true
  def handle_init(_context, _options) 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})
      |> via_in(:first_input)
      |> child(:mixer, Basic.Elements.Mixer)
      |> child(:output, %Basic.Elements.Sink{location: "output.txt"}),
      child(:input2, %Basic.Elements.Source{location: "input.B.txt"})
      |> child(:ordering_buffer2, Basic.Elements.OrderingBuffer)
      |> child(:depayloader2, %Basic.Elements.Depayloader{packets_per_frame: 4})
      |> via_in(:second_input)
      |> get_child(:mixer)
    ]

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

Remember to pass the desired file paths in the `:location` option! 

Now... that's it! :) 
The spec list using Membrane's DSL is enough to describe our pipeline's topology. The child keywords spawn components of the specified type and we can use the `|>` operator to link them together. When the pads that should be linked are unamibiguous this is straightforward but for links like those with `Mixer` we can specify the pads using `via_in/1`. There also exists a `via_out/1` keyword which works similarly for output pads. 

As you can see the first argument to `child/2` is a component identifier. If we had omitted it Membrane would generate a unique identifier under the hood. For more about the `child` functions please refer to [the docs](https://hexdocs.pm/membrane_core/Membrane.ChildrenSpec.html#child/1).

Our pipeline is ready! Let's try to launch it.
We will do so by starting the pipeline, and then playing it. For the ease of use we will do it in a script.

**_`start.exs`_**

```elixir
{:ok, _sup, pipeline} = Membrane.Pipeline.start_link(Basic.Pipeline)
Process.monitor(pipeline)

# Wait for the pipeline to terminate
receive do
  {:DOWN, _monitor, :process, ^pipeline, _reason} -> :ok
end
```

You can execute it by running `mix run start.exs` in the terminal.

In the output file (the one specified in the `handle_init/2` callback of the pipeline) you should see the recovered conversation.

In case of any problems you can refer to the code on the `template/end` branch of `membrane_basic_pipeline_tutorial` repository.

Now our solution is completed. You have acquired a basic understanding of Membrane, and you can implement a simple pipeline using elements.

If you wish to extend your knowledge of Membrane's concepts we encourage you to read the [extension to this tutorial](../basic_pipeline_extension/01_Introduction.md).
