Fix forwarding filter test

Author: FelonEkonom

lib/membrane/forwarding_filter.ex

@@ -11,7 +11,26 @@ defmodule Membrane.ForwardingFilter do
     availability: :on_request,
     max_instances: 1
 
-  def_options notify_on_stream_format?: [default: false], notify_on_event?: [default: false]
+  def_options notify_on_stream_format?: [
+                description: """
+                If `true`, #{inspect(__MODULE__)} will send `{:stream_format, pad_ref, stream_format}` \
+                notification to parent on every stream format arrival.
+
+                Defaults to `false`.
+                """,
+                spec: boolean(),
+                default: false
+              ],
+              notify_on_event?: [
+                description: """
+                If `true`, #{inspect(__MODULE__)} will send `{:event, pad_ref, event}` \
+                notification to parent on every event arrival.
+
+                Defaults to `false`.
+                """,
+                spec: boolean(),
+                default: false
+              ]
 
   defguardp flowing?(ctx, state)
             when ctx.playback == :playing and state.input != nil and state.output != nil
@@ -66,11 +85,7 @@ defmodule Membrane.ForwardingFilter do
     state |> Map.update!(:queue, &[{action, pad, item} | &1])
   end
 
-  defp flush_queue_if_flowing(ctx, state) do
-    if flowing?(ctx, state), do: flush_queue(ctx, state), else: {[], state}
-  end
-
-  defp flush_queue(ctx, state) when flowing?(ctx, state) do
+  defp flush_queue_if_flowing(ctx, state) when flowing?(ctx, state) do
     actions =
       state.queue
       |> Enum.reverse()
@@ -82,14 +97,16 @@ defmodule Membrane.ForwardingFilter do
     {actions, %{state | queue: nil}}
   end
 
+  defp flush_queue_if_flowing(_ctx, state), do: {[], state}
+
   defp opposite_pad(Pad.ref(:input, _ref), state), do: state.output
   defp opposite_pad(Pad.ref(:output, _ref), state), do: state.input
 
   defp maybe_notify_parent(:event, pad, event, %{notify_on_event?: true}),
-    do: [notiy_parent: {:event, pad, event}]
+    do: [notify_parent: {:event, pad, event}]
 
   defp maybe_notify_parent(:stream_format, pad, format, %{notify_on_stream_format?: true}),
-    do: [notiy_parent: {:stream_format, pad, format}]
+    do: [notify_parent: {:stream_format, pad, format}]
 
   defp maybe_notify_parent(_type, _pad, _item, _state), do: []
 end

test/membrane/integration/forwarding_filter_test.exs

@@ -4,6 +4,8 @@ defmodule Membrane.Integration.ForwardingFilterTest do
   import Membrane.ChildrenSpec
   import Membrane.Testing.Assertions
 
+  require Membrane.Pad, as: Pad
+
   alias Membrane.Buffer
   alias Membrane.ForwardingFilter
   alias Membrane.Testing
@@ -12,21 +14,16 @@ defmodule Membrane.Integration.ForwardingFilterTest do
     defstruct [:field]
   end
 
-  defmodule Event do
-    @derive Membrane.EventProtocol
-    defstruct [:field]
-  end
-
   defmodule Source do
     use Membrane.Source
     def_output_pad :output, accepted_format: _any, flow_control: :push
 
     @impl true
-    def handle_parent_notifications({action, item}, _ctx, state),
+    def handle_parent_notification({action, item}, _ctx, state),
       do: {[{action, {:output, item}}], state}
   end
 
-  test "ForwardingFilter buffers data until output pad is linked" do
+  test "Membrane.ForwardingFilter buffers data until output pad is linked" do
     pipeline =
       Testing.Pipeline.start_link_supervised!(
         spec:
@@ -42,25 +39,40 @@ defmodule Membrane.Integration.ForwardingFilterTest do
     data
     |> Enum.each(fn {type, item} ->
       Testing.Pipeline.notify_child(pipeline, :source, {type, item})
-      assert_pipeline_notified(pipeline, :filter, {^type, :input, ^item})
+
+      if type in [:stream_format, :event] do
+        assert_pipeline_notified(pipeline, :filter, {^type, Pad.ref(:input, _id), ^item})
+      end
+    end)
+
+    spec = get_child(:filter) |> child(:sink, Testing.Sink)
+    Testing.Pipeline.execute_actions(pipeline, spec: spec)
+
+    data
+    |> Enum.each(fn
+      {:buffer, item} -> assert_sink_buffer(pipeline, :sink, ^item)
+      {:event, item} -> assert_sink_event(pipeline, :sink, ^item)
+      {:stream_format, item} -> assert_sink_stream_format(pipeline, :sink, ^item)
     end)
 
-    Testing.Pipeline.execute_actions(
-      spec:
-        get_child(:filter)
-        |> child(:sink, Testing.Sink)
-    )
+    data = generate_data(100, [:stream_format, :buffer, :event], 200)
 
     data
     |> Enum.each(fn {type, item} ->
+      Testing.Pipeline.notify_child(pipeline, :source, {type, item})
+
+      if type in [:stream_format, :event] do
+        assert_pipeline_notified(pipeline, :filter, {^type, Pad.ref(:input, _id), ^item})
+      end
+
       case type do
         :buffer -> assert_sink_buffer(pipeline, :sink, ^item)
         :event -> assert_sink_event(pipeline, :sink, ^item)
         :stream_format -> assert_sink_stream_format(pipeline, :sink, ^item)
       end
     end)
 
-    data = generate_data(100, [:stream_format, :buffer, :event], 200)
+    Testing.Pipeline.terminate(pipeline)
   end
 
   defp generate_data(number, types, pts_offset \\ 0) do
@@ -69,7 +81,7 @@ defmodule Membrane.Integration.ForwardingFilterTest do
       |> Enum.map(fn i ->
         case Enum.random(types) do
           :stream_format -> {:stream_format, %Format{field: i}}
-          :event -> {:event, %Event{field: i}}
+          :event -> {:event, %Testing.Event{}}
           :buffer -> {:buffer, %Buffer{pts: i + pts_offset, payload: <<>>}}
         end
       end)