Packages
electric
1.1.11
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/replication/supervisor.ex
defmodule Electric.Replication.Supervisor do
@moduledoc """
Supervisor responsible for the entire shape subsystem.
It starts up and supervises the processes that manage shapes (create/remove), keep the
Postgres publication up to date, consume incoming transactions and write them to shape logs.
It also supervisers the consumer supervisor which starts a new consumer process for each
shape.
"""
use Supervisor
require Logger
def name(stack_id) when is_binary(stack_id) do
Electric.ProcessRegistry.name(stack_id, __MODULE__)
end
def name(opts), do: name(opts[:stack_id])
def start_link(opts) do
name = Access.get(opts, :name, name(opts))
Supervisor.start_link(__MODULE__, opts, name: name)
end
@impl Supervisor
def init(opts) do
Process.set_label({:replication_supervisor, opts[:stack_id]})
Logger.metadata(stack_id: opts[:stack_id])
Electric.Telemetry.Sentry.set_tags_context(stack_id: opts[:stack_id])
Logger.info("Starting shape replication pipeline")
shape_status_owner = Keyword.fetch!(opts, :shape_status_owner)
log_collector = Keyword.fetch!(opts, :log_collector)
publication_manager = Keyword.fetch!(opts, :publication_manager)
consumer_supervisor = Keyword.fetch!(opts, :consumer_supervisor)
shape_cache = Keyword.fetch!(opts, :shape_cache)
schema_reconciler = Keyword.fetch!(opts, :schema_reconciler)
expiry_manager = Keyword.fetch!(opts, :expiry_manager)
stack_id = Keyword.fetch!(opts, :stack_id)
children = [
{Task.Supervisor,
name: Electric.ProcessRegistry.name(stack_id, Electric.StackTaskSupervisor)},
shape_status_owner,
log_collector,
publication_manager,
consumer_supervisor,
expiry_manager,
shape_cache,
schema_reconciler
]
Supervisor.init(children, strategy: :one_for_all)
end
end