Update to core 0.11.0 (#13)

* Update to core 0.11.0

* Bump version to 0.5.0

README.md

@@ -15,7 +15,7 @@ The package can be installed by adding `membrane_ivf_plugin` to your list of dep
 ```elixir
 def deps do
   [
-    {:membrane_ivf_plugin, "~> 0.4.1"}
+    {:membrane_ivf_plugin, "~> 0.5.0"}
   ]
 end
 ```

lib/deserializer.ex

@@ -12,10 +12,11 @@ defmodule Membrane.Element.IVF.Deserializer do
   alias Membrane.{Buffer, RemoteStream, Time}
   alias Membrane.{VP8, VP9}
 
-  def_input_pad :input, caps: :any, demand_mode: :auto, demand_unit: :buffers
+  def_input_pad :input, accepted_format: _any, demand_mode: :auto, demand_unit: :buffers
 
   def_output_pad :output,
-    caps: {RemoteStream, content_format: one_of([VP9, VP8]), type: :packetized},
+    accepted_format:
+      %RemoteStream{content_format: format, type: :packetized} when format in [VP9, VP8],
     demand_mode: :auto
 
   defmodule State do
@@ -29,15 +30,15 @@ defmodule Membrane.Element.IVF.Deserializer do
   end
 
   @impl true
-  def handle_init(_options) do
-    {:ok, %State{}}
+  def handle_init(_ctx, _options) do
+    {[], %State{}}
   end
 
   @impl true
-  def handle_caps(_pad, _caps, _ctx, state) do
-    # ignore incoming caps, we will send our own
+  def handle_stream_format(_pad, _stream_format, _ctx, state) do
+    # ignore incoming stream_format, we will send our own
     # in handle_process
-    {:ok, state}
+    {[], state}
   end
 
   @impl true
@@ -46,21 +47,21 @@ defmodule Membrane.Element.IVF.Deserializer do
 
     with {:ok, file_header, rest} <- Headers.parse_ivf_header(state.frame_acc),
          {:ok, buffer, rest} <- get_buffer(rest, file_header.scale <|> file_header.rate) do
-      caps =
+      stream_format =
         case file_header.four_cc do
           "VP90" -> %Membrane.RemoteStream{content_format: VP9, type: :packetized}
           "VP80" -> %Membrane.RemoteStream{content_format: VP8, type: :packetized}
         end
 
-      {{:ok, caps: {:output, caps}, buffer: {:output, buffer}},
+      {[stream_format: {:output, stream_format}, buffer: {:output, buffer}],
        %State{
          frame_acc: rest,
          start_of_stream?: false,
          timebase: file_header.scale <|> file_header.rate
        }}
     else
       {:error, :too_short} ->
-        {:ok, state}
+        {[], state}
 
       {:error, reason} ->
         raise "Deserialization of IVF failed with reason: `#{inspect(reason)}`"
@@ -72,10 +73,10 @@ defmodule Membrane.Element.IVF.Deserializer do
 
     case flush_acc(state, []) do
       {:ok, buffers, state} ->
-        {{:ok, buffer: {:output, buffers}}, state}
+        {[buffer: {:output, buffers}], state}
 
       {:error, :too_short} ->
-        {:ok, state}
+        {[], state}
     end
   end
 
@@ -90,7 +91,7 @@ defmodule Membrane.Element.IVF.Deserializer do
   defp get_buffer(payload, timebase) do
     with {:ok, %FrameHeader{size_of_frame: size_of_frame, timestamp: timestamp}, rest} <-
            Headers.parse_ivf_frame_header(payload),
-         <<frame::binary-size(size_of_frame), rest::binary()>> <- rest do
+         <<frame::binary-size(size_of_frame), rest::binary>> <- rest do
       timestamp = Time.seconds(timestamp * timebase)
       {:ok, %Buffer{pts: timestamp, payload: frame}, rest}
     else

lib/membrane_element_ivf/headers.ex

@@ -74,9 +74,9 @@ defmodule Membrane.Element.IVF.Headers do
   # bytes 24-27  number of frames in file
   # bytes 28-31  unused
   @spec create_ivf_header(integer, integer, Ratio.t(), integer, any) :: binary
-  def create_ivf_header(width, height, timebase, frame_count, caps) do
+  def create_ivf_header(width, height, timebase, frame_count, stream_format) do
     codec_four_cc =
-      case caps do
+      case stream_format do
         %Membrane.RemoteStream{content_format: VP9} -> "VP90"
         %Membrane.RemoteStream{content_format: VP8} -> "VP80"
         _unknown -> "\0\0\0\0"
@@ -113,7 +113,7 @@ defmodule Membrane.Element.IVF.Headers do
   def parse_ivf_frame_header(payload) when byte_size(payload) < 12,
     do: {:error, :too_short}
 
-  def parse_ivf_frame_header(<<size_of_frame::32-little, timestamp::64-little, rest::binary()>>) do
+  def parse_ivf_frame_header(<<size_of_frame::32-little, timestamp::64-little, rest::binary>>) do
     {:ok, %FrameHeader{size_of_frame: size_of_frame, timestamp: timestamp}, rest}
   end
 
@@ -124,7 +124,7 @@ defmodule Membrane.Element.IVF.Headers do
   def parse_ivf_header(
         <<signature::binary-size(4), version::16-little, length_of_header::16-little,
           four_cc::binary-size(4), width::16-little, height::16-little, rate::32-little,
-          scale::32-little, frame_count::32-little, _unused::32, rest::binary()>>
+          scale::32-little, frame_count::32-little, _unused::32, rest::binary>>
       ) do
     if String.valid?(signature) and String.valid?(four_cc) do
       {:ok,

lib/serializer.ex

@@ -16,21 +16,22 @@ defmodule Membrane.Element.IVF.Serializer do
               frame_count: [spec: [integer], default: 0, description: "number of frames"]
 
   def_input_pad :input,
-    caps: {RemoteStream, content_format: one_of([VP9, VP8]), type: :packetized},
+    accepted_format:
+      %RemoteStream{content_format: format, type: :packetized} when format in [VP9, VP8],
     demand_unit: :buffers
 
-  def_output_pad :output, caps: :any
+  def_output_pad :output, accepted_format: _any
 
   defmodule State do
     @moduledoc false
     defstruct [:width, :height, :timebase, :first_frame, :frame_count]
   end
 
   @impl true
-  def handle_init(options) do
+  def handle_init(_ctx, options) do
     use Ratio
 
-    {:ok,
+    {[],
      %State{
        width: options.width,
        height: options.height,
@@ -42,7 +43,7 @@ defmodule Membrane.Element.IVF.Serializer do
 
   @impl true
   def handle_demand(:output, size, :buffers, _ctx, state) do
-    {{:ok, demand: {:input, size}}, state}
+    {[demand: {:input, size}], state}
   end
 
   @impl true
@@ -61,12 +62,12 @@ defmodule Membrane.Element.IVF.Serializer do
             state.height,
             state.timebase,
             state.frame_count,
-            ctx.pads.input.caps
+            ctx.pads.input.stream_format
           )
 
     ivf_buffer = (ivf_file_header || "") <> ivf_frame
 
-    {{:ok, buffer: {:output, %Buffer{buffer | payload: ivf_buffer}}, redemand: :output},
+    {[buffer: {:output, %Buffer{buffer | payload: ivf_buffer}}, redemand: :output],
      %State{state | first_frame: false}}
   end
 end

mix.exs

@@ -1,7 +1,7 @@
 defmodule Membrane.IVF.Plugin.MixProject do
   use Mix.Project
 
-  @version "0.4.1"
+  @version "0.5.0"
   @github_url "https://github.com/membraneframework/membrane_ivf_plugin"
 
   def project do
@@ -37,11 +37,11 @@ defmodule Membrane.IVF.Plugin.MixProject do
 
   defp deps do
     [
-      {:membrane_core, github: "membraneframework/membrane_core", override: true},
+      {:membrane_core, "~> 0.11.0"},
       {:ex_doc, ">= 0.0.0", only: :dev, runtime: false},
       {:dialyxir, ">= 0.0.0", only: :dev, runtime: false},
       {:credo, ">= 0.0.0", only: :dev, runtime: false},
-      {:membrane_file_plugin, "~> 0.12.0", only: :test}
+      {:membrane_file_plugin, "~> 0.13.0", only: :test}
     ]
   end
 

mix.lock

@@ -1,19 +1,19 @@
 %{
-  "bunch": {:hex, :bunch, "1.3.1", "f8fe80042f9eb474ef2801ae2c9372f9b13d11e7053265dcfc24b9d912e3750b", [:mix], [], "hexpm", "00e21b16ff9bb698b728a01a2fc4b3bf7fc0e87c4bb9c6e4a442324aa8c5e567"},
+  "bunch": {:hex, :bunch, "1.5.0", "78530e85bc3f53e1711dba654565692e2015cb6d1685e9b53bf7606b14a36c71", [:mix], [], "hexpm", "2c32f8da5d4c9e7a534c8c25aea150da696dd8624ce64f97c21ee193c12258e5"},
   "bunt": {:hex, :bunt, "0.2.1", "e2d4792f7bc0ced7583ab54922808919518d0e57ee162901a16a1b6664ef3b14", [:mix], [], "hexpm", "a330bfb4245239787b15005e66ae6845c9cd524a288f0d141c148b02603777a5"},
   "coerce": {:hex, :coerce, "1.0.1", "211c27386315dc2894ac11bc1f413a0e38505d808153367bd5c6e75a4003d096", [:mix], [], "hexpm", "b44a691700f7a1a15b4b7e2ff1fa30bebd669929ac8aa43cffe9e2f8bf051cf1"},
-  "credo": {:hex, :credo, "1.6.6", "f51f8d45db1af3b2e2f7bee3e6d3c871737bda4a91bff00c5eec276517d1a19c", [:mix], [{:bunt, "~> 0.2.0", [hex: :bunt, repo: "hexpm", optional: false]}, {:file_system, "~> 0.2.8", [hex: :file_system, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "625520ce0984ee0f9f1f198165cd46fa73c1e59a17ebc520038b8fce056a5bdc"},
+  "credo": {:hex, :credo, "1.6.7", "323f5734350fd23a456f2688b9430e7d517afb313fbd38671b8a4449798a7854", [:mix], [{:bunt, "~> 0.2.1", [hex: :bunt, repo: "hexpm", optional: false]}, {:file_system, "~> 0.2.8", [hex: :file_system, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "41e110bfb007f7eda7f897c10bf019ceab9a0b269ce79f015d54b0dcf4fc7dd3"},
   "dialyxir": {:hex, :dialyxir, "1.2.0", "58344b3e87c2e7095304c81a9ae65cb68b613e28340690dfe1a5597fd08dec37", [:mix], [{:erlex, ">= 0.2.6", [hex: :erlex, repo: "hexpm", optional: false]}], "hexpm", "61072136427a851674cab81762be4dbeae7679f85b1272b6d25c3a839aff8463"},
-  "earmark_parser": {:hex, :earmark_parser, "1.4.26", "f4291134583f373c7d8755566122908eb9662df4c4b63caa66a0eabe06569b0a", [:mix], [], "hexpm", "48d460899f8a0c52c5470676611c01f64f3337bad0b26ddab43648428d94aabc"},
+  "earmark_parser": {:hex, :earmark_parser, "1.4.29", "149d50dcb3a93d9f3d6f3ecf18c918fb5a2d3c001b5d3305c926cddfbd33355b", [:mix], [], "hexpm", "4902af1b3eb139016aed210888748db8070b8125c2342ce3dcae4f38dcc63503"},
   "erlex": {:hex, :erlex, "0.2.6", "c7987d15e899c7a2f34f5420d2a2ea0d659682c06ac607572df55a43753aa12e", [:mix], [], "hexpm", "2ed2e25711feb44d52b17d2780eabf998452f6efda104877a3881c2f8c0c0c75"},
-  "ex_doc": {:hex, :ex_doc, "0.28.5", "3e52a6d2130ce74d096859e477b97080c156d0926701c13870a4e1f752363279", [:mix], [{:earmark_parser, "~> 1.4.19", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_elixir, "~> 0.14", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1", [hex: :makeup_erlang, repo: "hexpm", optional: false]}], "hexpm", "d2c4b07133113e9aa3e9ba27efb9088ba900e9e51caa383919676afdf09ab181"},
+  "ex_doc": {:hex, :ex_doc, "0.29.0", "4a1cb903ce746aceef9c1f9ae8a6c12b742a5461e6959b9d3b24d813ffbea146", [:mix], [{:earmark_parser, "~> 1.4.19", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_elixir, "~> 0.14", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1", [hex: :makeup_erlang, repo: "hexpm", optional: false]}], "hexpm", "f096adb8bbca677d35d278223361c7792d496b3fc0d0224c9d4bc2f651af5db1"},
   "file_system": {:hex, :file_system, "0.2.10", "fb082005a9cd1711c05b5248710f8826b02d7d1784e7c3451f9c1231d4fc162d", [:mix], [], "hexpm", "41195edbfb562a593726eda3b3e8b103a309b733ad25f3d642ba49696bf715dc"},
-  "jason": {:hex, :jason, "1.3.0", "fa6b82a934feb176263ad2df0dbd91bf633d4a46ebfdffea0c8ae82953714946", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "53fc1f51255390e0ec7e50f9cb41e751c260d065dcba2bf0d08dc51a4002c2ac"},
+  "jason": {:hex, :jason, "1.4.0", "e855647bc964a44e2f67df589ccf49105ae039d4179db7f6271dfd3843dc27e6", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "79a3791085b2a0f743ca04cec0f7be26443738779d09302e01318f97bdb82121"},
   "makeup": {:hex, :makeup, "1.1.0", "6b67c8bc2882a6b6a445859952a602afc1a41c2e08379ca057c0f525366fc3ca", [:mix], [{:nimble_parsec, "~> 1.2.2 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "0a45ed501f4a8897f580eabf99a2e5234ea3e75a4373c8a52824f6e873be57a6"},
   "makeup_elixir": {:hex, :makeup_elixir, "0.16.0", "f8c570a0d33f8039513fbccaf7108c5d750f47d8defd44088371191b76492b0b", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}, {:nimble_parsec, "~> 1.2.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "28b2cbdc13960a46ae9a8858c4bebdec3c9a6d7b4b9e7f4ed1502f8159f338e7"},
   "makeup_erlang": {:hex, :makeup_erlang, "0.1.1", "3fcb7f09eb9d98dc4d208f49cc955a34218fc41ff6b84df7c75b3e6e533cc65f", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}], "hexpm", "174d0809e98a4ef0b3309256cbf97101c6ec01c4ab0b23e926a9e17df2077cbb"},
-  "membrane_core": {:git, "https://github.com/membraneframework/membrane_core.git", "33c0b4b3419426233d6898f873d7bb045025948c", []},
-  "membrane_file_plugin": {:hex, :membrane_file_plugin, "0.12.0", "eb940e7a2f2abf30e048bd0b7c2bef9c17c18aa58875b9a833c0bc7e7b1fd709", [:mix], [{:membrane_core, "~> 0.10.0", [hex: :membrane_core, repo: "hexpm", optional: false]}], "hexpm", "281b9bf9467beead3f973adce55b9844bc4206bb3f3f60f0db8320a4af4fc5ca"},
+  "membrane_core": {:hex, :membrane_core, "0.11.0", "63ae9f56834ec67680d634d8d69f71b2d46b94f4a0ec8fafcf22d8ce216b8f41", [:mix], [{:bunch, "~> 1.5", [hex: :bunch, repo: "hexpm", optional: false]}, {:qex, "~> 0.3", [hex: :qex, repo: "hexpm", optional: false]}, {:ratio, "~> 2.0", [hex: :ratio, repo: "hexpm", optional: false]}, {:telemetry, "~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "097584018fe948fa3013bfd6bcf002b3ad7cbd13f2259be4f1903f37a7aad7ab"},
+  "membrane_file_plugin": {:hex, :membrane_file_plugin, "0.13.0", "5d82476706410d372718208b48ace8bbf131f4466bca8f88ca015ebc4dbf5bad", [:mix], [{:membrane_core, "~> 0.11", [hex: :membrane_core, repo: "hexpm", optional: false]}], "hexpm", "efa63530c1a9cf7e8da6ca5b3af9598573a007c89fc7bedf02ffe4ffc1753e66"},
   "nimble_parsec": {:hex, :nimble_parsec, "1.2.3", "244836e6e3f1200c7f30cb56733fd808744eca61fd182f731eac4af635cc6d0b", [:mix], [], "hexpm", "c8d789e39b9131acf7b99291e93dae60ab48ef14a7ee9d58c6964f59efb570b0"},
   "numbers": {:hex, :numbers, "5.2.4", "f123d5bb7f6acc366f8f445e10a32bd403c8469bdbce8ce049e1f0972b607080", [:mix], [{:coerce, "~> 1.0", [hex: :coerce, repo: "hexpm", optional: false]}, {:decimal, "~> 1.9 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "eeccf5c61d5f4922198395bf87a465b6f980b8b862dd22d28198c5e6fab38582"},
   "qex": {:hex, :qex, "0.5.1", "0d82c0f008551d24fffb99d97f8299afcb8ea9cf99582b770bd004ed5af63fd6", [:mix], [], "hexpm", "935a39fdaf2445834b95951456559e9dc2063d0a055742c558a99987b38d6bab"},

test/integration_test.exs

@@ -16,29 +16,25 @@ defmodule Membrane.Element.IVF.IntegrationTest do
     use Membrane.Pipeline
 
     @impl true
-    def handle_init(options) do
-      spec = %ParentSpec{
-        children: [
-          file_source: %Membrane.File.Source{location: options.input.path},
-          deserializer: IVF.Deserializer,
-          serializer: %IVF.Serializer{
-            width: options.input.width,
-            height: options.input.height,
-            rate: 30
-          },
-          file_sink: %Membrane.File.Sink{location: options.result_file}
-        ],
-        links: [
-          link(:file_source) |> to(:deserializer) |> to(:serializer) |> to(:file_sink)
-        ]
-      }
-
-      {{:ok, spec: spec, playback: :playing}, %{}}
+    def handle_init(_ctx, options) do
+      spec = [
+        child(:serializer, %IVF.Serializer{
+          width: options.input.width,
+          height: options.input.height,
+          rate: 30
+        }),
+        child(:file_source, %Membrane.File.Source{location: options.input.path})
+        |> child(:deserializer, IVF.Deserializer)
+        |> get_child(:serializer)
+        |> child(:file_sink, %Membrane.File.Sink{location: options.result_file})
+      ]
+
+      {[spec: spec, playback: :playing], %{}}
     end
 
     @impl true
     def handle_child_notification(_notification, _child, _ctx, state) do
-      {:ok, state}
+      {[], state}
     end
   end
 
@@ -57,22 +53,21 @@ defmodule Membrane.Element.IVF.IntegrationTest do
 
     result_file = Path.join(@results_dir, result)
 
-    {:ok, pipeline} =
+    pipeline =
       [
         module: TestPipeline,
         custom_args: %{
           input: input,
           result_file: result_file
         }
       ]
-      |> Testing.Pipeline.start_link()
+      |> Testing.Pipeline.start_link_supervised!()
 
-    assert_pipeline_playback_changed(pipeline, _, :playing)
+    assert_pipeline_play(pipeline)
 
     assert_end_of_stream(pipeline, :file_sink)
 
     Testing.Pipeline.terminate(pipeline, blocking?: true)
-    assert_pipeline_playback_changed(pipeline, _, :stopped)
 
     assert File.read!(input.path) ==
              File.read!(result_file)

test/serializer_test.exs

@@ -17,32 +17,28 @@ defmodule Membrane.Element.IVF.SerializerTest do
     use Membrane.Pipeline
 
     @impl true
-    def handle_init(options) do
+    def handle_init(_ctx, options) do
       sink =
         if options.to_file?,
           do: %Membrane.File.Sink{location: options.result_file},
           else: Testing.Sink
 
-      spec = %ParentSpec{
-        children: [
-          ivf_serializer: %IVF.Serializer{width: 1080, height: 720, rate: 30},
-          source: %Testing.Source{
-            output: Testing.Source.output_from_buffers(options.buffers),
-            caps: %RemoteStream{content_format: VP9, type: :packetized}
-          },
-          sink: sink
-        ],
-        links: [
-          link(:source) |> to(:ivf_serializer) |> to(:sink)
-        ]
-      }
-
-      {{:ok, spec: spec, playback: :playing}, %{}}
+      spec = [
+        child(:source, %Testing.Source{
+          output: Testing.Source.output_from_buffers(options.buffers),
+          stream_format: %RemoteStream{content_format: VP9, type: :packetized}
+        }),
+        get_child(:source)
+        |> child(:ivf_serializer, %IVF.Serializer{width: 1080, height: 720, rate: 30})
+        |> child(:sink, sink)
+      ]
+
+      {[spec: spec, playback: :playing], %{}}
     end
 
     @impl true
     def handle_child_notification(_notification, _child, _ctx, state) do
-      {:ok, state}
+      {[], state}
     end
   end
 
@@ -56,25 +52,25 @@ defmodule Membrane.Element.IVF.SerializerTest do
   """
   test "appends headers correctly" do
     buffer_1 = %Buffer{payload: @frame, pts: 0}
-    buffer_2 = %Buffer{payload: @frame, pts: Membrane.Time.seconds(1 <|> 30)}
+    buffer_2 = %Buffer{payload: @frame, pts: Ratio.new(Membrane.Time.second(), 30)}
 
-    {:ok, pipeline} =
+    pipeline =
       [
         module: TestPipeline,
         custom_args: %{
           to_file?: false,
           buffers: [buffer_1, buffer_2]
         }
       ]
-      |> Testing.Pipeline.start_link()
+      |> Testing.Pipeline.start_supervised!()
 
-    assert_pipeline_playback_changed(pipeline, _, :playing)
+    assert_pipeline_play(pipeline)
 
     assert_start_of_stream(pipeline, :sink)
 
     assert_sink_buffer(pipeline, :sink, ivf_buffer)
 
-    <<file_header::binary-size(32), frame_header::binary-size(12), _frame::binary()>> =
+    <<file_header::binary-size(32), frame_header::binary-size(12), _frame::binary>> =
       ivf_buffer.payload
 
     <<signature::binary-size(4), version::binary-size(2), length_of_header::binary-size(2),
@@ -98,7 +94,7 @@ defmodule Membrane.Element.IVF.SerializerTest do
 
     assert_sink_buffer(pipeline, :sink, ivf_buffer)
 
-    <<frame_header::binary-size(12), _frame::binary()>> = ivf_buffer.payload
+    <<frame_header::binary-size(12), _frame::binary>> = ivf_buffer.payload
     <<size_of_frame::binary-size(4), timestamp::binary-size(8)>> = frame_header
 
     assert size_of_frame == <<byte_size(@frame)::32-little>>
@@ -117,7 +113,7 @@ defmodule Membrane.Element.IVF.SerializerTest do
   test "serialize real vp9 buffers" do
     buffers = File.read!(@fixtures_dir <> "capture.dump") |> :erlang.binary_to_term()
 
-    {:ok, pipeline} =
+    pipeline =
       [
         module: TestPipeline,
         custom_args: %{
@@ -126,9 +122,9 @@ defmodule Membrane.Element.IVF.SerializerTest do
           buffers: buffers
         }
       ]
-      |> Testing.Pipeline.start_link()
+      |> Testing.Pipeline.start_supervised!()
 
-    assert_pipeline_playback_changed(pipeline, _, :playing)
+    assert_pipeline_play(pipeline)
 
     assert_start_of_stream(pipeline, :sink)