Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion lib/flow.ex
Original file line number Diff line number Diff line change
Expand Up @@ -1538,9 +1538,13 @@ defmodule Flow do

defp stages(options) do
case Keyword.fetch(options, :stages) do
{:ok, _} ->
{:ok, stages} when is_integer(stages) and stages > 0 ->
options

{:ok, stages} ->
raise ArgumentError,
":stages must be a positive integer, got: #{inspect(stages)}"

:error ->
stages = System.schedulers_online()
[stages: stages] ++ options
Expand Down
18 changes: 18 additions & 0 deletions test/flow_test.exs
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,24 @@ defmodule FlowTest do
refute_received 1
end

describe ":stages option" do
test "requires a positive integer when constructing a flow" do
constructors = [
fn stages -> Flow.from_enumerable([1], stages: stages) end,
fn stages -> Flow.from_enumerables([[1]], stages: stages) end,
fn stages -> Flow.from_stages([self()], stages: stages) end,
fn stages -> Flow.from_specs([{Counter, 0}], stages: stages) end,
fn stages -> Flow.partition(Flow.from_enumerable([1]), stages: stages) end
]

for constructor <- constructors, stages <- [0, -1, :invalid] do
assert_raise ArgumentError, ~r/:stages must be a positive integer/, fn ->
constructor.(stages)
end
end
end
end

Comment thread
josevalim marked this conversation as resolved.
Outdated
describe "errors" do
test "on multiple reduce calls" do
message = ~r"cannot call group_by/reduce/emit_and_reduce on a flow after another"
Expand Down