 
   defp stream_deliver({:suspend, acc}, max, spawned, delivered, waiting, next,
                       reducer, monitor_pid, monitor_ref, timeout) do
-    {:suspended, acc, &stream_deliver(&1, max, spawned, delivered, waiting, next,
-                                      reducer, monitor_pid, monitor_ref, timeout)}
+    continuation = &stream_deliver(&1, max, spawned, delivered, waiting, next, reducer, monitor_pid, monitor_ref, timeout)
+    {:suspended, acc, continuation}
   end
   defp stream_deliver({:halt, acc}, max, spawned, delivered, waiting, next,
                       reducer, monitor_pid, monitor_ref, timeout) do

    # about our reference to it.
    send(monitor_pid, {parent, monitor_ref})

    stream_reduce(acc, max_concurrency, 0, 0, %{}, next,
                  reducer, monitor_pid, monitor_ref, timeout)
  end


        {{counter, _, _}, counters} = Map.pop(counters, ref)
        send(parent_pid, {:down, {monitor_ref, counter}, reason})
        stream_monitor_loop(parent_pid, parent_ref, mfa, spawn, monitor_ref, counters)

      {:EXIT, _, _} ->
        stream_monitor_loop(parent_pid, parent_ref, mfa, spawn, monitor_ref, counters)
    end
