Commit c3031dd3 authored by Dominik Charousset's avatar Dominik Charousset

Merge pull request #1519

parents 0b7ecad7 fe83dc57
...@@ -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/ignore_elements.test.cpp
caf/flow/step/skip_last.test.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
......
...@@ -213,6 +213,11 @@ public: ...@@ -213,6 +213,11 @@ public:
return fn(std::move(*this)); return fn(std::move(*this));
} }
/// @copydoc observable::ignore_elements
auto ignore_elements() && {
return add_step(step::ignore_elements<output_type>{});
}
/// @copydoc observable::skip /// @copydoc observable::skip
auto skip(size_t n) && { auto skip(size_t n) && {
return add_step(step::skip<output_type>{n}); return add_step(step::skip<output_type>{n});
...@@ -588,6 +593,11 @@ observable<T>::reduce(Init init, Reducer reducer) { ...@@ -588,6 +593,11 @@ observable<T>::reduce(Init init, Reducer reducer) {
return transform(step::reduce<Reducer>{std::move(init), std::move(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> template <class T>
transformation<step::skip<T>> observable<T>::skip(size_t n) { transformation<step::skip<T>> observable<T>::skip(size_t n) {
return transform(step::skip<T>{n}); return transform(step::skip<T>{n});
......
...@@ -100,6 +100,10 @@ public: ...@@ -100,6 +100,10 @@ public:
template <class Predicate> template <class Predicate>
transformation<step::filter<Predicate>> filter(Predicate prediate); transformation<step::filter<Predicate>> filter(Predicate prediate);
/// Returns a transformation that ignores all items and only forwards calls to
/// `on_complete` and `on_error`.
transformation<step::ignore_elements<T>> ignore_elements();
/// Returns a transformation that applies `f` to each input and emits the /// Returns a transformation that applies `f` to each input and emits the
/// result of the function application. /// result of the function application.
template <class F> template <class F>
......
...@@ -4,6 +4,7 @@ ...@@ -4,6 +4,7 @@
#include "caf/flow/step/do_on_error.hpp" #include "caf/flow/step/do_on_error.hpp"
#include "caf/flow/step/do_on_next.hpp" #include "caf/flow/step/do_on_next.hpp"
#include "caf/flow/step/filter.hpp" #include "caf/flow/step/filter.hpp"
#include "caf/flow/step/ignore_elements.hpp"
#include "caf/flow/step/map.hpp" #include "caf/flow/step/map.hpp"
#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"
......
...@@ -18,6 +18,9 @@ class do_on_next; ...@@ -18,6 +18,9 @@ class do_on_next;
template <class> template <class>
class filter; class filter;
template <class>
class ignore_elements;
template <class> template <class>
class map; 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/fwd.hpp"
namespace caf::flow::step {
/// A step that ignores all elements and only forwards calls to `on_complete`
/// and `on_error`.
template <class T>
class ignore_elements {
public:
using input_type = T;
using output_type = T;
template <class Next, class... Steps>
bool on_next(const input_type&, Next&, 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 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