Introduce declare/fetch/deallocate (#89)

* Add declare(!)/3,4
* Add fetch(!)/3,4
* Add deallocate(!)/3,4

Explicitly defining handle_first/4 and handle_next/4 is deprecated
because callback implementations that are required to differentiate also
need to track cursors in their state. A single fetch/4 is cleaner to use
and these callbacks forward to handle_fetch/4 on `use DBConnection`.

For first/next/fetch the return value is `{:cont | :halt, result, state}`
where `:cont` is continue (same as `:ok`) and `:halt` means both that the
cursor has finished enumerating results and the cursor is deallocated.
Therefore `:halt` does not require a `deallocate` call. This API is
chosen to avoid an extra roundtrip that current adapters do.
handle_first/4 and handle_next/4 still support `:ok` and `:deallocate`
tuples.

A callback implementation may need to deallocate if the transaction ends
without a cursor being deallocated (via `fetch/4` or `deallocate/4`).

Author: fishcakez

integration_test/cases/cursor_test.exs

@@ -0,0 +1,173 @@
+defmodule CursorTest do
+  use ExUnit.Case, async: true
+
+  alias TestPool, as: P
+  alias TestAgent, as: A
+  alias TestQuery, as: Q
+  alias TestCursor, as: C
+  alias TestResult, as: R
+
+  test "declare/fetch/deallocate return result" do
+    stack = [
+      {:ok, :state},
+      {:ok, %C{}, :new_state},
+      {:ok, %C{}, :newer_state},
+      {:cont, %R{}, :newest_state},
+      {:halt, %R{}, :state2},
+      {:ok, :deallocated, :new_state2},
+      {:ok, :deallocated, :newer_state2}
+      ]
+    {:ok, agent} = A.start_link(stack)
+
+    opts = [agent: agent, parent: self()]
+    {:ok, pool} = P.start_link(opts)
+    assert P.declare(pool, %Q{}, [:param]) == {:ok, %C{}}
+    assert P.declare!(pool, %Q{}, [:param], [key: :value]) == %C{}
+
+    assert P.fetch(pool, %Q{}, %C{}) == {:cont, %R{}}
+    assert P.fetch!(pool, %Q{}, %C{}, [key: :value]) == {:halt, %R{}}
+
+    assert P.deallocate(pool, %Q{}, %C{}) == {:ok, :deallocated}
+    assert P.deallocate!(pool, %Q{}, %C{}, [key: :value]) == :deallocated
+
+    assert [
+      connect: [_],
+      handle_declare: [%Q{}, [:param], _, :state],
+      handle_declare: [%Q{}, [:param], [{:key, :value} | _], :new_state],
+      handle_fetch: [%Q{}, %C{}, _, :newer_state],
+      handle_fetch: [%Q{}, %C{}, [{:key, :value} | _], :newest_state],
+      handle_deallocate: [%Q{}, %C{}, _, :state2],
+      handle_deallocate: [%Q{}, %C{}, [{:key, :value} | _], :new_state2]
+      ] = A.record(agent)
+  end
+
+  test "declare encodes params" do
+    stack = [
+      {:ok, :state},
+      {:ok, %C{}, :new_state},
+      ]
+    {:ok, agent} = A.start_link(stack)
+
+    opts = [agent: agent, parent: self()]
+    {:ok, pool} = P.start_link(opts)
+
+    opts2 = [encode: fn([:param]) -> :encoded end]
+    assert P.declare(pool, %Q{}, [:param], opts2) == {:ok, %C{}}
+
+    assert [
+      connect: [_],
+      handle_declare: [%Q{}, :encoded, _, :state]] = A.record(agent)
+  end
+
+  test "fetch decodes result" do
+    stack = [
+      {:ok, :state},
+      {:cont, %R{}, :new_state},
+      {:halt, %R{}, :newer_state}
+      ]
+    {:ok, agent} = A.start_link(stack)
+
+    opts = [agent: agent, parent: self()]
+    {:ok, pool} = P.start_link(opts)
+
+    opts2 = [decode: fn(%R{}) -> :decoded end]
+    assert P.fetch(pool, %Q{}, %C{}, opts2) == {:cont, :decoded}
+    assert P.fetch(pool, %Q{}, %C{}, opts2) == {:halt, :decoded}
+
+    assert [
+      connect: [_],
+      handle_fetch: [%Q{}, %C{}, _, :state],
+      handle_fetch: [%Q{}, %C{}, _, :new_state]
+      ] = A.record(agent)
+  end
+
+  test "declare/fetch/deallocate logs result" do
+    stack = [
+      {:ok, :state},
+      {:ok, %C{}, :new_state},
+      {:cont, %R{}, :newer_state},
+      {:halt, %R{}, :newest_state},
+      {:ok, :deallocated, :state2}
+      ]
+    {:ok, agent} = A.start_link(stack)
+
+    parent = self()
+    opts = [agent: agent, parent: parent]
+    {:ok, pool} = P.start_link(opts)
+
+    log = &send(parent, &1)
+    assert P.declare(pool, %Q{}, [:param], [log: log]) == {:ok, %C{}}
+
+    assert_receive %DBConnection.LogEntry{call: :declare, query: %Q{},
+      params: [:param], result: {:ok, %C{}}} = entry
+    assert is_integer(entry.pool_time)
+    assert entry.pool_time >= 0
+    assert is_integer(entry.connection_time)
+    assert entry.connection_time >= 0
+    assert is_nil(entry.decode_time)
+
+    assert P.fetch(pool, %Q{}, %C{}, [log: log]) == {:cont, %R{}}
+
+    assert_receive %DBConnection.LogEntry{call: :fetch, query: %Q{},
+      params: %C{}, result: {:ok, %R{}}} = entry
+    assert is_integer(entry.pool_time)
+    assert entry.pool_time >= 0
+    assert is_integer(entry.connection_time)
+    assert entry.connection_time >= 0
+    assert is_integer(entry.decode_time)
+    assert entry.decode_time >= 0
+
+    assert P.fetch(pool, %Q{}, %C{}, [log: log]) == {:halt, %R{}}
+
+    assert_receive %DBConnection.LogEntry{call: :fetch, query: %Q{},
+      params: %C{}, result: {:ok, %R{}}} = entry
+    assert is_integer(entry.pool_time)
+    assert entry.pool_time >= 0
+    assert is_integer(entry.connection_time)
+    assert entry.connection_time >= 0
+    assert is_integer(entry.decode_time)
+    assert entry.decode_time >= 0
+
+    assert P.deallocate(pool, %Q{}, %C{}, [log: log]) == {:ok, :deallocated}
+
+    assert_receive %DBConnection.LogEntry{call: :deallocate, query: %Q{},
+      params: %C{}, result: {:ok, :deallocated}} = entry
+    assert is_integer(entry.pool_time)
+    assert entry.pool_time >= 0
+    assert is_integer(entry.connection_time)
+    assert entry.connection_time >= 0
+    assert is_nil(entry.decode_time)
+
+    assert [
+      connect: [_],
+      handle_declare: [%Q{}, [:param], _, :state],
+      handle_fetch: [%Q{}, %C{}, _, :new_state],
+      handle_fetch: [%Q{}, %C{}, _, :newer_state],
+      handle_deallocate: [%Q{}, %C{}, _, :newest_state]
+      ] = A.record(agent)
+  end
+
+  test "declare/fetch/deallocate error returns error" do
+    err = RuntimeError.exception("oops")
+    stack = [
+      {:ok, :state},
+      {:error, err, :new_state},
+      {:error, err, :newer_state},
+      {:error, err, :newesr_state}
+      ]
+    {:ok, agent} = A.start_link(stack)
+
+    opts = [agent: agent, parent: self()]
+    {:ok, pool} = P.start_link(opts)
+    assert P.declare(pool, %Q{}, [:param]) == {:error, err}
+    assert P.fetch(pool, %Q{}, %C{}) == {:error, err}
+    assert P.deallocate(pool, %Q{}, %C{}) == {:error, err}
+
+    assert [
+      connect: [_],
+      handle_declare: [%Q{}, [:param], _, :state],
+      handle_fetch: [%Q{}, %C{}, _, :new_state],
+      handle_deallocate: [%Q{}, %C{}, _, :newer_state]
+      ] = A.record(agent)
+  end
+end

integration_test/cases/prepare_stream_test.exs

@@ -105,7 +105,7 @@ defmodule PrepareStreamTest do
     assert entry.connection_time >= 0
     assert is_nil(entry.decode_time)
 
-    assert_received %DBConnection.LogEntry{call: :first} = entry
+    assert_received %DBConnection.LogEntry{call: :fetch} = entry
     assert %{query: %Q{}, params: %C{}, result: {:ok, %R{}}} = entry
     assert is_nil(entry.pool_time)
     assert is_integer(entry.connection_time)

integration_test/cases/stream_test.exs

@@ -99,7 +99,7 @@ defmodule StreamTest do
     assert entry.connection_time >= 0
     assert is_nil(entry.decode_time)
 
-    assert_received %DBConnection.LogEntry{call: :first} = entry
+    assert_received %DBConnection.LogEntry{call: :fetch} = entry
     assert %{query: %Q{}, params: %C{}, result: {:ok, %R{}}} = entry
     assert is_nil(entry.pool_time)
     assert is_integer(entry.connection_time)
@@ -221,14 +221,19 @@ defmodule StreamTest do
 
     assert_received %DBConnection.LogEntry{call: :declare}
 
-    assert_received %DBConnection.LogEntry{call: :first} = entry
+    assert_received %DBConnection.LogEntry{call: :fetch} = entry
     assert %{query: %Q{}, params: %C{}, result: {:error, ^err}} = entry
     assert is_nil(entry.pool_time)
     assert is_integer(entry.connection_time)
     assert entry.connection_time >= 0
     assert is_nil(entry.decode_time)
 
-    refute_received %DBConnection.LogEntry{call: :deallocate}
+    assert_received %DBConnection.LogEntry{call: :deallocate} = entry
+    closed = DBConnection.ConnectionError.exception("connection is closed")
+    assert %{query: %Q{}, params: %C{}, result: {:error, ^closed}} = entry
+    assert is_nil(entry.pool_time)
+    assert is_nil(entry.connection_time)
+    assert is_nil(entry.decode_time)
 
     assert_receive :reconnected
 
@@ -297,12 +302,12 @@ defmodule StreamTest do
 
     assert P.transaction(pool, fn(conn) ->
       stream = P.stream(conn, %Q{}, [:param], [log: &send(parent, &1)])
-      assert_raise RuntimeError, "oops",  fn() -> Enum.take(stream, 1) end
+      assert Enum.take(stream, 1) == [%R{}]
       :hi
     end) == {:error, :rollback}
 
     assert_received %DBConnection.LogEntry{call: :declare}
-    assert_received %DBConnection.LogEntry{call: :first}
+    assert_received %DBConnection.LogEntry{call: :fetch}
 
     assert_received %DBConnection.LogEntry{call: :deallocate} = entry
     assert %{query: %Q{}, params: %C{}, result: {:error, ^err}} = entry

integration_test/tests.exs

@@ -2,6 +2,7 @@ Code.require_file "cases/after_connect_test.exs", __DIR__
 Code.require_file "cases/backoff_test.exs", __DIR__
 Code.require_file "cases/client_test.exs", __DIR__
 Code.require_file "cases/close_test.exs", __DIR__
+Code.require_file "cases/cursor_test.exs", __DIR__
 Code.require_file "cases/execute_test.exs", __DIR__
 Code.require_file "cases/idle_test.exs", __DIR__
 Code.require_file "cases/overflow_test.exs", __DIR__