Introduce Postgrex.Stage #312
Conversation
![sourcelevel-bot[bot]](https://avatars.githubusercontent.com/in/597?s=60&v=4){kind=small}
lib/postgrex/stage.ex
Outdated
| @@ -0,0 +1,93 @@ | |||
| defmodule Postgrex.Stage do | |||
There was a problem hiding this comment.
Modules should have a @moduledoc tag.
| end | ||
|
|
||
| defp copy_data(conn, data, %Postgrex.Copy{ref: ref} = copy, opts) do | ||
| try do |
There was a problem hiding this comment.
Prefer using an implicit try rather than explicit try.
lib/postgrex/protocol.ex
Outdated
| end | ||
|
|
||
| defp suspend(s, status, query, cursor, rows, buffer) do | ||
| defp suspend(s, status, query, _cursor, rows, buffer) do |
There was a problem hiding this comment.
Function takes too many parameters (arity is 6, max is 5).
test/stage_test.exs
Outdated
| test "produce COPY TO STDOUT", context do | ||
| query = "COPY (VALUES (1, 2), (3, 4)) TO STDOUT" | ||
| parent = self() | ||
| opts = [stream_mapper: fn(_, %{rows: rows}) -> send(parent, rows) ; rows end] |
There was a problem hiding this comment.
Don't use ; to separate statements and expressions
test/stage_test.exs
Outdated
| test "produce query in chunks", context do | ||
| query = prepare("", "SELECT * FROM generate_series(1, 3)") | ||
| parent = self() | ||
| opts = [stream_mapper: fn(_, %{rows: rows}) -> send(parent, rows) ; rows end] |
There was a problem hiding this comment.
Don't use ; to separate statements and expressions
test/stage_test.exs
Outdated
| defmodule StageTest do | ||
| use ExUnit.Case, async: true | ||
| import Postgrex.TestHelper | ||
| alias Postgrex.{CopyConsumer, Producer} |
There was a problem hiding this comment.
Most of the time you are using the multiple single line alias/require/import/use directives but here you are using the multi-alias/require/import/use syntax
|
Ebert has finished reviewing this Pull Request and has found:
But beware that this branch is 1 commit behind the You can see more details about this review at https://ebertapp.io/github/elixir-ecto/postgrex/pulls/312. |
| ## Examples | ||
| {:ok, pid} = Postgrex.Producer.start_link(pool, "SELECT * FROM posts, []) |
| {:ok, pid} = Postgrex.Producer.start_link(pool, "SELECT * FROM posts, []) | ||
| [{pid, [cancel: :transient]}] | ||
| |> GenStage.stream() |
There was a problem hiding this comment.
Instead of this example, let's make an imaginary consumer subscribe to the producer. That's because this use case is better done with streams.
|
❤️ 💚 💙 💛 💜 |
| ## Examples | ||
| {:ok, pid} = Postgrex.Producer.start_link(pool, "SELECT * FROM posts, []) |
There was a problem hiding this comment.
Thanks for taking a look, we aren't going to go ahead with this approach. Please see #321.
This is WIP. Includes a common real world example of a
:consumerstage.