Introduce DBConnection.Stage

.travis.yml

@@ -2,16 +2,12 @@ language: elixir
 env:
    - ELIXIR_ERL_OPTIONS="+T 9"
 elixir:
-    - 1.2.6
     - 1.3.0
     - 1.3.4
 otp_release:
     - 18.3
     - 19.1
-matrix:
-    include:
-        - elixir: 1.2.0
-          otp_release: 18.2
+    - 19.3
 sudo: false
 script:
       - mix test.all

integration_test/cases/consumer_test.exs

@@ -0,0 +1,35 @@
+defmodule ConsumerTest do
+  use ExUnit.Case, async: true
+
+  alias TestPool, as: P
+  alias TestAgent, as: A
+  alias TestQuery, as: Q
+  alias TestResult, as: R
+
+  test "start_link consumes events" do
+    stack = [
+      {:ok, :state},
+      {:ok, %R{}, :new_state},
+      {:ok, %R{}, :newer_state},
+      ]
+    {:ok, agent} = A.start_link(stack)
+
+    opts = [agent: agent, parent: self(), stage_transaction: false]
+    {:ok, pool} = P.start_link(opts)
+    {:ok, stage} = P.start_consumer(pool, &P.execute!(&1, %Q{}, &2, opts), opts)
+    mon = Process.monitor(stage)
+
+    {:ok, _} =
+      [1, 2]
+      |> Flow.from_enumerable()
+      |> Flow.into_stages([{stage, [cancel: :transient, max_demand: 1]}])
+
+    assert_receive {:DOWN, ^mon, :process, ^stage, :normal}
+
+    assert [
+      connect: [_],
+      handle_execute: [%Q{}, [1], _, :state],
+      handle_execute: [%Q{}, [2], _, :new_state]
+      ] = A.record(agent)
+  end
+end