Throw an exception when reducing an empty blocking_observable.

1 files changed, 7 insertions, 3 deletions

diff --git a/Rx/v2/src/rxcpp/rx-observable.hpp b/Rx/v2/src/rxcpp/rx-observable.hpp
index 7d048f6..5ed5ec0 100644
--- a/Rx/v2/src/rxcpp/rx-observable.hpp
+++ b/Rx/v2/src/rxcpp/rx-observable.hpp
@@ -281,19 +281,23 @@ public:
         rxu::maybe<T> result;
         composite_subscription cs;
         subscribe(cs, [&](T v){result.reset(v); cs.unsubscribe();});
-        if (result.empty()) abort();
+        if (result.empty()) throw std::runtime_error("No elements");
         return result.get();
     }
 
     T last() const {
         rxu::maybe<T> result;
         subscribe([&](T v){result.reset(v);});
-        if (result.empty()) abort();
+        if (result.empty()) throw std::runtime_error("No elements");
         return result.get();
     }
 
     int count() const {
-        return source.count().as_blocking().last();
+        rxu::maybe<T> result;
+        source.count().as_blocking().subscribe(
+            [&](T v){result.reset(v);},
+            [](std::exception_ptr){result.reset(0);});
+        return result.get();
     }
     T sum() const {
         return source.sum().as_blocking().last();