Merge branch 'fix-connector-bug' into crash-groups-guide

Author: FelonEkonom

CHANGELOG.md

@@ -1,6 +1,7 @@
 # Changelog
 
 ## Latest
+* Handle removed pads properly in `Membrane.Connector` [#1075](https://github.com/membraneframework/membrane_core/pull/1075)/
 * Improve remove_link action docs
 * Deprecate `:components` option for `:unsafely_name_processes_for_observer`
 * Deprecate `:links` option for `:unsafely_name_processes_for_observer` in favour of `:report_links_to_observer` configuration entry

lib/membrane/connector.ex

@@ -60,6 +60,12 @@ defmodule Membrane.Connector do
     {maybe_resume ++ maybe_flush, state}
   end
 
+  @impl true
+  def handle_pad_removed(Pad.ref(direction, _ref), ctx, state) do
+    state = state |> Map.put(direction, nil)
+    maybe_pause_auto_demand(ctx, state)
+  end
+
   @impl true
   def handle_start_of_stream(_input_pad, ctx, state), do: maybe_pause_auto_demand(ctx, state)
 

mix.exs

@@ -1,7 +1,7 @@
 defmodule Membrane.Mixfile do
   use Mix.Project
 
-  @version "1.2.6"
+  @version "1.2.7"
   @source_ref "v#{@version}"
   @hex_packages_path "scripts/elixir/hex_packages.exs"
 

test/membrane/integration/connector_test.exs

@@ -123,4 +123,41 @@ defmodule Membrane.Integration.ConnectorTest do
 
     [stream_format: %Format{field: 0}] ++ data
   end
+
+  test "Membrane.Connector doesn't raise after removing its pad" do
+    crash_group_spec = {
+      for i <- 2..5 do
+        child({:connector, i}, Membrane.Connector)
+      end,
+      group: :my_group, crash_group_mode: :temporary
+    }
+
+    children_beyond_crash_group = [
+      child(:source, Testing.Source)
+      |> child({:connector, 1}, Membrane.Connector),
+      child({:connector, 6}, Membrane.Connector)
+      |> child(:sink, Testing.Sink)
+    ]
+
+    connector_links =
+      for i <- 1..5 do
+        get_child({:connector, i})
+        |> get_child({:connector, i + 1})
+      end
+
+    spec = [crash_group_spec, children_beyond_crash_group, connector_links]
+    pipeline = Testing.Pipeline.start_link_supervised!(spec: spec)
+
+    Process.sleep(200)
+
+    {:ok, connector_pid} = Testing.Pipeline.get_child_pid(pipeline, {:connector, 3})
+    Process.exit(connector_pid, :kill)
+
+    Process.sleep(200)
+
+    assert Process.alive?(connector_pid) == false
+    assert Process.alive?(pipeline)
+
+    Testing.Pipeline.terminate(pipeline)
+  end
 end