Commit 605b9f29 authored by Shariar Azad Riday's avatar Shariar Azad Riday

Add ignore elements flow operator

parent 27b781cd
......@@ -153,6 +153,7 @@ caf_add_component(
caf/flow/observable_builder.cpp
caf/flow/op/interval.cpp
caf/flow/scoped_coordinator.cpp
caf/flow/step/ignore_elements.test.cpp
caf/flow/step/skip_last.test.cpp
caf/flow/step/take_last.test.cpp
caf/flow/subscription.cpp
......
......@@ -213,6 +213,11 @@ public:
return fn(std::move(*this));
}
/// @copydoc observable::ignore_elements
auto ignore_elements() && {
return add_step(step::ignore_elements<output_type>{});
}
/// @copydoc observable::skip
auto skip(size_t n) && {
return add_step(step::skip<output_type>{n});
......@@ -588,6 +593,11 @@ observable<T>::reduce(Init init, Reducer reducer) {
return transform(step::reduce<Reducer>{std::move(init), std::move(reducer)});
}
template <class T>
transformation<step::ignore_elements<T>> observable<T>::ignore_elements() {
return transform(step::ignore_elements<T>{});
}
template <class T>
transformation<step::skip<T>> observable<T>::skip(size_t n) {
return transform(step::skip<T>{n});
......
......@@ -100,6 +100,9 @@ public:
template <class Predicate>
transformation<step::filter<Predicate>> filter(Predicate prediate);
/// Returns a transformation that ignores all items and only forwards complete
transformation<step::ignore_elements<T>> ignore_elements();
/// Returns a transformation that applies `f` to each input and emits the
/// result of the function application.
template <class F>
......
......@@ -4,6 +4,7 @@
#include "caf/flow/step/do_on_error.hpp"
#include "caf/flow/step/do_on_next.hpp"
#include "caf/flow/step/filter.hpp"
#include "caf/flow/step/ignore_elements.hpp"
#include "caf/flow/step/map.hpp"
#include "caf/flow/step/on_error_complete.hpp"
#include "caf/flow/step/reduce.hpp"
......
......@@ -18,6 +18,9 @@ class do_on_next;
template <class>
class filter;
template <class>
class ignore_elements;
template <class>
class map;
......
// 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 ignore_elements {
public:
using input_type = T;
using output_type = T;
explicit ignore_elements() {
// nop
}
ignore_elements(ignore_elements&&) = default;
ignore_elements(const ignore_elements&) = default;
ignore_elements& operator=(ignore_elements&&) = default;
ignore_elements& operator=(const ignore_elements&) = default;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next& next, Steps&... steps) {
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...);
}
};
} // 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/ignore_elements.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 take(5) after ignore_elements on range(1, 10) produces []") {
auto result = std::vector<int>{};
SECTION("blueprint") {
ctx->make_observable().range(1, 10).ignore_elements().take(5).for_each(
[&](const int& result_observable) {
result.emplace_back(result_observable);
});
}
SECTION("observable") {
ctx->make_observable()
.range(1, 10)
.as_observable()
.ignore_elements()
.take(5)
.for_each([&](const int& result_observable) {
result.emplace_back(result_observable);
});
}
ctx->run();
check_eq(result.size(), 0u);
}
TEST("calling ignore_elements on range(1, 10) produces []") {
auto result = std::vector<int>{};
SECTION("blueprint") {
ctx->make_observable().range(1, 10).ignore_elements().for_each(
[&](const int& result_observable) {
result.emplace_back(result_observable);
});
}
SECTION("observable") {
ctx->make_observable()
.range(1, 10)
.as_observable()
.ignore_elements()
.for_each([&](const int& result_observable) {
result.emplace_back(result_observable);
});
}
ctx->run();
check_eq(result.size(), 0u);
}
TEST("ignore_elements operator forwards errors") {
caf::error result;
auto outputs = std::vector<int>{};
SECTION("blueprint") {
ctx->make_observable()
.fail<int>(make_error(sec::runtime_error))
.ignore_elements()
.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()
.ignore_elements()
.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