Implement live queueing strategy and callbacks system (#111)

* Add offline queue element. Change `new-compositor` scene to old API. Remove scene related modules

* Add updating stream format, scene and sending buffer test. Fix queue element

* Fix recursive frames poping

* Fix droping pads

* Add queue tests, fix and refactor events poping, actions, pads removing and logic separation

* Add update scene events handling

* Add update scene events handling test

* Add docs

* Add `CompositorCoreFormat` docs

* Renamed queue output from `:compositor_core` to `:output`

* Fix EOS action

* Fix readability

* Add module for queue contracts

* Fix `calculate_next_buffer_pts` readability

* Fix EOS handling add EOS test

* Fix lint

* Adjust core VC to the new input format from the queue.

* Change setting video. Update `VC Core`, `VC bin`, add `Offline Queue bin`, update tests

* Naming fixes

* Fix scene notification event and rust decoding

* Fix tests

* Fix dirty scheduling

* Add docs

* Use `Map.new` instead of Map -> Enum -> Map conversions

* Remove `wgpu_test` - covered in E2E tests

* Fix `pad_ref` offline queue pad refs handling

* Fix `handle_pad_removed`

* Fix plural variables naming

* Move queue spawing logic to seperate module

* Fix queue bin spawning

* Fix custom queue strategies states handling

* Moved putting events to state

* Remove unused alias

* Wrap scene update in `SceneChangeEvent`

* Remove EOS handling from Rust

* Add core `handle_process` scene presence check

* Remove `current` prefixes from `state` fields

* Remove `different_video_indexes` check. Scene doesn't have to contain all input pads refs

* Add scene validation

* Fix scene validation. The scene doesn't have to specify all input pads

* Add scene docs

* Fix scene validation

* Fix child removing

* Change VC modules structure

* Add empty video_configs check

* Change scene validation in `handle_process` to more descriptive

* Fix setting videos in VC core

* Fix scene validation

* Implement live queueing strategy

* Fix lint

* Move creating actions to State module

* Design VC callbacks

* Fix `handle_init` ctx typespec

* Add init metadata. Refine docs

* Fix lint

* Add matadata to VC `init_options`

* Move updating next buffer pts into state. Add live queue :start_timer message notification

* Add start timer check

* Fix naming ambiguity. Add `handle_info` callback

* Implement callbacks system. Moved pipeline to test. TODO: refactor queues, rewrite tests

* Move queues common logic into state

* Moved `transformations test` handler into a separate module

* Rewrite pipeline integration test

* Create common test handler

* Remove `pad_added` event. Fix eos pads dropping

* Add custom messages handling to offline queue

* Add custom msg handling to live queue

* Separate handler state

* Refactor init options passing

* Fix user messages handling

* Change demand mode of live queue to auto

* Fix live queue output flow control

* Fix handle_tick nearest index calculation

* Fix `:no_frame` filtering in live queue

* Fix timer_started check in live queue

* Fix timer_started? check

* Add eos handling to live queue

* Fixed eos pads dropping

* Fix live queue eos

* Fix pads_states dropping, add sending black frame on empty live queue

* Refine docs

* Refine queueing strategies docs

* Add live queue tests

* Add `sends stream format and scene only once` test

* Add `vc_init_options` description

* Fix test and typos

* Restructure modules

* Fix structure. Fix tests. Fix typos

lib/membrane/video_compositor/compositor_core_format.ex

@@ -1,7 +1,6 @@
 defmodule Membrane.VideoCompositor.CompositorCoreFormat do
-  @moduledoc """
-  Describes CoreVC input format.
-  """
+  @moduledoc false
+  # Describes CoreVC input format.
 
   alias Membrane.{Pad, RawVideo}
 

lib/membrane/video_compositor/core.ex

@@ -1,7 +1,6 @@
 defmodule Membrane.VideoCompositor.Core do
-  @moduledoc """
-  The element responsible for composing frames.
-  """
+  @moduledoc false
+  # The element responsible for composing frames.
 
   use Membrane.Filter
 
@@ -10,9 +9,8 @@ defmodule Membrane.VideoCompositor.Core do
   alias Membrane.VideoCompositor.{CompositorCoreFormat, Scene, SceneChangeEvent, WgpuAdapter}
 
   defmodule State do
-    @moduledoc """
-    The internal state of the compositor
-    """
+    @moduledoc false
+    # The internal state of the compositor
 
     @enforce_keys [:wgpu_state, :output_stream_format]
     defstruct @enforce_keys ++
@@ -97,7 +95,8 @@ defmodule Membrane.VideoCompositor.Core do
           wgpu_state: wgpu_state,
           scene: scene,
           input_stream_format: stream_format,
-          update_videos?: update_videos?
+          update_videos?: update_videos?,
+          output_stream_format: output_stream_format
         }
       ) do
     if update_videos? do
@@ -110,11 +109,14 @@ defmodule Membrane.VideoCompositor.Core do
     end
 
     {:ok, rendered_frame} =
-      payload
-      |> Map.to_list()
-      |> Enum.filter(fn {pad, _frame} -> Map.has_key?(scene.video_configs, pad) end)
-      |> Enum.map(fn {pad, frame} -> {Map.get(pads_to_ids, pad), frame, pts} end)
-      |> then(fn pads_frames -> send_pads_frames(wgpu_state, pads_frames) end)
+      if payload == %{} do
+        {:ok, get_blank_frame(output_stream_format)}
+      else
+        payload
+        |> Map.to_list()
+        |> Enum.map(fn {pad, frame} -> {Map.fetch!(pads_to_ids, pad), frame, pts} end)
+        |> then(fn pads_frames -> send_pads_frames(wgpu_state, pads_frames) end)
+      end
 
     output_buffer = %Buffer{pts: pts, dts: pts, payload: rendered_frame}
 
@@ -154,4 +156,13 @@ defmodule Membrane.VideoCompositor.Core do
         raise "Core should render frame only on last buffer"
     end
   end
+
+  @spec get_blank_frame(RawVideo.t()) :: binary()
+  defp get_blank_frame(%RawVideo{width: width, height: height}) do
+    # In YUV 420 pixel there is one full resolution plane (the luma component) and two
+    # 4x downsampled planes (chroma components). Therefore:
+    # output plane pixel count = width * height * (1 + 1/4 + 1/4) = 3/2 * width * height
+    pixels = div(width * height * 3, 2)
+    <<0::size(pixels)>>
+  end
 end

lib/membrane/video_compositor/handler.ex

@@ -11,6 +11,11 @@ defmodule Membrane.VideoCompositor.Handler do
   alias Membrane.VideoCompositor.Handler.InputProperties
   alias Membrane.VideoCompositor.Scene
 
+  @typedoc """
+  Module implementing `#{inspect(__MODULE__)}` behaviour.
+  """
+  @type t :: module()
+
   @typedoc """
   Type of user-managed inner state of the handler.
   """