add element_at

1 files changed, 292 insertions, 0 deletions

diff --git a/Rx/v2/test/operators/element_at.cpp b/Rx/v2/test/operators/element_at.cpp
new file mode 100644
index 0000000..06c3be1
--- /dev/null
+++ b/Rx/v2/test/operators/element_at.cpp
@@ -0,0 +1,292 @@
+#include "../test.h"
+
+SCENARIO("element_at - never", "[element_at][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("element_at is taken"){
+
+            auto res = w.start(
+                [xs]() {
+                    return xs.element_at(3);
+                }
+            );
+
+            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("element_at - empty", "[element_at][operators]"){
+    GIVEN("a source"){
+        auto sc = rxsc::make_test();
+        auto w = sc.create_worker();
+        const rxsc::test::messages<int> on;
+
+        std::runtime_error ex("element_at on_error from source");
+
+        auto xs = sc.make_hot_observable({
+            on.next(150, 1),
+            on.completed(250)
+        });
+
+        WHEN("element_at is taken"){
+
+            auto res = w.start(
+                [xs]() {
+                    return xs.element_at(0);
+                }
+            );
+
+            THEN("the output only contains an 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);
+            }
+
+        }
+    }
+}
+
+SCENARIO("element_at - first", "[element_at][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.completed(250)
+        });
+
+        WHEN("element_at is taken"){
+
+            auto res = w.start(
+                [xs]() {
+                    return xs.element_at(0);
+                }
+            );
+
+            THEN("the output contains the first element"){
+                auto required = rxu::to_vector({
+                    on.next(210, 2),
+                    on.completed(210)
+                });
+                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, 210)
+                });
+                auto actual = xs.subscriptions();
+                REQUIRE(required == actual);
+            }
+
+        }
+    }
+}
+
+SCENARIO("element_at - throw", "[element_at][operators]"){
+    GIVEN("a source"){
+        auto sc = rxsc::make_test();
+        auto w = sc.create_worker();
+        const rxsc::test::messages<int> on;
+
+        std::runtime_error ex("element_at on_error from source");
+
+        auto xs = sc.make_hot_observable({
+            on.next(150, 1),
+            on.error(250, ex)
+        });
+
+        WHEN("element_at is taken"){
+
+            auto res = w.start(
+                [xs]() {
+                    return xs.element_at(3);
+                }
+            );
+
+            THEN("the output contains an 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);
+            }
+
+        }
+    }
+}
+
+SCENARIO("element_at - non-first", "[element_at][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(220, 3),
+            on.next(230, 4), //
+            on.next(240, 5),
+            on.completed(250)
+        });
+
+        WHEN("element_at is taken"){
+
+            auto res = w.start(
+                [xs]() {
+                    return xs.element_at(2);
+                }
+            );
+
+            THEN("the output contains the element at requested index"){
+                auto required = rxu::to_vector({
+                    on.next(230, 4),
+                    on.completed(230)
+                });
+                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, 230)
+                });
+                auto actual = xs.subscriptions();
+                REQUIRE(required == actual);
+            }
+
+        }
+    }
+}
+
+SCENARIO("element_at - last in a sequence", "[element_at][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(220, 3),
+            on.next(230, 4),
+            on.next(240, 5), //
+            on.completed(250)
+        });
+
+        WHEN("element_at is taken"){
+
+            auto res = w.start(
+                [xs]() {
+                    return xs.element_at(3);
+                }
+            );
+
+            THEN("the output contains the element at requested index"){
+                auto required = rxu::to_vector({
+                    on.next(240, 5),
+                    on.completed(240)
+                });
+                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, 240)
+                });
+                auto actual = xs.subscriptions();
+                REQUIRE(required == actual);
+            }
+
+        }
+    }
+}
+
+SCENARIO("element_at - invalid index", "[element_at][operators]"){
+    GIVEN("a source"){
+        auto sc = rxsc::make_test();
+        auto w = sc.create_worker();
+        const rxsc::test::messages<int> on;
+
+        std::runtime_error ex("element_at on_error from source");
+
+        auto xs = sc.make_hot_observable({
+            on.next(150, 1),
+            on.next(210, 2), // #0
+            on.next(220, 3), // #1
+            on.next(230, 4), // #2
+            on.next(240, 5), // #3
+            on.completed(250)
+        });
+
+        WHEN("element_at is taken"){
+
+            auto res = w.start(
+                [xs]() {
+                    return xs.element_at(4);
+                }
+            );
+
+            THEN("the output contains an 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);
+            }
+
+        }
+    }
+}