Proxy info messages to the adapter

examples/tcp_connection/lib/tcp_connection.ex

@@ -148,8 +148,16 @@ defmodule TCPConnection do
   # The handle_info callback is optional and can be removed if not needed.
   # Here it is used to react to `:inet.monitor/1` messages which arrive
   # when socket gets closed while the connection is idle.
-  def handle_info({:DOWN, _ref, _type, sock, _info}, {sock, _buffer} = s) do
-    {:disconnect, TCPConnection.Error.exception({:idle, :closed}), s}
+  def handle_info({:DOWN, _ref, _type, sock, _info}, {sock, _buffer}) do
+    {:disconnect, TCPConnection.Error.exception({:idle, :closed})}
+  end
+
+  def handle_info(msg, state) do
+    Logger.info(fn ->
+      ["#{__MODULE__} (", inspect(self()), ") missed message: ", inspect(msg)]
+    end)
+
+    :ok
   end
 
   @impl true

integration_test/cases/info_test.exs

@@ -1,5 +1,60 @@
-defmodule TestInfo do
+defmodule InfoTest do
   use ExUnit.Case, async: true
+  alias TestPool, as: P
+  alias TestAgent, as: A
 
-  test "handle_info returns new state"
+  test "handle_info handles message and moves on" do
+    stack = [
+      fn opts ->
+        send(opts[:parent], {:connected, self()})
+        {:ok, :state}
+      end,
+      :ok,
+      {:idle, :state},
+      {:idle, :state}
+    ]
+
+    {:ok, agent} = A.start_link(stack)
+    {:ok, pool} = P.start_link(agent: agent, parent: self())
+
+    assert_receive {:connected, conn}
+    send(conn, "some harmless message")
+    assert P.run(pool, fn _ -> :result end) == :result
+
+    assert [
+             connect: _,
+             handle_info: _,
+             handle_status: _,
+             handle_status: _
+           ] = A.record(agent)
+  end
+
+  test "handle_info can force disconnect" do
+    stack = [
+      fn opts ->
+        send(opts[:parent], {:connected, self()})
+        {:ok, :state}
+      end,
+      {:disconnect, :reason},
+      :ok,
+      fn opts ->
+        send(opts[:parent], :reconnected)
+        {:ok, :state}
+      end
+    ]
+
+    {:ok, agent} = A.start_link(stack)
+    P.start_link(agent: agent, parent: self())
+
+    assert_receive {:connected, conn}
+    send(conn, "some harmful message")
+    assert_receive :reconnected
+
+    assert [
+             connect: _,
+             handle_info: _,
+             disconnect: _,
+             connect: _
+           ] = A.record(agent)
+  end
 end

integration_test/connection_pool/disconnect_all_test.exs

@@ -80,6 +80,4 @@ defmodule TestPoolDisconnectAll do
              handle_execute: [_, _, _, :final_state1]
            ] = A.record(agent)
   end
-
-  test "disconnect on info"
 end

lib/db_connection/connection.ex

@@ -333,11 +333,11 @@ defmodule DBConnection.Connection do
   def handle_event(:info, msg, :no_state, %{mod: mod, state: state} = s) do
     if function_exported?(mod, :handle_info, 2) do
       case apply(mod, :handle_info, [msg, state]) do
-        {:ok, state} ->
-          pool_update(state, s)
+        :ok ->
+          handle_timeout(s)
 
-        {:disconnect, err, state} ->
-          {:keep_state, %{s | state: state}, {:next_event, :internal, {:disconnect, {:log, err}}}}
+        {:disconnect, err} ->
+          {:keep_state, s, {:next_event, :internal, {:disconnect, {:log, err}}}}
       end
     else
       Logger.info(fn ->

test/test_support.exs

@@ -150,6 +150,10 @@ defmodule TestConnection do
     TestAgent.eval(:handle_deallocate, [query, cursor, opts, state])
   end
 
+  def handle_info(message, state) do
+    TestAgent.eval(:handle_info, [message, state])
+  end
+
   defp put_agent_from_opts(opts) do
     Process.get(:agent) || Process.put(:agent, agent_from_opts(opts))
   end