decouple map from observable (#281)

* decouple map from observable

* fix msvc compilation error

9 files changed, 193 insertions, 6 deletions

diff --git a/Rx/v2/test/operators/concat_map.cpp b/Rx/v2/test/operators/concat_map.cpp
index 531a638..2e891dc 100644
--- a/Rx/v2/test/operators/concat_map.cpp
+++ b/Rx/v2/test/operators/concat_map.cpp
@@ -1,6 +1,7 @@
 #include "../test.h"
 #include <rxcpp/operators/rx-reduce.hpp>
 #include <rxcpp/operators/rx-filter.hpp>
+#include <rxcpp/operators/rx-map.hpp>
 
 static const int static_tripletCount = 100;
 
diff --git a/Rx/v2/test/operators/flat_map.cpp b/Rx/v2/test/operators/flat_map.cpp
index 55af15b..e90b3ea 100644
--- a/Rx/v2/test/operators/flat_map.cpp
+++ b/Rx/v2/test/operators/flat_map.cpp
@@ -1,6 +1,7 @@
 #include "../test.h"
 #include <rxcpp/operators/rx-reduce.hpp>
 #include <rxcpp/operators/rx-filter.hpp>
+#include <rxcpp/operators/rx-map.hpp>
 
 static const int static_tripletCount = 100;
 
diff --git a/Rx/v2/test/operators/group_by.cpp b/Rx/v2/test/operators/group_by.cpp
index a4a3c3b..8363de7 100644
--- a/Rx/v2/test/operators/group_by.cpp
+++ b/Rx/v2/test/operators/group_by.cpp
@@ -1,6 +1,7 @@
 #include "../test.h"
 #include <rxcpp/operators/rx-group_by.hpp>
 #include <rxcpp/operators/rx-reduce.hpp>
+#include <rxcpp/operators/rx-map.hpp>
 
 #include <locale>
 
diff --git a/Rx/v2/test/operators/map.cpp b/Rx/v2/test/operators/map.cpp
index d672d7e..c96cc8d 100644
--- a/Rx/v2/test/operators/map.cpp
+++ b/Rx/v2/test/operators/map.cpp
@@ -1,7 +1,8 @@
 #include "../test.h"
+#include <rxcpp/operators/rx-map.hpp>
 
-SCENARIO("map stops on completion", "[map][operators]"){
-    GIVEN("a test hot observable of ints"){
+SCENARIO("map stops on completion", "[map][operators]") {
+    GIVEN("a test hot observable of ints") {
         auto sc = rxsc::make_test();
         auto w = sc.create_worker();
         const rxsc::test::messages<int> on;
@@ -19,7 +20,7 @@ SCENARIO("map stops on completion", "[map][operators]"){
             on.error(430, std::runtime_error("error on unsubscribed stream"))
         });
 
-        WHEN("mapped to ints that are one larger"){
+        WHEN("mapped to ints that are one larger") {
 
             auto res = w.start(
                 [xs, &invoked]() {
@@ -33,7 +34,7 @@ SCENARIO("map stops on completion", "[map][operators]"){
                 }
             );
 
-            THEN("the output stops on completion"){
+            THEN("the output stops on completion") {
                 auto required = rxu::to_vector({
                     on.next(210, 3),
                     on.next(240, 4),
@@ -45,7 +46,7 @@ SCENARIO("map stops on completion", "[map][operators]"){
                 REQUIRE(required == actual);
             }
 
-            THEN("there was one subscription and one unsubscription"){
+            THEN("there was one subscription and one unsubscription") {
                 auto required = rxu::to_vector({
                     on.subscribe(200, 400)
                 });
@@ -53,9 +54,187 @@ SCENARIO("map stops on completion", "[map][operators]"){
                 REQUIRE(required == actual);
             }
 
-            THEN("map was called until completed"){
+            THEN("map was called until completed") {
                 REQUIRE(4 == invoked);
             }
         }
     }
 }
+
+SCENARIO("map - never", "[map][operators]") {
+    GIVEN("a source") {
+        auto sc = rxsc::make_test();
+        auto w = sc.create_worker();
+        const rxsc::test::messages<int> on;
+
+        auto xs = sc.make_hot_observable({
+            on.next(150, 1)
+        });
+
+        WHEN("values are mapped") {
+
+            auto res = w.start(
+                [xs]() {
+                    return xs
+                        | rxo::map([](int x) {
+                            return x + 1;
+                        })
+                        // forget type to workaround lambda deduction bug on msvc 2013
+                        | rxo::as_dynamic();
+                }
+            );
+
+            THEN("the output is empty") {
+                auto required = std::vector<rxsc::test::messages<int>::recorded_type>();
+                auto actual = res.get_observer().messages();
+                REQUIRE(required == actual);
+            }
+
+            THEN("there was 1 subscription/unsubscription to the source") {
+                auto required = rxu::to_vector({
+                    on.subscribe(200, 1000)
+                });
+                auto actual = xs.subscriptions();
+                REQUIRE(required == actual);
+            }
+        }
+    }
+}
+
+SCENARIO("map - empty", "[map][operators]") {
+    GIVEN("a source") {
+        auto sc = rxsc::make_test();
+        auto w = sc.create_worker();
+        const rxsc::test::messages<int> on;
+
+        auto xs = sc.make_hot_observable({
+            on.next(150, 1),
+            on.completed(250)
+        });
+
+        WHEN("values are mapped") {
+
+            auto res = w.start(
+                [xs]() {
+                    return xs
+                        .map([](int x) {
+                            return x + 1;
+                        })
+                        // forget type to workaround lambda deduction bug on msvc 2013
+                        .as_dynamic();
+                }
+            );
+
+            THEN("the output only contains complete message") {
+                auto required = rxu::to_vector({
+                    on.completed(250)
+                });
+                auto actual = res.get_observer().messages();
+                REQUIRE(required == actual);
+            }
+
+            THEN("there was 1 subscription/unsubscription to the source") {
+                auto required = rxu::to_vector({
+                    on.subscribe(200, 250)
+                });
+                auto actual = xs.subscriptions();
+                REQUIRE(required == actual);
+            }
+
+        }
+    }
+}
+
+SCENARIO("map - items emitted", "[map][operators]") {
+    GIVEN("a source") {
+        auto sc = rxsc::make_test();
+        auto w = sc.create_worker();
+        const rxsc::test::messages<int> on;
+
+        auto xs = sc.make_hot_observable({
+            on.next(150, 1),
+            on.next(210, 2),
+            on.next(240, 3),
+            on.completed(300)
+        });
+
+        WHEN("values are mapped") {
+
+            auto res = w.start(
+                [xs]() {
+                    return xs
+                        .map([](int x) {
+                            return x + 1;
+                        })
+                        // forget type to workaround lambda deduction bug on msvc 2013
+                        .as_dynamic();
+                }
+            );
+
+            THEN("the output only contains items sent while subscribed") {
+                auto required = rxu::to_vector({
+                    on.next(210, 3),
+                    on.next(240, 4),
+                    on.completed(300)
+                });
+                auto actual = res.get_observer().messages();
+                REQUIRE(required == actual);
+            }
+
+            THEN("there was 1 subscription/unsubscription to the source") {
+                auto required = rxu::to_vector({
+                    on.subscribe(200, 300)
+                });
+                auto actual = xs.subscriptions();
+                REQUIRE(required == actual);
+            }
+
+        }
+    }
+}
+
+SCENARIO("map - throw", "[map][operators]") {
+    GIVEN("a source") {
+        auto sc = rxsc::make_test();
+        auto w = sc.create_worker();
+        const rxsc::test::messages<int> on;
+
+        std::runtime_error ex("map on_error from source");
+
+        auto xs = sc.make_hot_observable({
+            on.next(150, 1),
+            on.error(250, ex)
+        });
+
+        WHEN("values are mapped") {
+
+            auto res = w.start(
+                [xs]() {
+                    return xs
+                        .map([](int x) {
+                            return x + 1;
+                        })
+                        // forget type to workaround lambda deduction bug on msvc 2013
+                        .as_dynamic();
+                }
+            );
+
+            THEN("the output only contains only error") {
+                auto required = rxu::to_vector({
+                    on.error(250, ex)
+                });
+                auto actual = res.get_observer().messages();
+                REQUIRE(required == actual);
+            }
+
+            THEN("there was 1 subscription/unsubscription to the source") {
+                auto required = rxu::to_vector({
+                    on.subscribe(200, 250)
+                });
+                auto actual = xs.subscriptions();
+                REQUIRE(required == actual);
+            }
+
+        }
+    }
+}
diff --git a/Rx/v2/test/operators/scan.cpp b/Rx/v2/test/operators/scan.cpp
index b7a7276..430d971 100644
--- a/Rx/v2/test/operators/scan.cpp
+++ b/Rx/v2/test/operators/scan.cpp
@@ -1,4 +1,5 @@
 #include "../test.h"
+#include <rxcpp/operators/rx-map.hpp>
 
 SCENARIO("scan: issue 41", "[scan][operators][issue][hide]"){
     GIVEN("map of scan of interval"){
diff --git a/Rx/v2/test/operators/subscribe_on.cpp b/Rx/v2/test/operators/subscribe_on.cpp
index 30d4cdc..80e9b2c 100644
--- a/Rx/v2/test/operators/subscribe_on.cpp
+++ b/Rx/v2/test/operators/subscribe_on.cpp
@@ -1,5 +1,6 @@
 #include "../test.h"
 #include <rxcpp/operators/rx-reduce.hpp>
+#include <rxcpp/operators/rx-map.hpp>
 
 static const int static_subscriptions = 50000;
 
diff --git a/Rx/v2/test/operators/take_until.cpp b/Rx/v2/test/operators/take_until.cpp
index e821f81..92e2fa8 100644
--- a/Rx/v2/test/operators/take_until.cpp
+++ b/Rx/v2/test/operators/take_until.cpp
@@ -1,4 +1,5 @@
 #include "../test.h"
+#include <rxcpp/operators/rx-map.hpp>
 
 SCENARIO("take_until trigger on_next", "[take_until][take][operators]"){
     GIVEN("2 sources"){
diff --git a/Rx/v2/test/operators/window.cpp b/Rx/v2/test/operators/window.cpp
index 30f109e..8d022df 100644
--- a/Rx/v2/test/operators/window.cpp
+++ b/Rx/v2/test/operators/window.cpp
@@ -1,6 +1,7 @@
 #include "../test.h"
 
 #include <rxcpp/operators/rx-reduce.hpp>
+#include <rxcpp/operators/rx-map.hpp>
 
 SCENARIO("window count, basic", "[window][operators]"){
     GIVEN("1 hot observable of ints."){
diff --git a/Rx/v2/test/operators/window_toggle.cpp b/Rx/v2/test/operators/window_toggle.cpp
index 0c1cc9f..08dc210 100644
--- a/Rx/v2/test/operators/window_toggle.cpp
+++ b/Rx/v2/test/operators/window_toggle.cpp
@@ -1,4 +1,5 @@
 #include "../test.h"
+#include <rxcpp/operators/rx-map.hpp>
 
 SCENARIO("window toggle, basic", "[window_toggle][operators]"){
     GIVEN("1 hot observable of ints and hot observable of opens."){