defmodule My.Repo do
  use Ecto.Repo, otp_app: :mine
  use Bourne
end

import Ecto.Query
q = from(actor in Actor, where: actor.born <= 1980)

# You can stream through an `Enumerable`:
My.Repo.stream(q) |> Stream.each(&IO.inspect) |> Stream.run

# Alternatively, you can stream through a GenStage producer:
defmodule InspectorConsumer do
  use GenStage

  def start_link do
    GenStage.start_link(InspectorConsumer, [])
  end

  def init([]) do
    {:consumer, :ok}
  end

  def handle_events(rows, _from, state) do
    Enum.each(rows, &IO.inspect/1)
    {:noreply, [], state}
  end
end

method = Enum.take_random(~W{cursor keyset}a, 1)
{:ok, producer} = My.Repo.streamer(q, method: method)
{:ok, consumer} = InspectorConsumer.start_link
GenStage.sync_subscribe(consumer, to: producer)

def deps do
  [{:bourne, "~> 1.0"}]
end