Send events to dispatchers even if there are no consumers

If you are using the demand dispatcher (the default)
and a consumer leaves, any demand from this consumer
is stored as pending. The idea is, if a consumer asks
for 10 events, and then leaves, the producer was already
notified of those 10 events. If another consumer joins,
then those previously asked 10 events should be sent to
next consumer.

However, if a consumer left and there were no additional
consumers, any event sent by the producer would be
buffered immediately, and the dispatcher would never know
its pending demand was satisfied.

This would also happen in cases you had 2 consumers,
both asked for 5 events, and then one left. The producer
could send 8 events, and the dispatcher would never take
into account those 3 additional events were satisfying
a pending demand.

This patch addresses both issues.

Closes #311.
Closes #312.

Author: josevalim

lib/gen_stage.ex

@@ -2415,11 +2415,6 @@ defmodule GenStage do
     stage
   end
 
-  defp dispatch_events(events, _length, %{consumers: consumers} = stage)
-       when map_size(consumers) == 0 do
-    buffer_events(events, stage)
-  end
-
   defp dispatch_events(events, length, stage) do
     %{dispatcher_mod: dispatcher_mod, dispatcher_state: dispatcher_state} = stage
     {:ok, events, dispatcher_state} = dispatcher_mod.dispatch(events, length, dispatcher_state)

lib/gen_stage/dispatchers/demand_dispatcher.ex

@@ -78,16 +78,20 @@ defmodule GenStage.DemandDispatcher do
   end
 
   def dispatch(events, length, {demands, pending, max, false}) do
-    {events, demands} = dispatch_demand(events, length, demands)
-    {:ok, events, {demands, pending, max, false}}
+    {events, to_buffer, demands} = dispatch_demand(events, length, demands)
+    {:ok, events, {demands, max(pending - to_buffer, 0), max, false}}
   end
 
-  defp dispatch_demand([], _length, demands) do
-    {[], demands}
+  defp dispatch_demand([], length, demands) do
+    {[], length, demands}
   end
 
-  defp dispatch_demand(events, _length, [{0, _, _} | _] = demands) do
-    {events, demands}
+  defp dispatch_demand(events, length, []) do
+    {events, length, []}
+  end
+
+  defp dispatch_demand(events, length, [{0, _, _} | _] = demands) do
+    {events, length, demands}
   end
 
   defp dispatch_demand(events, length, [{counter, pid, ref} | demands]) do

test/gen_stage/broadcast_dispatcher_test.exs

@@ -35,6 +35,10 @@ defmodule GenStage.BroadcastDispatcherTest do
 
     {:ok, 0, disp} = D.cancel({pid, ref}, disp)
     assert disp == {[], 0, MapSet.new()}
+
+    # Now attempt to dispatch with no consumers
+    {:ok, [1, 2, 3], disp} = D.dispatch([1, 2, 3], 3, disp)
+    assert disp == {[], 0, MapSet.new()}
   end
 
   test "multiple subscriptions with early demand" do

test/gen_stage/demand_dispatcher_test.exs

@@ -49,6 +49,35 @@ defmodule GenStage.DemandDispatcherTest do
 
     {:ok, 0, disp} = D.cancel({pid, ref}, disp)
     assert disp == {[], 10, 10, @default_shuffle_flag}
+
+    # Now attempt to dispatch with no consumers
+    {:ok, [:a, :b, :c], disp} = D.dispatch([:a, :b, :c], 3, disp)
+    assert disp == {[], 7, 10, @default_shuffle_flag}
+
+    {:ok, _, disp} = D.dispatch(Enum.to_list(1..10), 10, disp)
+    assert disp == {[], 0, 10, @default_shuffle_flag}
+  end
+
+  test "subscribes twice, asks twice and cancel one" do
+    pid = self()
+    ref1 = make_ref()
+    ref2 = make_ref()
+    disp = dispatcher([])
+
+    # Subscribe twice and ask twice
+    {:ok, 0, disp} = D.subscribe([], {pid, ref1}, disp)
+    {:ok, 3, disp} = D.ask(3, {pid, ref1}, disp)
+    {:ok, 0, disp} = D.subscribe([], {pid, ref2}, disp)
+    {:ok, 3, disp} = D.ask(3, {pid, ref2}, disp)
+    assert disp == {[{3, pid, ref1}, {3, pid, ref2}], 0, 3, @default_shuffle_flag}
+
+    # Cancel and leave some demand
+    {:ok, 0, disp} = D.cancel({pid, ref1}, disp)
+    assert disp == {[{3, pid, ref2}], 3, 3, @default_shuffle_flag}
+
+    # Send events for both subscriptions
+    {:ok, [:d, :e], disp} = D.dispatch([:a, :b, :c, :d, :e], 5, disp)
+    assert disp == {[{0, pid, ref2}], 1, 3, @default_shuffle_flag}
   end
 
   test "subscribes, asks and dispatches" do

test/gen_stage/partition_dispatcher_test.exs

@@ -30,6 +30,10 @@ defmodule GenStage.PartitionDispatcherTest do
     assert {10, 5} = waiting_and_pending(disp)
     {:ok, 0, disp} = D.cancel({pid, ref}, disp)
     assert {10, 10} = waiting_and_pending(disp)
+
+    # Now attempt to dispatch with no consumers
+    {:ok, [], disp} = D.dispatch([0, 1, 2, 3], 4, disp)
+    assert {6, 10} = waiting_and_pending(disp)
   end
 
   test "subscribes, asks and dispatches" do