Packages
electric
1.6.6
1.7.8
1.7.7
1.7.6
1.7.5
1.7.4
1.7.3
1.7.2
1.7.1
1.7.0
1.6.10
1.6.9
1.6.8
1.6.7
1.6.6
1.6.5
1.6.4
1.6.3
1.6.2
1.6.1
1.6.0
1.5.1
1.5.0
1.4.16
1.4.16-beta-1
1.4.15
1.4.14
1.4.13
1.4.12
1.4.11
1.4.10
1.4.8
1.4.7
1.4.6
1.4.5
1.4.4
1.4.3
1.4.2
1.4.1
1.4.0
1.3.4
1.3.3
1.3.2
1.2.4
1.2.3
1.2.2
1.2.1
1.2.0
1.1.14
1.1.13
1.1.12
1.1.11
1.1.10
1.1.9
1.1.8
1.1.7
1.1.6
retired
1.1.5
retired
1.1.4
retired
1.1.3
retired
1.1.2
1.1.1
1.1.0
1.0.24
1.0.23
1.0.22
1.0.21
1.0.20
1.0.19
1.0.18
1.0.17
1.0.15
1.0.13
1.0.12
1.0.11
1.0.10
1.0.9
1.0.5
1.0.4
1.0.3
1.0.2
1.0.1
1.0.0
1.0.0-beta.23
1.0.0-beta.22
1.0.0-beta.20
1.0.0-beta.19
1.0.0-beta.18
1.0.0-beta.17
1.0.0-beta.16
1.0.0-beta.15
1.0.0-beta.14
1.0.0-beta.13
1.0.0-beta.12
1.0.0-beta.11
1.0.0-beta.10
1.0.0-beta.9
1.0.0-beta.8
1.0.0-beta.7
1.0.0-beta.6
1.0.0-beta.5
1.0.0-beta.4
1.0.0-beta.3
1.0.0-beta.2
1.0.0-beta.1
0.9.5
0.9.4
0.9.3
0.9.2
0.9.1
0.9.0
0.8.1
0.8.0
0.7.7
0.7.6
0.7.5
0.7.4
0.7.3
0.7.2
0.7.1
0.7.0
0.6.3
0.6.2
0.6.1
0.5.2
0.4.4
Postgres sync engine. Sync little subsets of your Postgres data into local apps and services.
Current section
Files
Jump to
Current section
Files
lib/electric/core_supervisor.ex
defmodule Electric.CoreSupervisor do
@moduledoc """
A supervisor that starts the core components of the Electric system.
This is divided into two subsystems:
1. The connection subsystem (processes that may exit on a connection failure), started with Connection.Supervisor
2. The shape subsystem (processes that are resilient to connection failures), started with Shapes.Supervisor
"""
use Supervisor, restart: :transient, significant: true
def name(stack_ref) do
Electric.ProcessRegistry.name(stack_ref, __MODULE__)
end
def start_link(opts) do
Supervisor.start_link(__MODULE__, opts, name: name(opts))
end
@impl true
def init(opts) do
stack_id = Keyword.fetch!(opts, :stack_id)
Process.set_label({:core_supervisor, stack_id})
Logger.metadata(stack_id: stack_id)
Electric.Telemetry.Sentry.set_tags_context(stack_id: stack_id)
connection_manager_opts = Keyword.fetch!(opts, :connection_manager_opts)
children = [
{Electric.Connection.Supervisor, connection_manager_opts}
]
Supervisor.init(children, strategy: :one_for_one, auto_shutdown: :any_significant)
end
@doc """
This function is supposed to be called from Connection.Manager at the right point in its
initialization sequence.
"""
def start_shapes_supervisor(opts) do
stack_id = Keyword.fetch!(opts, :stack_id)
shape_cache_opts = Keyword.fetch!(opts, :shape_cache_opts)
replication_opts = Keyword.fetch!(opts, :replication_opts)
inspector = Keyword.fetch!(opts, :inspector)
persistent_kv = Keyword.fetch!(opts, :persistent_kv)
tweaks = Keyword.fetch!(opts, :tweaks)
max_shapes = Keyword.fetch!(opts, :max_shapes)
consumer_supervisor_spec =
{Electric.Shapes.DynamicConsumerSupervisor,
[
stack_id: stack_id,
max_shapes: max_shapes,
partitions: Keyword.get(tweaks, :consumer_partitions)
]}
shape_cache_spec = {Electric.ShapeCache, shape_cache_opts}
publication_manager_spec =
{Electric.Replication.PublicationManager,
stack_id: stack_id,
publication_name: Keyword.fetch!(replication_opts, :publication_name),
manual_table_publishing?: Keyword.fetch!(opts, :manual_table_publishing?),
db_pool: Electric.Connection.Manager.admin_pool(stack_id),
update_debounce_timeout: Keyword.get(tweaks, :publication_alter_debounce_ms, 0),
refresh_period: Keyword.get(tweaks, :publication_refresh_period, 60_000)}
shape_log_collector_spec =
{Electric.Replication.ShapeLogCollector.Supervisor,
stack_id: stack_id, inspector: inspector, persistent_kv: persistent_kv}
schema_reconciler_spec =
{Electric.Replication.SchemaReconciler,
stack_id: stack_id,
inspector: inspector,
period: Keyword.get(tweaks, :schema_reconciler_period, 60_000)}
expiry_manager_spec =
{Electric.ShapeCache.ExpiryManager, max_shapes: max_shapes, stack_id: stack_id}
child_spec =
Supervisor.child_spec(
{
Electric.Shapes.Supervisor,
stack_id: stack_id,
consumer_supervisor: consumer_supervisor_spec,
shape_cache: shape_cache_spec,
publication_manager: publication_manager_spec,
log_collector: shape_log_collector_spec,
schema_reconciler: schema_reconciler_spec,
expiry_manager: expiry_manager_spec
},
restart: :transient
)
Supervisor.start_child(name(opts), child_spec)
end
@doc """
Stops the Shapes.Supervisor if it's currently running.
This is useful when you need to reset storage before starting a new supervisor.
Returns :ok if the supervisor was stopped or wasn't running.
"""
def stop_shapes_supervisor(opts) do
case Supervisor.terminate_child(name(opts), Electric.Shapes.Supervisor) do
:ok ->
Supervisor.delete_child(name(opts), Electric.Shapes.Supervisor)
:ok
{:error, :not_found} ->
:ok
end
end
end