defmodule Membrane.Core.Bin do @moduledoc false use Bunch use GenServer alias __MODULE__.{ActionHandler, PadController, State} alias Membrane.Core.Bin.CallbackContext alias Membrane.Core.{ CallbackHandler, Child, Message, Parent, ProcessHelper, SubprocessSupervisor, TimerController } alias Membrane.ResourceGuard require Membrane.Core.Utils, as: Utils require Membrane.Core.Message require Membrane.Core.LegacyTelemetry, as: LegacyTelemetry require Membrane.Logger @type options :: %{ name: atom, module: module, node: node | nil, parent: pid, user_options: Membrane.Bin.options(), parent_clock: Membrane.Clock.t(), parent_path: Membrane.ComponentPath.path(), log_metadata: Logger.metadata(), sync: :membrane_no_sync, subprocess_supervisor: pid(), parent_supervisor: pid() } @doc """ Starts the Bin based on given module and links it to the current process. Bin options are passed to module's `c:Membrane.Bin.handle_init/2` callback. Process options are internally passed to `GenServer.start_link/3`. Returns the same values as `GenServer.start_link/3`. """ @spec start_link(options) :: GenServer.on_start() def start_link(options), do: do_start(:start_link, options) @doc """ Works similarly to `start_link/3`, but does not link to the current process. """ @spec start(options()) :: GenServer.on_start() def start(options), do: do_start(:start, options) defp do_start(method, options) do %{module: module, name: name, node: node, user_options: user_options} = options Membrane.Logger.debug(""" Bin #{method}: #{inspect(name)} node: #{node}, module: #{inspect(module)}, bin options: #{inspect(user_options)} """) if node do result = :rpc.call(node, GenServer, :start, [__MODULE__, options]) _ignored = with {:start_link, {:ok, pid}} <- {method, result}, do: Process.link(pid) result else apply(GenServer, method, [__MODULE__, options]) end end @impl GenServer @spec init(%{ :log_metadata => [{any(), any()}], :module => atom(), :name => atom() | tuple(), :parent => pid(), :parent_clock => pid(), :parent_path => [binary()], :parent_supervisor => pid() | port(), :stalker => Membrane.Core.Stalker.t(), :subprocess_supervisor => pid(), :user_options => any(), optional(any()) => any() }) :: {:ok, %{:internal_state => any(), :module => atom(), optional(atom()) => any()}, {:continue, :setup}} def init(options) do Utils.log_on_error do do_init(options) end end defp do_init(options) do Process.link(options.parent_supervisor) %{name: name, module: module} = options observability_config = %{ name: name, component_type: :bin, pid: self(), parent_path: options.parent_path, log_metadata: options.log_metadata } Membrane.Core.Stalker.register_component(options.stalker, observability_config) SubprocessSupervisor.set_parent_component(options.subprocess_supervisor, observability_config) {:ok, clock_proxy} = SubprocessSupervisor.start_utility( options.subprocess_supervisor, {Membrane.Clock, proxy: true} ) clock = if Bunch.Module.check_behaviour(module, :membrane_clock?), do: clock_proxy, else: nil Message.send(options.parent, :clock, [name, clock]) {:ok, resource_guard} = SubprocessSupervisor.start_utility(options.subprocess_supervisor, {ResourceGuard, self()}) LegacyTelemetry.report_init(:bin) ResourceGuard.register(resource_guard, fn -> LegacyTelemetry.report_terminate(:bin) end) state = %State{ module: module, name: name, parent_pid: options.parent, synchronization: %{ parent_clock: options.parent_clock, timers: %{}, clock: clock, clock_provider: %{clock: nil, provider: nil}, clock_proxy: clock_proxy, # This is a sync for siblings. This is not yet allowed. stream_sync: Membrane.Sync.no_sync(), latency: 0 }, children_log_metadata: options.log_metadata, subprocess_supervisor: options.subprocess_supervisor, resource_guard: resource_guard, stalker: options.stalker } |> Child.PadSpecHandler.init_pads() state = CallbackHandler.exec_and_handle_callback( :handle_init, ActionHandler, %{context: &CallbackContext.from_state/1}, [], %{state | internal_state: options.user_options} ) {:ok, state, {:continue, :setup}} end @impl GenServer def handle_continue(:setup, state) do Utils.log_on_error do state = Parent.LifecycleController.handle_setup(state) {:noreply, state} end end @impl GenServer def handle_info(message, state) do Utils.log_on_error do do_handle_info(message, state) end end @compile {:inline, do_handle_info: 2} defp do_handle_info(Message.new(:handle_unlink, pad_ref), state) do state = PadController.handle_unlink(pad_ref, state) {:noreply, state} end defp do_handle_info(Message.new(:parent_notification, notification), state) do state = Child.LifecycleController.handle_parent_notification(notification, state) {:noreply, state} end defp do_handle_info(Message.new(:link_request, [pad_ref, direction, link_id, pad_props]), state) do state = PadController.handle_external_link_request(pad_ref, direction, link_id, pad_props, state) {:noreply, state} end defp do_handle_info( Message.new(:stream_management_event, [element_name, pad_ref, event, event_params]), state ) do state = Parent.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 = Parent.ChildLifeController.handle_child_pad_removed(child, pad, state) {:noreply, state} end defp do_handle_info(Message.new(:child_notification, [from, notification]), state) do state = Parent.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 = Parent.ChildLifeController.handle_link_response(link_id, direction, state) {:noreply, state} end defp do_handle_info(Message.new(:linking_timeout, [pad_ref, linking_timeout_id]), state) do :ok = PadController.handle_linking_timeout(pad_ref, linking_timeout_id, state) {:noreply, state} end defp do_handle_info(Message.new(:child_death, [name, reason]), state) do case Parent.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(:play), state) do state = Parent.LifecycleController.handle_playing(state) {:noreply, state} end defp do_handle_info(Message.new(:initialized, child), state) do state = Parent.ChildLifeController.handle_child_initialized(child, state) {:noreply, state} end defp do_handle_info(Message.new(:terminate), state) do state = Parent.LifecycleController.handle_terminate_request(state) {:noreply, state} end defp do_handle_info(Message.new(_type, _args, _opts) = message, _state) do raise Membrane.BinError, "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 = Parent.LifecycleController.handle_info(message, state) {:noreply, state} end @impl GenServer def handle_call(request, from, state) do Utils.log_on_error do do_handle_call(request, from, state) end end defp do_handle_call( Message.new(:handle_link, [direction, this, other, params]), _from, state ) do {reply, state} = PadController.handle_link(direction, this, other, params, state) {:reply, reply, state} end defp do_handle_call(Message.new(:get_clock), _from, state) do {:reply, state.synchronization.clock, 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 end