decouple skip_until from observable

1 files changed, 55 insertions, 2 deletions

diff --git a/Rx/v2/test/operators/skip_until.cpp b/Rx/v2/test/operators/skip_until.cpp
index 697361b..3d5f408 100644
--- a/Rx/v2/test/operators/skip_until.cpp
+++ b/Rx/v2/test/operators/skip_until.cpp
@@ -1,4 +1,5 @@
 #include "../test.h"
+#include <rxcpp/operators/rx-skip_until.hpp>
 
 SCENARIO("skip_until, some data next", "[skip_until][skip][operators]"){
     GIVEN("2 sources"){
@@ -26,9 +27,9 @@ SCENARIO("skip_until, some data next", "[skip_until][skip][operators]"){
             auto res = w.start(
                 [&]() {
                     return l
-                        .skip_until(r)
+                        | rxo::skip_until(r)
                         // forget type to workaround lambda deduction bug on msvc 2013
-                        .as_dynamic();
+                        | rxo::as_dynamic();
                 }
             );
 
@@ -557,3 +558,55 @@ SCENARIO("skip_until, never never", "[skip_until][skip][operators]"){
         }
     }
 }
+
+SCENARIO("skip_until time point, some data next", "[skip_until][skip][operators]"){
+    GIVEN("2 sources"){
+        using clock_type = rxsc::detail::test_type::clock_type;
+
+        auto sc = rxsc::make_test();
+        auto so = rx::synchronize_in_one_worker(sc);
+        auto w = sc.create_worker();
+        const rxsc::test::messages<int> on;
+
+        auto l = sc.make_hot_observable({
+            on.next(150, 1),
+            on.next(210, 2),
+            on.next(220, 3),
+            on.next(230, 4),
+            on.next(240, 5),
+            on.completed(250)
+        });
+
+        clock_type::time_point t(std::chrono::milliseconds(225));
+
+        WHEN("invoked with a time point"){
+
+            auto res = w.start(
+                [&]() {
+                    return l
+                        | rxo::skip_until(t, so)
+                        // forget type to workaround lambda deduction bug on msvc 2013
+                        | rxo::as_dynamic();
+                }
+            );
+
+            THEN("the output only contains items sent while subscribed"){
+                auto required = rxu::to_vector({
+                    on.next(231, 4),
+                    on.next(241, 5),
+                    on.completed(251)
+                });
+                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 = l.subscriptions();
+                REQUIRE(required == actual);
+            }
+        }
+    }
+}
\ No newline at end of file