From 01c698def46b6e8a221d33b09f0fd6225fa8e9bf Mon Sep 17 00:00:00 2001
From: Tomasz Grabiec <tgrabiec@cloudius-systems.com>
Date: Fri, 3 Jul 2015 12:31:22 +0200
Subject: [PATCH] mutation_reader: Make consume() work with deferring callbacks
 too

---
 mutation_reader.hh | 9 +++++----
 1 file changed, 5 insertions(+), 4 deletions(-)

diff --git a/mutation_reader.hh b/mutation_reader.hh
index a912e77e50..6d15dfada8 100644
--- a/mutation_reader.hh
+++ b/mutation_reader.hh
@@ -36,15 +36,16 @@ mutation_reader make_empty_reader();
 template <typename Consumer>
 inline
 future<> consume(mutation_reader& reader, Consumer consumer) {
-    static_assert(std::is_same<stop_iteration, std::result_of_t<Consumer(mutation&&)>>::value, "bad Consumer signature");
+    static_assert(std::is_same<future<stop_iteration>, futurize_t<std::result_of_t<Consumer(mutation&&)>>>::value, "bad Consumer signature");
+    using futurator = futurize<std::result_of_t<Consumer(mutation&&)>>;
 
     return do_with(std::move(consumer), [&reader] (Consumer& c) -> future<> {
         return repeat([&reader, &c] () {
-            return reader().then([&c] (mutation_opt&& mo) {
+            return reader().then([&c] (mutation_opt&& mo) -> future<stop_iteration> {
                 if (!mo) {
-                    return stop_iteration::yes;
+                    return make_ready_future<stop_iteration>(stop_iteration::yes);
                 }
-                return c(std::move(*mo));
+                return futurator::apply(c, std::move(*mo));
             });
         });
     });