defmodule Membrane.Core.Pipeline do @moduledoc false use GenServer alias __MODULE__.{ActionHandler, State} alias Membrane.{Clock, ResourceGuard} alias Membrane.Core.{CallbackHandler, ProcessHelper, Stalker, SubprocessSupervisor} alias Membrane.Core.Parent.{ChildLifeController, LifecycleController} alias Membrane.Core.Pipeline.CallbackContext alias Membrane.Core.TimerController require Membrane.Core.Utils, as: Utils require Membrane.Core.Message, as: Message require Membrane.Core.Component require Membrane.Core.LegacyTelemetry, as: LegacyTelemetry @spec get_stalker(pipeline :: pid()) :: Membrane.Core.Stalker.t() def get_stalker(pipeline) do case Message.call(pipeline, :get_stalker) do {:ok, stalker} -> stalker {:error, _reason} -> raise Membrane.PipelineError, "Pipeline #{inspect(pipeline)} not found" end end @impl GenServer def init(params) do Utils.log_on_error do do_init(params) end end defp do_init(params) do observability_config = %{ name: params.name || "Pipeline#{:erlang.pid_to_list(self())}", component_type: :pipeline } %{subprocess_supervisor: subprocess_supervisor} = params SubprocessSupervisor.set_parent_component(subprocess_supervisor, observability_config) stalker = Stalker.new(observability_config, subprocess_supervisor) {:ok, resource_guard} = SubprocessSupervisor.start_utility(subprocess_supervisor, {ResourceGuard, self()}) LegacyTelemetry.report_init(:pipeline) ResourceGuard.register(resource_guard, fn -> LegacyTelemetry.report_terminate(:pipeline) end) {:ok, clock_proxy} = SubprocessSupervisor.start_utility( params.subprocess_supervisor, {Clock, proxy: true} ) state = %State{ module: params.module, synchronization: %{ clock_proxy: clock_proxy, clock_provider: %{clock: nil, provider: nil}, timers: %{} }, subprocess_supervisor: subprocess_supervisor, resource_guard: resource_guard, stalker: stalker } state = CallbackHandler.exec_and_handle_callback( :handle_init, ActionHandler, %{context: &CallbackContext.from_state/1}, [], %{state | internal_state: params.options} ) {:ok, state, {:continue, :setup}} end @impl GenServer def handle_continue(:setup, state) do Utils.log_on_error do state = LifecycleController.handle_setup(state) {:noreply, state} end end @impl GenServer def handle_info(msg, state) do Utils.log_on_error do do_handle_info(msg, state) end end defp do_handle_info( Message.new(:stream_management_event, [element_name, pad_ref, event, event_params]), state ) do state = LifecycleController.handle_stream_management_event( event, element_name, pad_ref, event_params, state ) {:noreply, state} end defp do_handle_info(Message.new(:child_pad_removed, [child, pad]), state) do state = ChildLifeController.handle_child_pad_removed(child, pad, state) {:noreply, state} end defp do_handle_info(Message.new(:child_notification, [from, notification]), state) do state = LifecycleController.handle_child_notification(from, notification, state) {:noreply, state} end defp do_handle_info(Message.new(:timer_tick, timer_id), state) do state = TimerController.handle_tick(timer_id, state) {:noreply, state} end defp do_handle_info(Message.new(:link_response, [link_id, direction]), state) do state = ChildLifeController.handle_link_response(link_id, direction, state) {:noreply, state} end defp do_handle_info(Message.new(:initialized, child), state) do state = ChildLifeController.handle_child_initialized(child, state) {:noreply, state} end defp do_handle_info(Message.new(:child_death, [name, reason]), state) do case ChildLifeController.handle_child_death(name, reason, state) do {:stop, reason, _state} -> ProcessHelper.notoelo(reason) {:continue, state} -> {:noreply, state} end end defp do_handle_info(Message.new(:terminate), state) do state = LifecycleController.handle_terminate_request(state) {:noreply, state} end defp do_handle_info(Message.new(_type, _args, _opts) = message, _state) do raise Membrane.PipelineError, "Received invalid message #{inspect(message)}" end defp do_handle_info({:membrane_clock_ratio, clock, ratio}, state) do state = TimerController.handle_clock_update(clock, ratio, state) {:noreply, state} end defp do_handle_info(message, state) do state = LifecycleController.handle_info(message, state) {:noreply, state} end @impl GenServer def handle_call(msg, from, state) do Utils.log_on_error do do_handle_call(msg, from, state) end end defp do_handle_call(Message.new(:get_stalker), _from, state) do {:reply, {:ok, state.stalker}, state} end defp do_handle_call(Message.new(:get_child_pid, child_name), _from, state) do reply = with %State{children: %{^child_name => %{pid: child_pid}}} <- state do {:ok, child_pid} else _other -> {:error, :child_not_found} end {:reply, reply, state} end defp do_handle_call(message, from, state) do context = &CallbackContext.from_state(&1, from: from) state = CallbackHandler.exec_and_handle_callback( :handle_call, Membrane.Core.Pipeline.ActionHandler, %{context: context}, [message], state ) {:noreply, state} end end