Ensure flow options effectively override producer dispatcher

Author: josevalim

lib/flow.ex

@@ -1291,7 +1291,7 @@ defmodule Flow do
   """
   @spec partition(t | [t], keyword()) :: t
   def partition(flow_or_flows, options \\ []) when is_list(options) do
-    merge(List.wrap(flow_or_flows), GenStage.PartitionDispatcher, options)
+    merge(flow_or_flows, GenStage.PartitionDispatcher, options)
   end
 
   @doc """

lib/flow/materialize.ex

@@ -13,7 +13,7 @@ defmodule Flow.Materialize do
 
   def materialize(%Flow{} = flow, demand, start_link, type, type_options) do
     %{operations: operations, options: options, producers: producers, window: window} = flow
-    options = Keyword.merge(type_options, options)
+    options = Keyword.merge(options, type_options)
     {ops, batchers} = split_operations(operations)
 
     {producers, consumers, ops, window} =
@@ -148,9 +148,9 @@ defmodule Flow.Materialize do
          ops,
          start_link,
          window,
-         options
+         _options
        ) do
-    {producers, consumers} = materialize(flow, :forward, start_link, :producer_consumer, options)
+    {producers, consumers} = materialize(flow, :forward, start_link, :producer_consumer, [])
     {type, {acc, fun, trigger}, ops} = ensure_ops(ops)
 
     stages = Keyword.fetch!(flow.options, :stages)
@@ -195,7 +195,7 @@ defmodule Flow.Materialize do
          options
        ) do
     {producers, intermediary} =
-      materialize(flow, :forward, start_link, :producer_consumer, options)
+      materialize(flow, :forward, start_link, :producer_consumer, [])
 
     timeout = Keyword.get(options, :subscribe_timeout, 5_000)
 

test/flow_test.exs

@@ -1099,6 +1099,20 @@ defmodule FlowTest do
              |> Enum.sort() == [[1, 2, 4, 5], [3, 6]]
     end
 
+    test "allows function based partitioning after shuffling" do
+      enumerables = [
+        [%{key: 1, value: 1}, %{key: 2, value: 2}, %{key: 3, value: 3}],
+        [%{key: 1, value: 4}, %{key: 2, value: 5}, %{key: 3, value: 6}]
+      ]
+
+      assert Flow.from_enumerables(enumerables)
+             |> Flow.shuffle(stages: 2)
+             |> Flow.partition(key: & &1.key, stages: 2)
+             |> Flow.reduce(fn -> [] end, &[&1 | &2])
+             |> Flow.on_trigger(fn acc -> {[acc |> Enum.map(& &1.value) |> Enum.sort()], acc} end)
+             |> Enum.sort() == [[1, 2, 4, 5], [3, 6]]
+    end
+
     test "allows custom windowing" do
       window =
         Flow.Window.fixed(1, :second, fn