Commit f9561cb6 authored by Dominik Charousset's avatar Dominik Charousset

Implement prefix_and_tail flow operator

parent 5bbe822f
......@@ -288,6 +288,7 @@ caf_add_component(
flow.for_each
flow.merge
flow.observe_on
flow.prefix_and_tail
flow.single
function_view
fused_downstream_manager
......
This diff is collapsed.
......@@ -426,6 +426,10 @@ private:
namespace caf::flow {
/// Creates an observer from given callbacks.
/// @param on_next Callback for handling incoming elements.
/// @param on_error Callback for handling an error.
/// @param on_complete Callback for handling the end-of-stream event.
template <class OnNext, class OnError, class OnComplete>
auto make_observer(OnNext on_next, OnError on_error, OnComplete on_complete) {
using impl_type = detail::default_observer_impl<OnNext, OnError, OnComplete>;
......@@ -435,6 +439,9 @@ auto make_observer(OnNext on_next, OnError on_error, OnComplete on_complete) {
return observer<input_type>{std::move(ptr)};
}
/// Creates an observer from given callbacks.
/// @param on_next Callback for handling incoming elements.
/// @param on_error Callback for handling an error.
template <class OnNext, class OnError>
auto make_observer(OnNext on_next, OnError on_error) {
using impl_type = detail::default_observer_impl<OnNext, OnError>;
......@@ -443,6 +450,8 @@ auto make_observer(OnNext on_next, OnError on_error) {
return observer<input_type>{std::move(ptr)};
}
/// Creates an observer from given callbacks.
/// @param on_next Callback for handling incoming elements.
template <class OnNext>
auto make_observer(OnNext on_next) {
using impl_type = detail::default_observer_impl<OnNext>;
......@@ -451,4 +460,17 @@ auto make_observer(OnNext on_next) {
return observer<input_type>{std::move(ptr)};
}
/// Creates an observer from a smart pointer to a custom object that implements
/// `on_next`, `on_error` and `on_complete` as member functions.
/// @param ptr Smart pointer to a custom object.
template <class SmartPointer>
auto make_observer_from_ptr(SmartPointer ptr) {
using obj_t = std::remove_reference_t<decltype(*ptr)>;
using on_next_fn = decltype(&obj_t::on_next);
using value_type = typename detail::on_next_trait_t<on_next_fn>::value_type;
return make_observer([ptr](const value_type& x) { ptr->on_next(x); },
[ptr](const error& what) { ptr->on_error(what); },
[ptr] { ptr->on_complete(); });
}
} // namespace caf::flow
......@@ -150,7 +150,7 @@ struct flat_map_optional_step {
};
template <class T, class Fn>
struct on_complete_step {
struct do_on_complete_step {
using input_type = T;
using output_type = T;
......@@ -175,7 +175,7 @@ struct on_complete_step {
};
template <class T, class Fn>
struct on_error_step {
struct do_on_error_step {
using input_type = T;
using output_type = T;
......@@ -200,7 +200,7 @@ struct on_error_step {
};
template <class T, class Fn>
struct finally_step {
struct do_finally_step {
using input_type = T;
using output_type = T;
......@@ -225,4 +225,28 @@ struct finally_step {
}
};
/// Catches errors by converting them into complete events instead.
template <class T>
struct on_error_complete_step {
using input_type = T;
using output_type = T;
template <class Next, class... Steps>
bool on_next(const input_type& item, Next& next, Steps&... steps) {
return next.on_next(item, steps...);
}
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&, Next& next, Steps&... steps) {
next.on_complete(steps...);
}
};
} // namespace caf::flow
// 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.
#define CAF_SUITE flow.prefix_and_tail
#include "caf/flow/observable.hpp"
#include "core-test.hpp"
#include <memory>
#include "caf/flow/coordinator.hpp"
#include "caf/flow/merge.hpp"
#include "caf/flow/observable_builder.hpp"
#include "caf/flow/observer.hpp"
#include "caf/flow/scoped_coordinator.hpp"
using namespace caf;
namespace {
template <class T>
struct test_observer {
void on_next(T x) {
values.emplace_back(std::move(x));
}
void on_error(const error& what) {
had_error = true;
err = what;
}
void on_complete() {
had_complete = true;
}
std::vector<T> values;
bool had_error = false;
bool had_complete = false;
error err;
};
struct fixture : test_coordinator_fixture<> {
flow::scoped_coordinator_ptr ctx = flow::make_scoped_coordinator();
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("prefix_and_tail splits off initial elements") {
using tuple_t = cow_tuple<std::vector<int>, flow::observable<int>>;
GIVEN("a generation with 0 values") {
WHEN("calling prefix_and_tail(2)") {
THEN("the observer of prefix_and_tail only receives on_complete") {
auto inputs = std::vector<int>{};
auto obs = std::make_shared<test_observer<tuple_t>>();
ctx->make_observable()
.from_container(inputs)
.prefix_and_tail(2)
.subscribe(flow::make_observer_from_ptr(obs));
ctx->run();
CHECK(obs->had_complete);
CHECK(!obs->had_error);
CHECK(obs->values.empty());
}
}
}
GIVEN("a generation with 1 values") {
WHEN("calling prefix_and_tail(2)") {
THEN("the observer of prefix_and_tail only receives on_complete") {
auto inputs = std::vector<int>{1};
auto obs = std::make_shared<test_observer<tuple_t>>();
ctx->make_observable()
.from_container(inputs)
.prefix_and_tail(2)
.subscribe(flow::make_observer_from_ptr(obs));
ctx->run();
CHECK(obs->had_complete);
CHECK(!obs->had_error);
CHECK(obs->values.empty());
}
}
}
GIVEN("a generation with 2 values") {
WHEN("calling prefix_and_tail(2)") {
THEN("the observer receives the first 2 elements plus empty remainder") {
auto inputs = std::vector<int>{1, 2};
auto prefix_vals = std::vector<int>{1, 2};
auto tail_vals = std::vector<int>{};
auto obs = std::make_shared<test_observer<int>>();
auto flat_map_calls = 0;
ctx->make_observable()
.from_container(inputs)
.prefix_and_tail(2)
.flat_map([&](const tuple_t& x) {
++flat_map_calls;
auto& [prefix, tail] = x.data();
CHECK_EQ(prefix, prefix_vals);
return tail;
})
.subscribe(flow::make_observer_from_ptr(obs));
ctx->run();
CHECK_EQ(flat_map_calls, 1);
CHECK_EQ(obs->values, tail_vals);
CHECK(obs->had_complete);
CHECK(!obs->had_error);
}
}
}
GIVEN("a generation with 8 values") {
WHEN("calling prefix_and_tail(2)") {
THEN("the observer receives the first 2 elements plus remainder") {
auto inputs = std::vector<int>{1, 2, 4, 8, 16, 32, 64, 128};
auto prefix_vals = std::vector<int>{1, 2};
auto tail_vals = std::vector<int>{4, 8, 16, 32, 64, 128};
auto obs = std::make_shared<test_observer<int>>();
auto flat_map_calls = 0;
ctx->make_observable()
.from_container(inputs)
.prefix_and_tail(2)
.flat_map([&](const tuple_t& x) {
++flat_map_calls;
auto& [prefix, tail] = x.data();
CHECK_EQ(prefix, prefix_vals);
return tail;
})
.subscribe(flow::make_observer_from_ptr(obs));
ctx->run();
CHECK_EQ(flat_map_calls, 1);
CHECK_EQ(obs->values, tail_vals);
CHECK(obs->had_complete);
CHECK(!obs->had_error);
}
}
}
}
SCENARIO("head_and_tail splits off the first element") {
using tuple_t = cow_tuple<int, flow::observable<int>>;
GIVEN("a generation with 0 values") {
WHEN("calling head_and_tail") {
THEN("the observer of head_and_tail only receives on_complete") {
auto inputs = std::vector<int>{};
auto obs = std::make_shared<test_observer<tuple_t>>();
ctx->make_observable()
.from_container(inputs)
.head_and_tail()
.subscribe(flow::make_observer_from_ptr(obs));
ctx->run();
CHECK(obs->had_complete);
CHECK(!obs->had_error);
CHECK(obs->values.empty());
}
}
}
GIVEN("a generation with 1 values") {
WHEN("calling head_and_tail()") {
THEN("the observer receives the first element plus empty remainder") {
auto inputs = std::vector<int>{1};
auto prefix_val = 1;
auto tail_vals = std::vector<int>{};
auto obs = std::make_shared<test_observer<int>>();
auto flat_map_calls = 0;
ctx->make_observable()
.from_container(inputs)
.head_and_tail()
.flat_map([&](const tuple_t& x) {
++flat_map_calls;
auto& [prefix, tail] = x.data();
CHECK_EQ(prefix, prefix_val);
return tail;
})
.subscribe(flow::make_observer_from_ptr(obs));
ctx->run();
CHECK_EQ(flat_map_calls, 1);
CHECK_EQ(obs->values, tail_vals);
CHECK(obs->had_complete);
CHECK(!obs->had_error);
}
}
}
GIVEN("a generation with 2 values") {
WHEN("calling head_and_tail()") {
THEN("the observer receives the first element plus remainder") {
auto inputs = std::vector<int>{1, 2};
auto prefix_val = 1;
auto tail_vals = std::vector<int>{2};
auto obs = std::make_shared<test_observer<int>>();
auto flat_map_calls = 0;
ctx->make_observable()
.from_container(inputs)
.head_and_tail()
.flat_map([&](const tuple_t& x) {
++flat_map_calls;
auto& [prefix, tail] = x.data();
CHECK_EQ(prefix, prefix_val);
return tail;
})
.subscribe(flow::make_observer_from_ptr(obs));
ctx->run();
CHECK_EQ(flat_map_calls, 1);
CHECK_EQ(obs->values, tail_vals);
CHECK(obs->had_complete);
CHECK(!obs->had_error);
}
}
}
}
END_FIXTURE_SCOPE()
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