Check the return value of the :hash function (#251)

Author: whatyouhide

lib/gen_stage/dispatchers/partition_dispatcher.ex

@@ -20,9 +20,9 @@ defmodule GenStage.PartitionDispatcher do
       named `:odd` and `:even`.
 
     * `:hash` - the hashing algorithm, which receives the event and returns
-      a tuple with two elements, the event to be dispatched as first argument 
+      a tuple with two elements, the event to be dispatched as first argument
       and the partition as second. The partition must be one of the partitions
-      specified in `:partitions` above. The default uses 
+      specified in `:partitions` above. The default uses
       `fn event -> {event, :erlang.phash2(event, Enum.count(partitions))} end`
       on the event to select the partition.
 
@@ -242,7 +242,14 @@ defmodule GenStage.PartitionDispatcher do
     {deliver_now, deliver_later, waiting} = split_events(events, waiting, [])
 
     for event <- deliver_now do
-      {event, partition} = hash.(event)
+      {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 ->

test/gen_stage/partition_dispatcher_test.exs

@@ -208,6 +208,18 @@ defmodule GenStage.PartitionDispatcherTest do
     assert error =~ "The known partitions are [:bar, :foo]"
   end
 
+  test "errors if the :hash function returns a bad value" do
+    pid = self()
+    ref = make_ref()
+    disp = dispatcher(partitions: [:foo, :bar], hash: fn _ -> :not_a_tuple end)
+    {:ok, 0, disp} = D.subscribe([partition: :foo], {pid, ref}, disp)
+    {:ok, 3, disp} = D.ask(3, {pid, ref}, disp)
+
+    assert_raise RuntimeError, ~r/the :hash function should return/, fn ->
+      D.dispatch([1, 2, 5], 3, disp)
+    end
+  end
+
   test "errors on init" do
     assert_raise ArgumentError, ~r/the enumerable of :partitions is required/, fn ->
       dispatcher([])