Implement new behavior of reduce operator

1. Reducing an empty observable returns result_selector(seed).
2. Exceptions from accumulator() and result_selector() are caught and emitted as on_error().
3. first(), last(), sum(), average() emit rxcpp::empty_error when applied to an empty observable.
4. count() of an empty observable emits 0.
5. If the source observable for count(), sum(), average() calls on_error(), the resulting observable behaves like it was on_completed().

3 files changed, 120 insertions, 13 deletions

diff --git a/Rx/v2/examples/doxygen/blocking_observable.cpp b/Rx/v2/examples/doxygen/blocking_observable.cpp
index eb3dc30..2ed5526 100644
--- a/Rx/v2/examples/doxygen/blocking_observable.cpp
+++ b/Rx/v2/examples/doxygen/blocking_observable.cpp
@@ -17,7 +17,7 @@ SCENARIO("blocking first empty sample"){
     try {
         auto first = values.first();
         printf("first = %d\n", first);
-    } catch (const std::exception& ex) {
+    } catch (const rxcpp::empty_error& ex) {
         printf("Exception: %s\n", ex.what());
     }
     printf("//! [blocking first empty sample]\n");
@@ -37,7 +37,7 @@ SCENARIO("blocking last empty sample"){
     try {
         auto last = values.last();
         printf("last = %d\n", last);
-    } catch (const std::exception& ex) {
+    } catch (const rxcpp::empty_error& ex) {
         printf("Exception: %s\n", ex.what());
     }
     printf("//! [blocking last empty sample]\n");
@@ -51,6 +51,16 @@ SCENARIO("blocking count sample"){
     printf("//! [blocking count sample]\n");
 }
 
+SCENARIO("blocking count error sample"){
+    printf("//! [blocking count error sample]\n");
+    auto values = rxcpp::observable<>::range(1, 3).
+        concat(rxcpp::observable<>::error<int>(std::runtime_error("Error from source"))).
+        as_blocking();
+    auto count = values.count();
+    printf("count = %d\n", count);
+    printf("//! [blocking count error sample]\n");
+}
+
 SCENARIO("blocking sum sample"){
     printf("//! [blocking sum sample]\n");
     auto values = rxcpp::observable<>::range(1, 3).as_blocking();
@@ -65,17 +75,27 @@ SCENARIO("blocking sum empty sample"){
     try {
         auto sum = values.sum();
         printf("sum = %d\n", sum);
-    } catch (const std::exception& ex) {
+    } catch (const rxcpp::empty_error& ex) {
         printf("Exception: %s\n", ex.what());
     }
     printf("//! [blocking sum empty sample]\n");
 }
 
+SCENARIO("blocking sum error sample"){
+    printf("//! [blocking sum error sample]\n");
+    auto values = rxcpp::observable<>::range(1, 3).
+        concat(rxcpp::observable<>::error<int>(std::runtime_error("Error from source"))).
+        as_blocking();
+    auto sum = values.sum();
+    printf("sum = %d\n", sum);
+    printf("//! [blocking sum error sample]\n");
+}
+
 SCENARIO("blocking average sample"){
     printf("//! [blocking average sample]\n");
-    auto values = rxcpp::observable<>::range(1, 3).as_blocking();
+    auto values = rxcpp::observable<>::range(1, 4).as_blocking();
     auto average = values.average();
-    printf("average = %d\n", average);
+    printf("average = %lf\n", average);
     printf("//! [blocking average sample]\n");
 }
 
@@ -84,9 +104,19 @@ SCENARIO("blocking average empty sample"){
     auto values = rxcpp::observable<>::empty<int>().as_blocking();
     try {
         auto average = values.average();
-        printf("average = %d\n", average);
-    } catch (const std::exception& ex) {
+        printf("average = %lf\n", average);
+    } catch (const rxcpp::empty_error& ex) {
         printf("Exception: %s\n", ex.what());
     }
     printf("//! [blocking average empty sample]\n");
 }
+
+SCENARIO("blocking average error sample"){
+    printf("//! [blocking average error sample]\n");
+    auto values = rxcpp::observable<>::range(1, 4).
+        concat(rxcpp::observable<>::error<int>(std::runtime_error("Error from source"))).
+        as_blocking();
+    auto average = values.average();
+    printf("average = %lf\n", average);
+    printf("//! [blocking average error sample]\n");
+}
diff --git a/Rx/v2/examples/doxygen/math.cpp b/Rx/v2/examples/doxygen/math.cpp
index bc782c0..3e612c0 100644
--- a/Rx/v2/examples/doxygen/math.cpp
+++ b/Rx/v2/examples/doxygen/math.cpp
@@ -21,7 +21,7 @@ SCENARIO("first empty sample"){
             [](int v){printf("OnNext: %d\n", v);},
             [](std::exception_ptr ep){
                 try {std::rethrow_exception(ep);}
-                catch (const std::exception& ex) {
+                catch (const rxcpp::empty_error& ex) {
                     printf("OnError: %s\n", ex.what());
                 }
             },
@@ -47,7 +47,7 @@ SCENARIO("last empty sample"){
             [](int v){printf("OnNext: %d\n", v);},
             [](std::exception_ptr ep){
                 try {std::rethrow_exception(ep);}
-                catch (const std::exception& ex) {
+                catch (const rxcpp::empty_error& ex) {
                     printf("OnError: %s\n", ex.what());
                 }
             },
@@ -65,6 +65,18 @@ SCENARIO("count sample"){
     printf("//! [count sample]\n");
 }
 
+SCENARIO("count error sample"){
+    printf("//! [count error sample]\n");
+    auto values = rxcpp::observable<>::range(1, 3).
+        concat(rxcpp::observable<>::error<int>(std::runtime_error("Error from source"))).
+        count();
+    values.
+        subscribe(
+            [](int v){printf("OnNext: %d\n", v);},
+            [](){printf("OnCompleted\n");});
+    printf("//! [count error sample]\n");
+}
+
 SCENARIO("sum sample"){
     printf("//! [sum sample]\n");
     auto values = rxcpp::observable<>::range(1, 3).sum();
@@ -83,7 +95,7 @@ SCENARIO("sum empty sample"){
             [](int v){printf("OnNext: %d\n", v);},
             [](std::exception_ptr ep){
                 try {std::rethrow_exception(ep);}
-                catch (const std::exception& ex) {
+                catch (const rxcpp::empty_error& ex) {
                     printf("OnError: %s\n", ex.what());
                 }
             },
@@ -91,6 +103,18 @@ SCENARIO("sum empty sample"){
     printf("//! [sum empty sample]\n");
 }
 
+SCENARIO("sum error sample"){
+    printf("//! [sum error sample]\n");
+    auto values = rxcpp::observable<>::range(1, 3).
+        concat(rxcpp::observable<>::error<int>(std::runtime_error("Error from source"))).
+        sum();
+    values.
+        subscribe(
+            [](int v){printf("OnNext: %d\n", v);},
+            [](){printf("OnCompleted\n");});
+    printf("//! [sum error sample]\n");
+}
+
 SCENARIO("average sample"){
     printf("//! [average sample]\n");
     auto values = rxcpp::observable<>::range(1, 4).average();
@@ -109,10 +133,22 @@ SCENARIO("average empty sample"){
             [](double v){printf("OnNext: %lf\n", v);},
             [](std::exception_ptr ep){
                 try {std::rethrow_exception(ep);}
-                catch (const std::exception& ex) {
+                catch (const rxcpp::empty_error& ex) {
                     printf("OnError: %s\n", ex.what());
                 }
             },
             [](){printf("OnCompleted\n");});
     printf("//! [average empty sample]\n");
 }
+
+SCENARIO("average error sample"){
+    printf("//! [average error sample]\n");
+    auto values = rxcpp::observable<>::range(1, 4).
+        concat(rxcpp::observable<>::error<int>(std::runtime_error("Error from source"))).
+        average();
+    values.
+        subscribe(
+            [](double v){printf("OnNext: %lf\n", v);},
+            [](){printf("OnCompleted\n");});
+    printf("//! [average error sample]\n");
+}
diff --git a/Rx/v2/examples/doxygen/reduce.cpp b/Rx/v2/examples/doxygen/reduce.cpp
index 9035e5e..d79e6a1 100644
--- a/Rx/v2/examples/doxygen/reduce.cpp
+++ b/Rx/v2/examples/doxygen/reduce.cpp
@@ -28,7 +28,25 @@ SCENARIO("reduce empty sample"){
     auto values = rxcpp::observable<>::empty<int>().
         reduce(
             1,
-            [](int seed, int){return seed;},
+            [](int,int){return 0;},
+            [](int res){return res;});
+    values.
+        subscribe(
+            [](int v){printf("OnNext: %d\n", v);},
+            [](){printf("OnCompleted\n");});
+    printf("//! [reduce empty sample]\n");
+}
+
+SCENARIO("reduce exception from accumulator sample"){
+    printf("//! [reduce exception from accumulator sample] <<<<<<<<<<<<<<<<<<<<<<<<<<<<<\n");
+    auto values = rxcpp::observable<>::range(1, 3).
+        reduce(
+            0,
+            [](int seed, int v){
+                if (v == 2)
+                    throw std::runtime_error("Exception from accumulator");
+                return seed;
+            },
             [](int res){return res;});
     values.
         subscribe(
@@ -40,5 +58,28 @@ SCENARIO("reduce empty sample"){
                 }
             },
             [](){printf("OnCompleted\n");});
-    printf("//! [reduce empty sample]\n");
+    printf("//! [reduce exception from accumulator sample]\n");
+}
+
+SCENARIO("reduce exception from result selector sample"){
+    printf("//! [reduce exception from result selector sample]\n");
+    auto values = rxcpp::observable<>::range(1, 3).
+        reduce(
+            0,
+            [](int seed, int v){return seed + v;},
+            [](int res){
+                throw std::runtime_error("Exception from result selector");
+                return res;
+            });
+    values.
+        subscribe(
+            [](int v){printf("OnNext: %d\n", v);},
+            [](std::exception_ptr ep){
+                try {std::rethrow_exception(ep);}
+                catch (const std::exception& ex) {
+                    printf("OnError: %s\n", ex.what());
+                }
+            },
+            [](){printf("OnCompleted\n");});
+    printf("//! [reduce exception from result selector sample]\n");
 }