Release v1.0.0

Author: josevalim

.travis.yml

@@ -4,15 +4,12 @@ elixir:
 sudo: false # to use faster container based build environment
 notifications:
   recipients:
-    - jose.valim@plataformatec.com.br
+    - jose.valim@dashbit.co
 otp_release:
   - 22.0
 script:
   - if [[ $(elixir --version) = *"1.6"* ]]; then mix format --check-formatted; fi
   - mix test
-after_script:
-  - mix deps.get --only docs
-  - MIX_ENV=docs mix inch.report
 matrix:
   include:
     - elixir: 1.5.1

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## v1.0.0 (2020-02-03)
+
+### Enhancements
+
+  * Allow events to be discarded in PartitionDispatcher by returning `:none`
+  * Raise for unknown partitionis in PartitionDispatcher
+
 ## v0.14.3 (2019-10-28)
 
 ### Enhancements

lib/gen_stage/dispatchers/partition_dispatcher.ex

@@ -24,7 +24,8 @@ defmodule GenStage.PartitionDispatcher do
       and the partition as second. The partition must be one of the partitions
       specified in `:partitions` above. The default uses
       `fn event -> {event, :erlang.phash2(event, Enum.count(partitions))} end`
-      on the event to select the partition.
+      on the event to select the partition. If it returns `:none`, the event
+      is discarded.
 
   ### Examples
 
@@ -239,30 +240,7 @@ defmodule GenStage.PartitionDispatcher do
 
   @doc false
   def dispatch(events, _length, {tag, hash, waiting, pending, partitions, references, infos}) do
-    {deliver_now, deliver_later, waiting} = split_events(events, waiting, [])
-
-    for event <- deliver_now do
-      {event, partition} =
-        case hash.(event) do
-          {event, partition} ->
-            {event, partition}
-
-          other ->
-            raise "the :hash function should return {event, partition}, got: #{inspect(other)}"
-        end
-
-      case :erlang.get(partition) do
-        :undefined ->
-          Logger.error(fn ->
-            "Unknown partition #{inspect(partition)} computed for GenStage/Flow event " <>
-              "#{inspect(event)}. The known partitions are #{inspect(Map.keys(partitions))}. " <>
-              "See the :partitions option to set your own. This event has been discarded."
-          end)
-
-        current ->
-          Process.put(partition, [event | current])
-      end
-    end
+    {deliver_later, waiting} = split_events(events, waiting, hash, partitions)
 
     partitions =
       partitions
@@ -273,11 +251,30 @@ defmodule GenStage.PartitionDispatcher do
     {:ok, deliver_later, {tag, hash, waiting, pending, partitions, references, infos}}
   end
 
-  defp split_events(events, 0, acc), do: {:lists.reverse(acc), events, 0}
-  defp split_events([], counter, acc), do: {:lists.reverse(acc), [], counter}
+  defp split_events(events, 0, _hash, _partitions), do: {events, 0}
+  defp split_events([], counter, _hash, _partitions), do: {[], counter}
+
+  defp split_events([event | events], counter, hash, partitions) do
+    case hash.(event) do
+      {event, partition} ->
+        case :erlang.get(partition) do
+          :undefined ->
+            raise "unknown partition #{inspect(partition)} computed for GenStage event " <>
+                    "#{inspect(event)}. The known partitions are #{inspect(Map.keys(partitions))}. " <>
+                    "See the :partitions option to set your own. This event has been discarded."
+
+          current ->
+            Process.put(partition, [event | current])
+            split_events(events, counter - 1, hash, partitions)
+        end
+
+      :none ->
+        split_events(events, counter, hash, partitions)
 
-  defp split_events([event | events], counter, acc),
-    do: split_events(events, counter - 1, [event | acc])
+      other ->
+        raise "the :hash function should return {event, partition}, got: #{inspect(other)}"
+    end
+  end
 
   defp dispatch_per_partition([{partition, {pid, ref, demand_or_queue} = value} | rest]) do
     case Process.put(partition, []) do

mix.exs

@@ -1,7 +1,7 @@
 defmodule GenStage.Mixfile do
   use Mix.Project
 
-  @version "0.14.3"
+  @version "1.0.0"
 
   def project do
     [
@@ -26,8 +26,7 @@ defmodule GenStage.Mixfile do
 
   defp deps do
     [
-      {:ex_doc, "~> 0.12", only: :docs},
-      {:inch_ex, ">= 0.4.0", only: :docs}
+      {:ex_doc, "~> 0.12", only: :docs}
     ]
   end
 

mix.lock

@@ -1,9 +1,9 @@
 %{
-  "earmark": {:hex, :earmark, "1.4.2", "3aa0bd23bc4c61cf2f1e5d752d1bb470560a6f8539974f767a38923bb20e1d7f", [:mix], [], "hexpm"},
-  "ex_doc": {:hex, :ex_doc, "0.21.2", "caca5bc28ed7b3bdc0b662f8afe2bee1eedb5c3cf7b322feeeb7c6ebbde089d6", [:mix], [{:earmark, "~> 1.3.3 or ~> 1.4", [hex: :earmark, repo: "hexpm", optional: false]}, {:makeup_elixir, "~> 0.14", [hex: :makeup_elixir, repo: "hexpm", optional: false]}], "hexpm"},
+  "earmark": {:hex, :earmark, "1.4.3", "364ca2e9710f6bff494117dbbd53880d84bebb692dafc3a78eb50aa3183f2bfd", [:mix], [], "hexpm"},
+  "ex_doc": {:hex, :ex_doc, "0.21.3", "857ec876b35a587c5d9148a2512e952e24c24345552259464b98bfbb883c7b42", [:mix], [{:earmark, "~> 1.4", [hex: :earmark, repo: "hexpm", optional: false]}, {:makeup_elixir, "~> 0.14", [hex: :makeup_elixir, repo: "hexpm", optional: false]}], "hexpm"},
   "inch_ex": {:hex, :inch_ex, "0.5.5", "b63f57e281467bd3456461525fdbc9e158c8edbe603da6e3e4671befde796a3d", [:mix], [{:poison, "~> 1.5 or ~> 2.0 or ~> 3.0", [hex: :poison, repo: "hexpm", optional: false]}], "hexpm"},
   "makeup": {:hex, :makeup, "1.0.0", "671df94cf5a594b739ce03b0d0316aa64312cee2574b6a44becb83cd90fb05dc", [:mix], [{:nimble_parsec, "~> 0.5.0", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm"},
   "makeup_elixir": {:hex, :makeup_elixir, "0.14.0", "cf8b7c66ad1cff4c14679698d532f0b5d45a3968ffbcbfd590339cb57742f1ae", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}], "hexpm"},
-  "nimble_parsec": {:hex, :nimble_parsec, "0.5.1", "c90796ecee0289dbb5ad16d3ad06f957b0cd1199769641c961cfe0b97db190e0", [:mix], [], "hexpm"},
+  "nimble_parsec": {:hex, :nimble_parsec, "0.5.3", "def21c10a9ed70ce22754fdeea0810dafd53c2db3219a0cd54cf5526377af1c6", [:mix], [], "hexpm"},
   "poison": {:hex, :poison, "3.0.0", "625ebd64d33ae2e65201c2c14d6c85c27cc8b68f2d0dd37828fde9c6920dd131", [:mix], [], "hexpm"},
 }

test/gen_stage/partition_dispatcher_test.exs

@@ -76,6 +76,33 @@ defmodule GenStage.PartitionDispatcherTest do
     assert_received {:"$gen_consumer", {_, ^ref}, [5, 7, 9, 11, 13]}
   end
 
+  test "subscribes, asks and dispatches to partitions or none" do
+    pid = self()
+    even_ref = make_ref()
+    odd_ref = make_ref()
+
+    hash_fun = fn event ->
+      cond do
+        rem(event, 3) == 0 -> :none
+        rem(event, 2) == 0 -> {event, :even}
+        true -> {event, :odd}
+      end
+    end
+
+    disp = dispatcher(partitions: [:odd, :even], hash: hash_fun)
+
+    {:ok, 0, disp} = D.subscribe([partition: :even], {pid, even_ref}, disp)
+    {:ok, 0, disp} = D.subscribe([partition: :odd], {pid, odd_ref}, disp)
+
+    {:ok, 4, disp} = D.ask(4, {pid, even_ref}, disp)
+    {:ok, 4, disp} = D.ask(4, {pid, odd_ref}, disp)
+    {:ok, [12], disp} = D.dispatch([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12], 12, disp)
+
+    assert_received {:"$gen_consumer", {_, ^even_ref}, [2, 4, 8, 10]}
+    assert_received {:"$gen_consumer", {_, ^odd_ref}, [1, 5, 7, 11]}
+    assert {0, 0} = waiting_and_pending(disp)
+  end
+
   test "buffers events before subscription" do
     disp = dispatcher(partitions: 2)
 
@@ -195,17 +222,17 @@ defmodule GenStage.PartitionDispatcherTest do
     assert_received :hello
   end
 
-  test "logs on unknown partition" do
+  test "raises on unknown partition" do
     pid = self()
     ref = make_ref()
     disp = dispatcher(partitions: [:foo, :bar], hash: &{&1, :oops})
 
     {:ok, 0, disp} = D.subscribe([partition: :foo], {pid, ref}, disp)
     {:ok, 3, disp} = D.ask(3, {pid, ref}, disp)
 
-    error = ExUnit.CaptureLog.capture_log(fn -> D.dispatch([1, 2, 5], 3, disp) end)
-    assert error =~ "[error] Unknown partition :oops computed for GenStage/Flow event 1"
-    assert error =~ "The known partitions are [:bar, :foo]"
+    assert_raise RuntimeError, ~r/unknown partition :oops/, fn ->
+      D.dispatch([1, 2, 5], 3, disp)
+    end
   end
 
   test "errors if the :hash function returns a bad value" do