 
   defp stream_reduce({:suspend, acc}, max, spawned, delivered, waiting, next,
                      reducer, monitor_pid, monitor_ref, timeout) do
-    {:suspended, acc, &stream_reduce(&1, max, spawned, delivered, waiting, next,
-                                     reducer, monitor_pid, monitor_ref, timeout)}
-  end
-
-  # All spawned, all delivered, next is done.
+    continuation = &stream_reduce(&1, max, spawned, delivered, waiting, next, reducer, monitor_pid, monitor_ref, timeout)
+    {:suspended, acc, continuation}
+  end
+
+  # All spawned, all delivered, next is :done.
   defp stream_reduce({:cont, acc}, _max, spawned, delivered, _waiting, next,
                      _reducer, monitor_pid, monitor_ref, timeout)
        when spawned == delivered and next == :done do


  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)}
  end
  defp stream_deliver({:halt, acc}, max, spawned, delivered, waiting, next,
                      reducer, monitor_pid, monitor_ref, timeout) do

    end
  end

  # This function spawns a task for the given "value", and puts the pid of this
  # new task in the map of "waiting" tasks, which is returned.
  defp stream_spawn(value, spawned, waiting, monitor_pid, monitor_ref, timeout) do
