Commit 703dc991 authored by Dominik Charousset's avatar Dominik Charousset

Revamp expect() implementation, add received()

parent de53c367
......@@ -57,7 +57,7 @@ behavior file_reader(stateful_actor<file_reader_state>* self, size_t buf_size) {
},
// get next element
[=](buf& xs, downstream<int>& out, size_t num) {
CAF_MESSAGE("push " << num << " more messages downstream");
CAF_MESSAGE("push " << num << " messages downstream");
auto n = std::min(num, xs.size());
for (size_t i = 0; i < n; ++i)
out.push(xs[i]);
......@@ -117,31 +117,27 @@ error fail_state(const actor& x) {
CAF_TEST_FIXTURE_SCOPE(local_streaming_tests, fixture)
CAF_TEST(depth_2_pipeline) {
CAF_TEST(depth_2_pipeline_10_items) {
std::chrono::microseconds cycle{cfg.streaming_credit_round_interval_us};
sched.clock().current_time += cycle;
auto src = sys.spawn(file_reader, 10);
auto snk = sys.spawn(sum_up);
auto pipeline = snk * src;
CAF_MESSAGE(CAF_ARG(self) << CAF_ARG(src) << CAF_ARG(snk));
// self --("numbers.txt")-----------> source sink
CAF_MESSAGE("initiate stream handshake");
self->send(pipeline, "numbers.txt");
expect((string), from(self).to(src).with("numbers.txt"));
// self source --(open_stream_msg)-------> sink
expect((open_stream_msg), from(self).to(snk));
// self source <--------------(ack_open)-- sink
expect((upstream_msg::ack_open), from(snk).to(src));
// self source --(batch)-----------------> sink
CAF_MESSAGE("start data transmission (a single batch)");
expect((downstream_msg::batch), from(src).to(snk));
// self source <-------------(ack_batch)-- sink
sched.clock().current_time += cycle;
sched.dispatch();
expect((timeout_msg), from(snk).to(snk));
expect((timeout_msg), from(src).to(src));
expect((upstream_msg::ack_batch), from(snk).to(src));
// self source --(close)-----------------> sink
CAF_MESSAGE("expect close message from src and then result from snk");
expect((downstream_msg::close), from(src).to(snk));
// self <-----------------------------------------------------(result)-- sink
expect((int), from(snk).to(self).with(45));
CAF_CHECK_EQUAL(fail_state(snk), exit_reason::normal);
CAF_CHECK_EQUAL(fail_state(src), exit_reason::normal);
......
......@@ -470,7 +470,7 @@ public:
tmp->send(mma, publish_atom::value, port,
actor_cast<strong_actor_ptr>(whom), std::move(sigs), "", false);
expect((atom_value, uint16_t, strong_actor_ptr, sig_t, std::string, bool),
from(tmp).to(mma).with(_));
from(tmp).to(mma));
expect((uint16_t), from(mma).to(tmp).with(port));
}
};
......
......@@ -529,7 +529,7 @@ public:
actor_cast<strong_actor_ptr>(whom), std::move(sigs), "", false);
CAF_MESSAGE("publish from tmp to mma with port _");
expect((atom_value, uint16_t, strong_actor_ptr, sig_t, std::string, bool),
from(tmp).to(mma).with(_));
from(tmp).to(mma));
CAF_MESSAGE("publish: from mma to tmp with port " << port);
expect((uint16_t), from(mma).to(tmp).with(port));
}
......
......@@ -149,72 +149,180 @@ private:
const Tup& xs_;
};
caf::mailbox_element* peek_mailbox(caf::local_actor* ptr) {
auto sptr = dynamic_cast<caf::scheduled_actor*>(ptr);
// -- unified access to all actor handles in CAF -------------------------------
/// Reduces any of CAF's handle types to an `abstract_actor` pointer.
class caf_handle {
public:
using pointer = caf::abstract_actor*;
template <class T>
caf_handle(const T& x) {
*this = x;
}
caf_handle& operator=(caf::abstract_actor* x) {
ptr_ = x;
return *this;
}
template <class T,
class E = caf::detail::enable_if_t<!std::is_pointer<T>::value>>
caf_handle& operator=(const T& x) {
ptr_ = caf::actor_cast<pointer>(x);
return *this;
}
caf_handle& operator=(const caf_handle&) = default;
pointer get() const {
return ptr_;
}
private:
caf_handle() : ptr_(nullptr) {
// nop
}
caf::abstract_actor* ptr_;
};
// -- access to an actor's mailbox ---------------------------------------------
/// Returns a pointer to the next element in an actor's mailbox without taking
/// it out of the mailbox.
/// @pre `ptr` is alive and either a `scheduled_actor` or `blocking_actor`
caf::mailbox_element* next_mailbox_element(caf_handle x) {
CAF_ASSERT(x.get() != nullptr);
auto sptr = dynamic_cast<caf::scheduled_actor*>(x.get());
if (sptr != nullptr)
return sptr->mailbox().peek();
auto bptr = dynamic_cast<caf::blocking_actor*>(ptr);
CAF_REQUIRE(bptr != nullptr);
auto bptr = dynamic_cast<caf::blocking_actor*>(x.get());
CAF_ASSERT(bptr != nullptr);
return bptr->mailbox().peek();
}
template <class Derived>
class expect_clause_base {
// -- introspection of the next mailbox element --------------------------------
/// @private
template <class... Ts>
caf::optional<std::tuple<Ts...>> default_extract(caf_handle x) {
auto ptr = next_mailbox_element(x);
if (ptr == nullptr || !ptr->content().template match_elements<Ts...>())
return caf::none;
return ptr->content().template get_as_tuple<Ts...>();
}
/// @private
template <class T>
caf::optional<std::tuple<T>> unboxing_extract(caf_handle x) {
auto tup = default_extract<typename T::outer_type>(x);
if (tup == caf::none)
return caf::none;
return std::make_tuple(get<T>(get<0>(*tup)));
}
/// Dispatches to `unboxing_extract` if
/// `sizeof...(Ts) == 0 && has_outer_type<T>::value`, otherwise dispatches to
/// `default_extract`.
/// @private
template <class T, bool HasOuterType, class... Ts>
struct extract_impl {
caf::optional<std::tuple<T, Ts...>> operator()(caf_handle x) {
return default_extract<T, Ts...>(x);
}
};
template <class T>
struct extract_impl<T, true> {
caf::optional<std::tuple<T>> operator()(caf_handle x) {
return unboxing_extract<T>(x);
}
};
/// Returns the content of the next mailbox element without taking it out of
/// the mailbox. Fails on an empty mailbox or if the content of the next
/// element does not match `<T, Ts...>`.
template <class T, class... Ts>
std::tuple<T, Ts...> extract(caf_handle x) {
extract_impl<T, has_outer_type<T>::value, Ts...> f;
auto result = f(x);
if (result == caf::none) {
auto ptr = next_mailbox_element(x);
if (ptr == nullptr)
CAF_FAIL("Mailbox is empty");
CAF_FAIL("Message does not match expected pattern: "
<< to_string(ptr->content()));
}
return std::move(*result);
}
template <class T, class... Ts>
bool received(caf_handle x) {
extract_impl<T, has_outer_type<T>::value, Ts...> f;
return f(x) != caf::none;
}
template <class... Ts>
class expect_clause {
public:
expect_clause_base(caf::scheduler::test_coordinator& sched)
expect_clause(caf::scheduler::test_coordinator& sched)
: sched_(sched),
dest_(nullptr) {
// nop
peek_ = [=] {
/// The extractor will call CAF_FAIL on a type mismatch, essentially
/// performing a type check when ignoring the result.
extract<Ts...>(dest_);
};
}
expect_clause_base(expect_clause_base&& other)
: sched_(other.sched_),
src_(std::move(other.src_)) {
// nop
}
expect_clause(expect_clause&& other) = default;
virtual ~expect_clause_base() {
CAF_ASSERT(peek_ != nullptr);
peek_();
run_once();
~expect_clause() {
if (peek_ != nullptr) {
peek_();
run_once();
}
}
Derived& from(const wildcard&) {
return dref();
expect_clause& from(const wildcard&) {
return *this;
}
template <class Handle>
Derived& from(const Handle& whom) {
expect_clause& from(const Handle& whom) {
src_ = caf::actor_cast<caf::strong_actor_ptr>(whom);
return dref();
return *this;
}
template <class Handle>
Derived& to(const Handle& whom) {
expect_clause& to(const Handle& whom) {
CAF_REQUIRE(sched_.prioritize(whom));
dest_ = &sched_.next_job<caf::scheduled_actor>();
auto ptr = peek_mailbox(dest_);
auto ptr = next_mailbox_element(dest_);
CAF_REQUIRE(ptr != nullptr);
if (src_)
CAF_CHECK_EQUAL(ptr->sender, src_);
return dref();
return *this;
}
Derived& to(const caf::scoped_actor& whom) {
expect_clause& to(const caf::scoped_actor& whom) {
dest_ = whom.ptr();
return dref();
return *this;
}
template <class... Ts>
std::tuple<const Ts&...> peek() {
CAF_REQUIRE(dest_ != nullptr);
auto ptr = peek_mailbox(dest_);
CAF_REQUIRE(ptr != nullptr);
if (!ptr->content().template match_elements<Ts...>()) {
CAF_FAIL("Message does not match expected pattern: "
<< to_string(ptr->content()));
}
return ptr->content().template get_as_tuple<Ts...>();
template <class... Us>
void with(Us&&... xs) {
// TODO: move tmp into lambda when switching to C++14
auto tmp = std::make_tuple(std::forward<Us>(xs)...);
peek_ = [=] {
using namespace caf::detail;
elementwise_compare_inspector<decltype(tmp)> inspector{tmp};
auto ys = extract<Ts...>(dest_);
auto ys_indices = get_indices(ys);
CAF_CHECK(apply_args(inspector, ys_indices, ys));
};
}
protected:
......@@ -226,115 +334,12 @@ protected:
dptr->dequeue(); // Drop message.
}
Derived& dref() {
return *static_cast<Derived*>(this);
}
caf::scheduler::test_coordinator& sched_;
caf::strong_actor_ptr src_;
caf::local_actor* dest_;
std::function<void ()> peek_;
};
template <class... Ts>
class expect_clause : public expect_clause_base<expect_clause<Ts...>> {
public:
template <class... Us>
expect_clause(Us&&... xs)
: expect_clause_base<expect_clause<Ts...>>(std::forward<Us>(xs)...) {
this->peek_ = [=] {
this->template peek<Ts...>();
};
}
template <class... Us>
void with(Us&&... xs) {
// TODO: move tmp into lambda when switching to C++14
auto tmp = std::make_tuple(std::forward<Us>(xs)...);
this->peek_ = [=] {
elementwise_compare_inspector<decltype(tmp)> inspector{tmp};
auto ys = this->template peek<Ts...>();
CAF_CHECK(inspector(get<0>(ys)));
};
}
};
/// The single-argument expect-clause allows to automagically unwrap T
/// if it's a variant-like wrapper.
template <class T>
class expect_clause<T> : public expect_clause_base<expect_clause<T>> {
public:
template <class... Us>
expect_clause(Us&&... xs)
: expect_clause_base<expect_clause<T>>(std::forward<Us>(xs)...) {
this->peek_ = [=] {
std::integral_constant<bool, has_outer_type<T>::value> token;
type_check<T>(token);
};
}
template <class... Us>
void with(Us&&... xs) {
// TODO: move tmp into lambda when switching to C++14
auto tmp = std::make_tuple(std::forward<Us>(xs)...);
this->peek_ = [=] {
std::integral_constant<bool, has_outer_type<T>::value> token;
with_content(token, tmp);
};
}
private:
template <class U>
void with_content(std::integral_constant<bool, false>, const U& x) {
elementwise_compare_inspector<U> inspector{x};
auto xs = this->template peek<T>();
CAF_CHECK(inspector(get<0>(xs)));
}
template <class U>
void with_content(std::integral_constant<bool, true>, const U& x) {
elementwise_compare_inspector<U> inspector{x};
auto xs = this->template peek<typename T::outer_type>();
auto& x0 = get<0>(xs);
if (!is<T>(x0)) {
CAF_FAIL("is<T>(x0) !! " << caf::deep_to_string(x0));
}
CAF_CHECK(inspect(inspector, const_cast<T&>(get<T>(x0))));
}
template <class U>
void type_check(std::integral_constant<bool, false>) {
this->template peek<U>();
}
template <class U>
void type_check(std::integral_constant<bool, true>) {
auto xs = this->template peek<typename U::outer_type>();
auto& x0 = get<0>(xs);
if (!is<U>(x0)) {
CAF_FAIL("is<T>(x0) !! " << caf::deep_to_string(x0));
}
}
};
template <>
class expect_clause<void> : public expect_clause_base<expect_clause<void>> {
public:
template <class... Us>
expect_clause(Us&&... xs)
: expect_clause_base<expect_clause<void>>(std::forward<Us>(xs)...) {
// nop
}
void with() {
CAF_REQUIRE(dest_ != nullptr);
auto ptr = peek_mailbox(dest_);
CAF_CHECK(ptr->content().empty());
this->run_once();
}
};
template <class Derived>
class disallow_clause_base {
public:
......@@ -475,7 +480,7 @@ public:
void with() {
if (dest_ == nullptr)
return;
auto ptr = peek_mailbox(dest_);
auto ptr = next_mailbox_element(dest_);
CAF_REQUIRE(!ptr->content().empty());
}
};
......@@ -529,16 +534,6 @@ struct test_coordinator_fixture {
CAF_REQUIRE(ptr != nullptr);
return dynamic_cast<T&>(*ptr);
}
template <class... Ts>
expect_clause<Ts...> expect_impl() {
return {sched};
}
template <class... Ts>
disallow_clause<Ts...> disallow_impl() {
return {sched};
}
};
} // namespace <anonymous>
......
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