Commit f8e31587 authored by Shariar Azad Riday's avatar Shariar Azad Riday

Add skip_last step for observables

parent 29879209
...@@ -153,6 +153,7 @@ caf_add_component( ...@@ -153,6 +153,7 @@ caf_add_component(
caf/flow/observable_builder.cpp caf/flow/observable_builder.cpp
caf/flow/op/interval.cpp caf/flow/op/interval.cpp
caf/flow/scoped_coordinator.cpp caf/flow/scoped_coordinator.cpp
caf/flow/step/skip_last.test.cpp
caf/flow/step/take_last.test.cpp caf/flow/step/take_last.test.cpp
caf/flow/subscription.cpp caf/flow/subscription.cpp
caf/forwarding_actor_proxy.cpp caf/forwarding_actor_proxy.cpp
......
...@@ -218,6 +218,11 @@ public: ...@@ -218,6 +218,11 @@ public:
return add_step(step::skip<output_type>{n}); return add_step(step::skip<output_type>{n});
} }
/// @copydoc observable::skip_last
auto skip_last(size_t n) && {
return add_step(step::skip_last<output_type>{n});
}
/// @copydoc observable::take /// @copydoc observable::take
auto take(size_t n) && { auto take(size_t n) && {
return add_step(step::take<output_type>{n}); return add_step(step::take<output_type>{n});
...@@ -588,6 +593,11 @@ transformation<step::skip<T>> observable<T>::skip(size_t n) { ...@@ -588,6 +593,11 @@ transformation<step::skip<T>> observable<T>::skip(size_t n) {
return transform(step::skip<T>{n}); return transform(step::skip<T>{n});
} }
template <class T>
transformation<step::skip_last<T>> observable<T>::skip_last(size_t n) {
return transform(step::skip_last<T>{n});
}
template <class T> template <class T>
transformation<step::take<T>> observable<T>::take(size_t n) { transformation<step::take<T>> observable<T>::take(size_t n) {
return transform(step::take<T>{n}); return transform(step::take<T>{n});
......
...@@ -118,6 +118,9 @@ public: ...@@ -118,6 +118,9 @@ public:
/// Returns a transformation that selects all but the first `n` items. /// Returns a transformation that selects all but the first `n` items.
transformation<step::skip<T>> skip(size_t n); transformation<step::skip<T>> skip(size_t n);
/// Returns a transformation that discards only the last `n` items.
transformation<step::skip_last<T>> skip_last(size_t n);
/// Returns a transformation that selects only the first `n` items. /// Returns a transformation that selects only the first `n` items.
transformation<step::take<T>> take(size_t n); transformation<step::take<T>> take(size_t n);
......
...@@ -8,6 +8,7 @@ ...@@ -8,6 +8,7 @@
#include "caf/flow/step/on_error_complete.hpp" #include "caf/flow/step/on_error_complete.hpp"
#include "caf/flow/step/reduce.hpp" #include "caf/flow/step/reduce.hpp"
#include "caf/flow/step/skip.hpp" #include "caf/flow/step/skip.hpp"
#include "caf/flow/step/skip_last.hpp"
#include "caf/flow/step/take.hpp" #include "caf/flow/step/take.hpp"
#include "caf/flow/step/take_last.hpp" #include "caf/flow/step/take_last.hpp"
#include "caf/flow/step/take_while.hpp" #include "caf/flow/step/take_while.hpp"
...@@ -30,6 +30,9 @@ class reduce; ...@@ -30,6 +30,9 @@ class reduce;
template <class> template <class>
class skip; class skip;
template <class>
class skip_last;
template <class> template <class>
class take; class take;
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/detail/ring_buffer.hpp"
#include "caf/fwd.hpp"
#include <cstddef>
namespace caf::flow::step {
template <class T>
class skip_last {
public:
using input_type = T;
using output_type = T;
explicit skip_last(size_t num) : elements_(num) {
// nop
}
skip_last(skip_last&&) = default;
skip_last(const skip_last&) = default;
skip_last& operator=(skip_last&&) = default;
skip_last& operator=(const skip_last&) = default;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next& next, Steps&... steps) {
if (elements_.full()) {
if (!next.on_next(elements_.front(), steps...))
return false;
elements_.pop_front();
}
elements_.push_back(item);
return true;
}
template <class Next, class... Steps>
void on_complete(Next& next, Steps&... steps) {
next.on_complete(steps...);
}
template <class Next, class... Steps>
void on_error(const error& what, Next& next, Steps&... steps) {
next.on_error(what, steps...);
}
private:
detail::ring_buffer<output_type> elements_;
};
} // namespace caf::flow::step
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include "caf/flow/step/skip_last.hpp"
#include "caf/test/caf_test_main.hpp"
#include "caf/test/test.hpp"
#include "caf/flow/scoped_coordinator.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include <vector>
using namespace caf;
using caf::flow::make_observer;
struct fixture {
flow::scoped_coordinator_ptr ctx = flow::make_scoped_coordinator();
template <class... Ts>
static auto ls(Ts... xs) {
return std::vector<int>{xs...};
}
};
WITH_FIXTURE(fixture) {
TEST("calling skip_last(5) on range(1, 10) produces [1, 2, 3, 4, 5]") {
auto result = std::vector<int>{};
SECTION("blueprint") {
ctx->make_observable().range(1, 10).skip_last(5).for_each(
[&](const int& result_observable) {
result.emplace_back(result_observable);
});
}
SECTION("observable") {
ctx->make_observable().range(1, 10).as_observable().skip_last(5).for_each(
[&](const int& result_observable) {
result.emplace_back(result_observable);
});
}
ctx->run();
check_eq(result.size(), 5u);
check_eq(result, ls(1, 2, 3, 4, 5));
}
TEST("calling skip_last(5) on range(1, 5) produces []") {
auto result = std::vector<int>{};
SECTION("blueprint") {
ctx->make_observable().range(1, 5).skip_last(5).for_each(
[&](const int& result_observable) {
result.emplace_back(result_observable);
});
}
SECTION("observable") {
ctx->make_observable().range(1, 5).as_observable().skip_last(5).for_each(
[&](const int& result_observable) {
result.emplace_back(result_observable);
});
}
ctx->run();
check_eq(result.size(), 0u);
}
TEST("calling skip_last(5) on range(1, 3) produces []") {
auto result = std::vector<int>{};
SECTION("blueprint") {
ctx->make_observable().range(1, 3).skip_last(5).for_each(
[&](const int& result_observable) {
result.emplace_back(result_observable);
});
}
SECTION("observable") {
ctx->make_observable().range(1, 3).as_observable().skip_last(5).for_each(
[&](const int& result_observable) {
result.emplace_back(result_observable);
});
}
ctx->run();
check_eq(result.size(), 0u);
}
TEST("calling take(3) on skip_last(5) ignores the last two items") {
auto result = std::vector<int>{};
SECTION("blueprint") {
ctx->make_observable().range(1, 10).skip_last(5).take(3).for_each(
[&](const int& result_observable) {
result.emplace_back(result_observable);
});
}
SECTION("observable") {
ctx->make_observable()
.range(1, 100)
.as_observable()
.skip_last(5)
.take(3)
.for_each([&](const int& result_observable) {
result.emplace_back(result_observable);
});
}
ctx->run();
check_eq(result.size(), 3u);
check_eq(result, ls(1, 2, 3));
}
TEST("skip_last operator forwards errors") {
caf::error result;
auto outputs = std::vector<int>{};
SECTION("blueprint") {
ctx->make_observable()
.fail<int>(make_error(sec::runtime_error))
.skip_last(5)
.do_on_error([&result](const error& err) { result = err; })
.for_each([&](const int& result_observable) {
outputs.emplace_back(result_observable);
});
}
SECTION("observable") {
ctx->make_observable()
.fail<int>(make_error(sec::runtime_error))
.as_observable()
.skip_last(5)
.do_on_error([&result](const error& err) { result = err; })
.for_each([&](const int& result_observable) {
outputs.emplace_back(result_observable);
});
}
ctx->run();
check_eq(result, caf::sec::runtime_error);
check_eq(outputs.size(), 0u);
}
} // WITH_FIXTURE(fixture)
CAF_TEST_MAIN()
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment