Packages

A toolkit for data mapping and language integrated query for Elixir

Current section

Files

Jump to
ecto_tc lib ecto repo supervisor.ex
Raw

lib/ecto/repo/supervisor.ex

defmodule Ecto.Repo.Supervisor do
@moduledoc false
use Supervisor
@defaults [timeout: 15000, pool_size: 10]
@integer_url_query_params ["timeout", "pool_size"]
@doc """
Starts the repo supervisor.
"""
def start_link(repo, otp_app, adapter, opts) do
name = Keyword.get(opts, :name, repo)
sup_opts = if name, do: [name: name], else: []
Supervisor.start_link(__MODULE__, {name, repo, otp_app, adapter, opts}, sup_opts)
end
@doc """
Retrieves the runtime configuration.
"""
def runtime_config(type, repo, otp_app, opts) do
config = Application.get_env(otp_app, repo, [])
config = [otp_app: otp_app] ++ (@defaults |> Keyword.merge(config) |> Keyword.merge(opts))
config = Keyword.put_new_lazy(config, :telemetry_prefix, fn -> telemetry_prefix(repo) end)
case repo_init(type, repo, config) do
{:ok, config} ->
{url, config} = Keyword.pop(config, :url)
{:ok, Keyword.merge(config, parse_url(url || ""))}
:ignore ->
:ignore
end
end
defp telemetry_prefix(repo) do
repo
|> Module.split()
|> Enum.map(& &1 |> Macro.underscore() |> String.to_atom())
end
defp repo_init(type, repo, config) do
if Code.ensure_loaded?(repo) and function_exported?(repo, :init, 2) do
repo.init(type, config)
else
{:ok, config}
end
end
@doc """
Retrieves the compile time configuration.
"""
def compile_config(_repo, opts) do
otp_app = Keyword.fetch!(opts, :otp_app)
adapter = opts[:adapter]
unless adapter do
raise ArgumentError, "missing :adapter option on use Ecto.Repo"
end
if Code.ensure_compiled(adapter) != {:module, adapter} do
raise ArgumentError, "adapter #{inspect adapter} was not compiled, " <>
"ensure it is correct and it is included as a project dependency"
end
behaviours =
for {:behaviour, behaviours} <- adapter.__info__(:attributes),
behaviour <- behaviours,
do: behaviour
unless Ecto.Adapter in behaviours do
raise ArgumentError,
"expected :adapter option given to Ecto.Repo to list Ecto.Adapter as a behaviour"
end
{otp_app, adapter, behaviours}
end
@doc """
Parses an Ecto URL allowed in configuration.
The format must be:
"ecto://username:password@hostname:port/database?ssl=true&timeout=1000"
"""
def parse_url(""), do: []
def parse_url(url) when is_binary(url) do
info = URI.parse(url)
if is_nil(info.host) do
raise Ecto.InvalidURLError, url: url, message: "host is not present"
end
if is_nil(info.path) or not (info.path =~ ~r"^/([^/])+$") do
raise Ecto.InvalidURLError, url: url, message: "path should be a database name"
end
destructure [username, password], info.userinfo && String.split(info.userinfo, ":")
"/" <> database = info.path
url_opts = [
username: username,
password: password,
database: database,
port: info.port
]
url_opts = put_hostname_if_present(url_opts, info.host)
query_opts = parse_uri_query(info)
for {k, v} <- url_opts ++ query_opts,
not is_nil(v),
do: {k, if(is_binary(v), do: URI.decode(v), else: v)}
end
defp put_hostname_if_present(keyword, "") do
keyword
end
defp put_hostname_if_present(keyword, hostname) when is_binary(hostname) do
Keyword.put(keyword, :hostname, hostname)
end
defp parse_uri_query(%URI{query: nil}),
do: []
defp parse_uri_query(%URI{query: query} = url) do
query
|> URI.query_decoder()
|> Enum.reduce([], fn
{"ssl", "true"}, acc ->
[{:ssl, true}] ++ acc
{"ssl", "false"}, acc ->
[{:ssl, false}] ++ acc
{key, value}, acc when key in @integer_url_query_params ->
[{String.to_atom(key), parse_integer!(key, value, url)}] ++ acc
{key, value}, acc ->
[{String.to_atom(key), value}] ++ acc
end)
end
defp parse_integer!(key, value, url) do
case Integer.parse(value) do
{int, ""} ->
int
_ ->
raise Ecto.InvalidURLError,
url: url,
message: "can not parse value `#{value}` for parameter `#{key}` as an integer"
end
end
## Callbacks
@doc false
def init({name, repo, otp_app, adapter, opts}) do
# Normalize name to atom, ignore via/global names
name = if is_atom(name), do: name, else: nil
case runtime_config(:supervisor, repo, otp_app, opts) do
{:ok, opts} ->
:telemetry.execute(
[:ecto, :repo, :init],
%{system_time: System.system_time()},
%{repo: repo, opts: opts}
)
{:ok, child, meta} = adapter.init([repo: repo] ++ opts)
cache = Ecto.Query.Planner.new_query_cache(name)
meta = Map.merge(meta, %{repo: repo, cache: cache})
child_spec = wrap_child_spec(child, [name, adapter, meta])
Supervisor.init([child_spec], strategy: :one_for_one, max_restarts: 0)
:ignore ->
:ignore
end
end
def start_child({mod, fun, args}, name, adapter, meta) do
case apply(mod, fun, args) do
{:ok, pid} ->
meta = Map.put(meta, :pid, pid)
Ecto.Repo.Registry.associate(self(), name, {adapter, meta})
{:ok, pid}
other ->
other
end
end
defp wrap_child_spec({id, start, restart, shutdown, type, mods}, args) do
{id, {__MODULE__, :start_child, [start | args]}, restart, shutdown, type, mods}
end
defp wrap_child_spec(%{start: start} = spec, args) do
%{spec | start: {__MODULE__, :start_child, [start | args]}}
end
end