From fba0f2d04f8d6ac53994a994353f6ae3cb6a2219 Mon Sep 17 00:00:00 2001
From: Kirk Shoop <kirk.shoop@microsoft.com>
Date: Tue, 20 May 2014 08:35:26 -0700
Subject: add synchronize subject and fix lifetime issues

This finally forced a rewrite of subscription. pretty happy with the
result.
flat map removed all sync and takes a new parameter that is applied to
every source and can be used to synchronize them
also added multicast
---
 Rx/v2/src/rxcpp/rx-observer.hpp | 25 +++++++++++++++++++++++++
 1 file changed, 25 insertions(+)

(limited to 'Rx/v2/src/rxcpp/rx-observer.hpp')

diff --git a/Rx/v2/src/rxcpp/rx-observer.hpp b/Rx/v2/src/rxcpp/rx-observer.hpp
index 6b573ae..b6bad50 100644
--- a/Rx/v2/src/rxcpp/rx-observer.hpp
+++ b/Rx/v2/src/rxcpp/rx-observer.hpp
@@ -366,6 +366,31 @@ auto make_observer_dynamic(OnNext&& on, OnError&& oe, OnCompleted&& oc)
                         dynamic_observer<T>(make_observer<T>(std::forward<OnNext>(on), std::forward<OnError>(oe), std::forward<OnCompleted>(oc))));
 }
 
+
+template<class F, class OnError>
+auto on_exception(const F& f, const OnError& c)
+    ->  typename std::enable_if<detail::is_on_error<OnError>::value, rxu::detail::maybe<decltype(f())>>::type {
+    rxu::detail::maybe<decltype(f())> r;
+    try {
+        r.reset(f());
+    } catch (...) {
+        c(std::current_exception());
+    }
+    return r;
+}
+
+template<class F, class Subscriber>
+auto on_exception(const F& f, const Subscriber& s)
+    ->  typename std::enable_if<is_subscriber<Subscriber>::value, rxu::detail::maybe<decltype(f())>>::type {
+    rxu::detail::maybe<decltype(f())> r;
+    try {
+        r.reset(f());
+    } catch (...) {
+        s.on_error(std::current_exception());
+    }
+    return r;
+}
+
 }
 
 #endif
-- 
cgit v1.2.3