Execute handle SoS before handling any stream format or event

membraneframework · Feb 6, 2024 · 315772c · 315772c
1 parent 3ea52d2
commit 315772c
Show file tree

Hide file tree

Showing 4 changed files with 24 additions and 5 deletions.
diff --git a/lib/membrane/core/element/event_controller.ex b/lib/membrane/core/element/event_controller.ex
@@ -39,6 +39,11 @@ 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?,
+          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,

diff --git a/lib/membrane/core/element/stream_format_controller.ex b/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

diff --git a/test/membrane/core/element/event_controller_test.exs b/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

diff --git a/test/membrane/core/element/stream_format_controller_test.exs b/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