Add disconnect_all/2 in Repo (#378)

Introduces Ecto.Adapters.SQL.disconnect_all/2,3 and convenience functions in Repo module.

Bumps db_connection to 2.4.1, required for DBConnection.disconnect_all/3

Author: v0idpwn

integration_test/sql/sql.exs

@@ -1,6 +1,7 @@
 defmodule Ecto.Integration.SQLTest do
   use Ecto.Integration.Case, async: true
 
+  alias Ecto.Integration.PoolRepo
   alias Ecto.Integration.TestRepo
   alias Ecto.Integration.Barebone
   alias Ecto.Integration.Post
@@ -42,6 +43,10 @@ defmodule Ecto.Integration.SQLTest do
     assert result.rows == [[1]]
   end
 
+  test "disconnect_all/2" do
+    assert :ok = PoolRepo.disconnect_all(0)
+  end
+
   test "to_sql/3" do
     {sql, []} = TestRepo.to_sql(:all, Barebone)
     assert sql =~ "SELECT"

lib/ecto/adapters/sql.ex

@@ -321,6 +321,38 @@ defmodule Ecto.Adapters.SQL do
      end
   end
 
+  @doc """
+  Forces all connections in the repo pool to disconnect within the given interval.
+
+  Once this function is called, the pool will disconnect all of its connections
+  as they are checked in or as they are pinged. Checked in connections will be
+  randomly disconnected within the given time interval. Pinged connections are
+  immediately disconnected - as they are idle (according to `:idle_interval`).
+
+  If the connection has a backoff configured (which is the case by default),
+  disconnecting means an attempt at a new connection will be done immediately
+  after, without starting a new process for each connection. However, if backoff
+  has been disabled, the connection process will terminate. In such cases,
+  disconnecting all connections may cause the pool supervisor to restart
+  depending on the max_restarts/max_seconds configuration of the pool,
+  so you will want to set those carefully.
+
+  For convenience, this function is also available in the repository:
+
+      iex> MyRepo.disconnect_all(60_000)
+      :ok
+  """
+  @spec disconnect_all(pid | Ecto.Repo.t | Ecto.Adapter.adapter_meta, non_neg_integer, opts :: Keyword.t()) :: :ok
+  def disconnect_all(repo, interval, opts \\ [])
+
+  def disconnect_all(repo, interval, opts) when is_atom(repo) or is_pid(repo) do
+    disconnect_all(Ecto.Adapter.lookup_meta(repo), interval, opts)
+  end
+
+  def disconnect_all(%{pid: pid} = _adapter_meta, interval, opts) do
+    DBConnection.disconnect_all(pid, interval, opts)
+  end
+
   @doc """
   Returns a stream that runs a custom SQL query on given repo when reduced.
 
@@ -572,6 +604,16 @@ defmodule Ecto.Adapters.SQL do
       def explain(operation, queryable, opts \\ []) do
         Ecto.Adapters.SQL.explain(get_dynamic_repo(), operation, queryable, opts)
       end
+
+      @doc """
+      A convenience function for SQL-based repositories that forces all connections in the
+      pool to disconnect within the given interval.
+
+      See `Ecto.Adapters.SQL.disconnect_all/3` for more information.
+      """
+      def disconnect_all(interval, opts \\ []) do
+        Ecto.Adapters.SQL.disconnect_all(get_dynamic_repo(), interval, opts)
+      end
     end
   end
 

mix.exs

@@ -56,7 +56,7 @@ defmodule EctoSQL.MixProject do
       {:telemetry, "~> 0.4.0 or ~> 1.0"},
 
       # Drivers
-      {:db_connection, "~> 2.2"},
+      {:db_connection, "~> 2.5 or ~> 2.4.1"},
       postgrex_dep(),
       myxql_dep(),
       tds_dep(),

mix.lock

@@ -2,7 +2,7 @@
   "benchee": {:hex, :benchee, "0.11.0", "cf96e328ff5d69838dd89c21a9db22716bfcc6ef772e9d9dddf7ba622102722d", [:mix], [{:deep_merge, "~> 0.1", [hex: :deep_merge, repo: "hexpm", optional: false]}], "hexpm", "c345e090e0a61bf33e0385aa3ad394fcb7d863e313bc3fca522e390c7f39166e"},
   "benchee_json": {:hex, :benchee_json, "0.4.0", "59d3277829bd1dca8373cdb20b916cb435c2647be235d09963fc0959db908c36", [:mix], [{:benchee, "~> 0.10", [hex: :benchee, repo: "hexpm", optional: false]}, {:poison, ">= 1.4.0", [hex: :poison, repo: "hexpm", optional: false]}], "hexpm", "71a3edb6a30708de2a01368aa8f288e1c0ed7897b125adc396ce7c2c7245b1e7"},
   "connection": {:hex, :connection, "1.1.0", "ff2a49c4b75b6fb3e674bfc5536451607270aac754ffd1bdfe175abe4a6d7a68", [:mix], [], "hexpm", "722c1eb0a418fbe91ba7bd59a47e28008a189d47e37e0e7bb85585a016b2869c"},
-  "db_connection": {:hex, :db_connection, "2.3.1", "4c9f3ed1ef37471cbdd2762d6655be11e38193904d9c5c1c9389f1b891a3088e", [:mix], [{:connection, "~> 1.0", [hex: :connection, repo: "hexpm", optional: false]}], "hexpm", "abaab61780dde30301d840417890bd9f74131041afd02174cf4e10635b3a63f5"},
+  "db_connection": {:hex, :db_connection, "2.4.1", "6411f6e23f1a8b68a82fa3a36366d4881f21f47fc79a9efb8c615e62050219da", [:mix], [{:connection, "~> 1.0", [hex: :connection, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "ea36d226ec5999781a9a8ad64e5d8c4454ecedc7a4d643e4832bf08efca01f00"},
   "decimal": {:hex, :decimal, "1.9.0", "83e8daf59631d632b171faabafb4a9f4242c514b0a06ba3df493951c08f64d07", [:mix], [], "hexpm", "b1f2343568eed6928f3e751cf2dffde95bfaa19dd95d09e8a9ea92ccfd6f7d85"},
   "deep_merge": {:hex, :deep_merge, "0.2.0", "c1050fa2edf4848b9f556fba1b75afc66608a4219659e3311d9c9427b5b680b3", [:mix], [], "hexpm", "e3bf435a54ed27b0ba3a01eb117ae017988804e136edcbe8a6a14c310daa966e"},
   "earmark_parser": {:hex, :earmark_parser, "1.4.16", "607709303e1d4e3e02f1444df0c821529af1c03b8578dfc81bb9cf64553d02b9", [:mix], [], "hexpm", "69fcf696168f5a274dd012e3e305027010658b2d1630cef68421d6baaeaccead"},