mtwilliams/bourne

:speedboat: Better streaming for Ecto.

Bourne

Bourne provides more powerful streaming mechanisms than those offered by Ecto or Tributary. Notably, it provides both cursor and keyset pagination methods, as well as the ability to create a GenStage producer with similar semantics to GenStage.from_enumerable.

Example

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)

Installation

1. Add bourne to your list of dependencies in mix.exs:

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

2. Fetch and compile your new dependency:

mix do deps.get bourne, deps.compile

3. Drink your 🍵

4. That's it!

Usage

Refer to the documentation.

License

Bourne is free and unencumbered software released into the public domain, with fallback provisions for jurisdictions that don't recognize the public domain.

For details, see LICENSE.md.

bourne's Issues

Improve docs about cursor and keyset methods

I am not sure which method to choose. And what are the use-cases for them and how are they different.

I would appreciate a clarification!

function Bourne.stream/1 is undefined or private

defmodule MyApp.Repo do
  use Ecto.Repo, otp_app: :my_app
  use Bourne
end

I'm trying to use this

  def list_queries_by_job(id, attrs \\ []) do
    (from q in Query, where: q.job_id == ^id) |> not_done_queries()
    |> queries_by_type(attrs[:types])
    |> Bourne.stream()
  end

and get error
** (UndefinedFunctionError) function Bourne.stream/1 is undefined or private

I use elixir: "~> 1.6",

Error Handling?

Hey,

Been spending a while looking at your code and couldn't work out what happens if the app/server/gen_stage process crashes?

Does it pick up where it left off on restart or does it go ahead (using your example) and re-read everyone born before 1980? Am I wrong in assuming this is an infinite stream?

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

Support ecto 3

Failed to use "ecto" (version 3.0.6) because
  bourne (version 1.0.0) requires ~> 2.0
  ecto_sql (version 3.0.4) requires ~> 3.0.6
  phoenix_ecto (version 4.0.0) requires ~> 3.0
  mix.lock specifies 3.0.6


Failed to use "ecto" (version 3.0.6) because
  bourne (versions 1.0.1 and 1.1.0) requires ~> 2.1
  ecto_sql (version 3.0.4) requires ~> 3.0.6
  phoenix_ecto (version 4.0.0) requires ~> 3.0
  mix.lock specifies 3.0.6

If you know what needs to change, I'm happy to help out with a PR