Execute handle SoS before handling any stream format or event

CHANGELOG.md

@@ -6,6 +6,7 @@
  * Log messages in the default handle_info implementation [#680](https://github.com/membraneframework/membrane_core/pull/680)
  * Fix typespecs in Membrane.UtilitySupervisor [#681](https://github.com/membraneframework/membrane_core/pull/681)
  * Improve callback return types and group actions types [#702](https://github.com/membraneframework/membrane_core/pull/702) 
+ * Ensure `c:Membrane.Element.WithInputPads.handle_start_of_stream/3` is always executed before any `c:Membrane.Element.Base.handle_event/4` or `c:Membrane.Element.WithInputPads.handle_stream_format/4`. [#743](https://github.com/membraneframework/membrane_core/pull/743)
 
 ## 1.0.0
  * Introduce `:remove_link` action in pipelines and bins.

lib/membrane/core/element/event_controller.ex

@@ -39,6 +39,12 @@ defmodule Membrane.Core.Element.EventController do
           playback: %State{playback: :playing} <- state do
       Telemetry.report_metric(:event, 1, inspect(pad_ref))
 
+      state =
+        if event in [%Events.StartOfStream{}, %Events.EndOfStream{}] or data.start_of_stream? or
+             data.direction == :output,
+           do: state,
+           else: handle_start_of_stream(pad_ref, state)
+
       if not Event.async?(event) and buffers_before_event_present?(data) do
         PadModel.update_data!(
           state,

lib/membrane/core/element/stream_format_controller.ex

@@ -8,7 +8,15 @@ defmodule Membrane.Core.Element.StreamFormatController do
   alias Membrane.{Pad, StreamFormat}
   alias Membrane.Core.{CallbackHandler, Telemetry}
   alias Membrane.Core.Child.PadModel
-  alias Membrane.Core.Element.{ActionHandler, CallbackContext, InputQueue, PlaybackQueue, State}
+
+  alias Membrane.Core.Element.{
+    ActionHandler,
+    CallbackContext,
+    EventController,
+    InputQueue,
+    PlaybackQueue,
+    State
+  }
 
   require Membrane.Core.Child.PadModel
   require Membrane.Core.Telemetry
@@ -26,6 +34,11 @@ defmodule Membrane.Core.Element.StreamFormatController do
       %{direction: :input} = data
       Telemetry.report_metric(:stream_format, 1, inspect(pad_ref))
 
+      state =
+        if data.start_of_stream?,
+          do: state,
+          else: EventController.handle_start_of_stream(pad_ref, state)
+
       queue = data.input_queue
 
       if queue && not InputQueue.empty?(queue) do

test/membrane/core/element/event_controller_test.exs

@@ -43,13 +43,13 @@ defmodule Membrane.Core.Element.EventControllerTest do
       })
 
     state =
-      struct(State,
+      struct!(State,
         module: MockEventHandlingElement,
         name: :test_name,
         type: :filter,
         playback: :playing,
         parent_pid: self(),
-        synchronization: %{clock: nil, parent_clock: nil, stream_sync: nil},
+        synchronization: %{clock: nil, parent_clock: nil, stream_sync: Membrane.Sync.no_sync()},
         handling_action?: false,
         pads_to_snapshot: MapSet.new(),
         pads_data: %{
@@ -59,7 +59,7 @@ defmodule Membrane.Core.Element.EventControllerTest do
               direction: :input,
               pid: self(),
               flow_control: :manual,
-              start_of_stream?: false,
+              start_of_stream?: true,
               end_of_stream?: false,
               input_queue: input_queue,
               demand: 0

test/membrane/core/element/stream_format_controller_test.exs

@@ -41,7 +41,7 @@ defmodule Membrane.Core.Element.StreamFormatControllerTest do
         parent: self(),
         type: :filter,
         playback: :playing,
-        synchronization: %{clock: nil, parent_clock: nil},
+        synchronization: %{clock: nil, parent_clock: nil, stream_sync: Membrane.Sync.no_sync()},
         handling_action?: false,
         pads_to_snapshot: MapSet.new(),
         pads_data: %{
@@ -50,6 +50,7 @@ defmodule Membrane.Core.Element.StreamFormatControllerTest do
               direction: :input,
               name: :input,
               pid: self(),
+              start_of_stream?: true,
               flow_control: :manual,
               input_queue: input_queue,
               demand: 0