Add test wip

Author: FelonEkonom

lib/membrane/forwarding_filter.ex

@@ -13,7 +13,8 @@ defmodule Membrane.ForwardingFilter do
 
   def_options notify_on_stream_format?: [default: false], notify_on_event?: [default: false]
 
-  defguardp flowing?(ctx, state) when ctx.playback == :playing and state.input != nil and state.output != nil
+  defguardp flowing?(ctx, state)
+            when ctx.playback == :playing and state.input != nil and state.output != nil
 
   @impl true
   def handle_init(_ctx, opts) do
@@ -39,13 +40,15 @@ defmodule Membrane.ForwardingFilter do
     @impl true
     def unquote(callback)(pad, item, ctx, state) when flowing?(ctx, state) do
       actions = [{unquote(action), {opposite_pad(pad, state), item}}]
+      actions = actions ++ maybe_notify_parent(unquote(action), pad, item, state)
       {actions, state}
     end
 
     @impl true
     def unquote(callback)(pad, item, _ctx, state) do
       state = unquote(action) |> store_in_queue(pad, item, state)
-      {[], state}
+      actions = unquote(action) |> maybe_notify_parent(pad, item, state)
+      {actions, state}
     end
   end)
 
@@ -81,4 +84,12 @@ defmodule Membrane.ForwardingFilter do
 
   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}]
+
+  defp maybe_notify_parent(:stream_format, pad, format, %{notify_on_stream_format?: true}),
+    do: [notiy_parent: {:stream_format, pad, format}]
+
+  defp maybe_notify_parent(_type, _pad, _item, _state), do: []
 end

test/membrane/integration/forwarding_filter_test.exs

@@ -0,0 +1,79 @@
+defmodule Membrane.Integration.ForwardingFilterTest do
+  use ExUnit.Case, async: true
+
+  import Membrane.ChildrenSpec
+  import Membrane.Testing.Assertions
+
+  alias Membrane.Buffer
+  alias Membrane.ForwardingFilter
+  alias Membrane.Testing
+
+  defmodule Format 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),
+      do: {[{action, {:output, item}}], state}
+  end
+
+  test "ForwardingFilter buffers data until output pad is linked" do
+    pipeline =
+      Testing.Pipeline.start_link_supervised!(
+        spec:
+          child(:source, Source)
+          |> child(:filter, %ForwardingFilter{
+            notify_on_event?: true,
+            notify_on_stream_format?: true
+          })
+      )
+
+    data = generate_data(100, [:stream_format, :buffer, :event])
+
+    data
+    |> Enum.each(fn {type, item} ->
+      Testing.Pipeline.notify_child(pipeline, :source, {type, item})
+      assert_pipeline_notified(pipeline, :filter, {^type, :input, ^item})
+    end)
+
+    Testing.Pipeline.execute_actions(
+      spec:
+        get_child(:filter)
+        |> child(:sink, Testing.Sink)
+    )
+
+    data
+    |> Enum.each(fn {type, item} ->
+      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)
+  end
+
+  defp generate_data(number, types, pts_offset \\ 0) do
+    data =
+      1..(number - 1)
+      |> Enum.map(fn i ->
+        case Enum.random(types) do
+          :stream_format -> {:stream_format, %Format{field: i}}
+          :event -> {:event, %Event{field: i}}
+          :buffer -> {:buffer, %Buffer{pts: i + pts_offset, payload: <<>>}}
+        end
+      end)
+
+    [stream_format: %Format{field: 0}] ++ data
+  end
+end