connect/disconnect callbacks

Makefile

@@ -10,7 +10,7 @@ ct:
 tests: eunit ct
 
 console:
-	erl -pa _build/default/lib/*/ebin -s epgsql_pool test_run
+	erl -pa _build/default/lib/*/ebin -s epgsql_test test_run
 
 d:
 	rebar3 dialyzer

src/epgsql_pool.app.src

@@ -14,6 +14,13 @@
          {pooler_max_queue, 1000},
          {max_reconnect_timeout, 5000},
          {min_reconnect_timeout, 100},
-         {keep_alive_timeout, 60000}
+         {keep_alive_timeout, 60000},
+         %%
+         %% callback is function of arity 1 which accepts pool name
+         %%
+         %% {on_connect_callback, {Module :: atom(), Function :: atom()}},
+         {on_connect_callback, undefined},
+         %% {on_disconnect_callback, {Module :: atom(), Function :: atom()}}
+         {on_disconnect_callback, undefined}
         ]}
  ]}.

src/epgsql_pool.erl

@@ -5,8 +5,7 @@
          query/2, query/3, query/4,
          squery/2, squery/3,
          transaction/2,
-         get_settings/0, set_settings/1,
-         test_run/0
+         get_settings/0, set_settings/1
         ]).
 
 -include("epgsql_pool.hrl").
@@ -168,24 +167,6 @@ set_settings(Map) ->
     ok.
 
 
--spec test_run() -> ok.
-test_run() ->
-    application:ensure_all_started(epgsql_pool),
-
-    Params = #{host => "localhost",
-               port => 5432,
-               username => "test",
-               password => "test",
-               database => "testdb"},
-    {ok, _} = epgsql_pool:start(my_pool, 2, 2, Params),
-
-
-    Res1 = epgsql_pool:query(my_pool, "select * from category"),
-    error_logger:info_msg("Res1: ~p", [Res1]),
-
-    ok.
-
-
 %%% inner functions
 
 -spec get_worker(pool_name()) -> {ok, pid()} | {error, term()}.

src/epgsql_pool_worker.erl

@@ -99,6 +99,7 @@ handle_info(open_connection, #state{pool_name = PoolName, connection = Connectio
             {ok, KeepAliveTimeout} = application:get_env(epgsql_pool, keep_alive_timeout),
             erlang:cancel_timer(Send_KA_Timer),
             Send_KA_Timer2 = erlang:send_after(KeepAliveTimeout, self(), keep_alive),
+            self() ! on_connect,
             {noreply, State#state{connection = Connection2, send_keep_alive_timer = Send_KA_Timer2}};
         {error, Reason, Connection3} ->
             error_logger:error_msg("Pool:~p, Worker:~p could not to connect to DB:~p", [PoolName, self(), Reason]),
@@ -106,6 +107,22 @@ handle_info(open_connection, #state{pool_name = PoolName, connection = Connectio
             {noreply, State#state{connection = Connection4}}
     end;
 
+handle_info(on_connect, #state{pool_name = PoolName} = State) ->
+    case application:get_env(epgsql_pool, on_connect_callback) of
+        undefined -> ignore;
+        {ok, undefined} -> ignore;
+        {ok, {M, F}} -> M:F(PoolName)
+    end,
+    {noreply, State};
+
+handle_info(on_disconnect, #state{pool_name = PoolName} = State) ->
+    case application:get_env(epgsql_pool, on_disconnect_callback) of
+        undefined -> ignore;
+        {ok, undefined} -> ignore;
+        {ok, {M, F}} -> M:F(PoolName)
+    end,
+    {noreply, State};
+
 handle_info(keep_alive, #state{connection = #epgsql_connection{sock = undefined}} = State) ->
     do_nothing,
     {noreply, State};
@@ -143,6 +160,7 @@ handle_info(no_reply_to_keep_alive, #state{connection = Connection,
     erlang:cancel_timer(NR_KA_Timer),
     Connection2 = epgsql_pool_utils:close_connection(Connection),
     Connection3 = epgsql_pool_utils:reconnect(Connection2),
+    self() ! on_disconnect,
     {noreply, State#state{connection = Connection3}};
 
 handle_info({'EXIT', _Sock, normal},
@@ -155,6 +173,7 @@ handle_info({'EXIT', Sock, Reason},
     error_logger:error_msg("DB Connection ~p~nEXIT with reason:~p", [Connection, Reason]),
     Connection2 = epgsql_pool_utils:close_connection(Connection),
     Connection3 = epgsql_pool_utils:reconnect(Connection2),
+    self() ! on_disconnect,
     {noreply, State#state{connection = Connection3}};
 
 handle_info({'EXIT', _Sock, econnrefused},

src/epgsql_test.erl

@@ -0,0 +1,44 @@
+-module(epgsql_test).
+
+-export([test_run/0, on_connect/1, on_disconnect/1]).
+
+-spec test_run() -> ok.
+test_run() ->
+    application:ensure_all_started(epgsql_pool),
+    application:set_env(epgsql_pool, on_connect_callback, {?MODULE, on_connect}),
+    application:set_env(epgsql_pool, on_disconnect_callback, {?MODULE, on_disconnect}),
+
+    Params = #{host => "localhost",
+        port => 5432,
+        username => "test",
+        password => "test",
+        database => "testdb"},
+    {ok, _} = epgsql_pool:start(my_pool, 2, 2, Params),
+
+    Qs = [
+        "create table category (id int, name varchar)",
+        "insert into category values (1, 'some')",
+        "select * from category"
+    ],
+
+    lists:foreach(
+        fun(Q) ->
+            Res = epgsql_pool:query(my_pool, Q),
+            error_logger:info_msg("Q:~p~nRes:~p~n", [Q, Res])
+        end,
+        Qs),
+    ok.
+
+
+-spec on_connect(term()) -> ok.
+on_connect(PoolName) ->
+    error_logger:info_msg("On Connect ~p", [PoolName]),
+    ok.
+
+
+-spec on_disconnect(term()) -> ok.
+on_disconnect(PoolName) ->
+    error_logger:info_msg("On Disconnect ~p", [PoolName]),
+    ok.
+
+