Unverified Commit d283fb2c authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #957

Replace messy request_response unit test
parents 52697cba 6a57d870
...@@ -72,7 +72,11 @@ public: ...@@ -72,7 +72,11 @@ public:
then_impl(f, e); then_impl(f, e);
} }
Self* self() { message_id id() const noexcept {
return mid_;
}
Self* self() noexcept {
return self_; return self_;
} }
...@@ -157,7 +161,11 @@ public: ...@@ -157,7 +161,11 @@ public:
self_->varargs_receive(rc, mid_, std::move(g), std::move(f)); self_->varargs_receive(rc, mid_, std::move(g), std::move(f));
} }
Self* self() { message_id id() const noexcept {
return mid_;
}
Self* self() noexcept {
return self_; return self_;
} }
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#define CAF_SUITE mixin.requester
#include "caf/mixin/requester.hpp"
#include "caf/test/dsl.hpp"
#include "caf/event_based_actor.hpp"
using namespace caf;
namespace {
using discarding_server_type = typed_actor<replies_to<int, int>::with<void>>;
using adding_server_type = typed_actor<replies_to<int, int>::with<int>>;
using result_type = variant<none_t, unit_t, int>;
struct fixture : test_coordinator_fixture<> {
fixture() {
result = std::make_shared<result_type>(none);
discarding_server = make_server([](int, int) {});
adding_server = make_server([](int x, int y) { return x + y; });
run();
}
template <class F>
auto make_server(F f)
-> typed_actor<replies_to<int, int>::with<decltype(f(1, 2))>> {
using impl = typed_actor<replies_to<int, int>::with<decltype(f(1, 2))>>;
auto init = [f]() -> typename impl::behavior_type {
return {
[f](int x, int y) { return f(x, y); },
};
};
return sys.spawn(init);
}
template <class T>
T make_delegator(T dest) {
auto f = [=](typename T::pointer self) -> typename T::behavior_type {
return {
[=](int x, int y) { return self->delegate(dest, x, y); },
};
};
return sys.spawn<lazy_init>(f);
}
std::shared_ptr<result_type> result = std::make_shared<result_type>(none);
discarding_server_type discarding_server;
adding_server_type adding_server;
};
} // namespace
#define ERROR_HANDLER [&](error& err) { CAF_FAIL(sys.render(err)); }
#define SUBTEST(message) \
*result = none; \
run(); \
CAF_MESSAGE("subtest: " message); \
for (int subtest_dummy = 0; subtest_dummy < 1; ++subtest_dummy)
CAF_TEST_FIXTURE_SCOPE(requester_tests, fixture)
CAF_TEST(requests without result) {
auto server = discarding_server;
SUBTEST("request.then") {
auto client = sys.spawn([=](event_based_actor* self) {
self->request(server, infinite, 1, 2).then([=] { *result = unit; });
});
run_once();
expect((int, int), from(client).to(server).with(1, 2));
expect((void), from(server).to(client));
CAF_CHECK_EQUAL(*result, unit);
}
SUBTEST("request.await") {
auto client = sys.spawn([=](event_based_actor* self) {
self->request(server, infinite, 1, 2).await([=] { *result = unit; });
});
run_once();
expect((int, int), from(client).to(server).with(1, 2));
expect((void), from(server).to(client));
CAF_CHECK_EQUAL(*result, unit);
}
SUBTEST("request.receive") {
auto res_hdl = self->request(server, infinite, 1, 2);
run();
res_hdl.receive([&] { *result = unit; }, ERROR_HANDLER);
CAF_CHECK_EQUAL(*result, unit);
}
}
CAF_TEST(requests with integer result) {
auto server = adding_server;
SUBTEST("request.then") {
auto client = sys.spawn([=](event_based_actor* self) {
self->request(server, infinite, 1, 2).then([=](int x) { *result = x; });
});
run_once();
expect((int, int), from(client).to(server).with(1, 2));
expect((int), from(server).to(client).with(3));
CAF_CHECK_EQUAL(*result, 3);
}
SUBTEST("request.await") {
auto client = sys.spawn([=](event_based_actor* self) {
self->request(server, infinite, 1, 2).await([=](int x) { *result = x; });
});
run_once();
expect((int, int), from(client).to(server).with(1, 2));
expect((int), from(server).to(client).with(3));
CAF_CHECK_EQUAL(*result, 3);
}
SUBTEST("request.receive") {
auto res_hdl = self->request(server, infinite, 1, 2);
run();
res_hdl.receive([&](int x) { *result = x; }, ERROR_HANDLER);
CAF_CHECK_EQUAL(*result, 3);
}
}
CAF_TEST(delegated request with integer result) {
auto worker = adding_server;
auto server = make_delegator(worker);
auto client = sys.spawn([=](event_based_actor* self) {
self->request(server, infinite, 1, 2).then([=](int x) { *result = x; });
});
run_once();
expect((int, int), from(client).to(server).with(1, 2));
expect((int, int), from(client).to(worker).with(1, 2));
expect((int), from(worker).to(client).with(3));
CAF_CHECK_EQUAL(*result, 3);
}
CAF_TEST_FIXTURE_SCOPE_END()
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include <utility>
#include "caf/config.hpp"
#define CAF_SUITE request_response
#include "caf/test/dsl.hpp"
#include "caf/all.hpp"
#define ERROR_HANDLER \
[&](error& err) { CAF_FAIL(system.render(err)); }
using namespace std;
using namespace caf;
using std::chrono::milliseconds;
namespace {
using f_atom = atom_constant<atom("f")>;
using i_atom = atom_constant<atom("i")>;
using idle_atom = atom_constant<atom("idle")>;
using error_atom = atom_constant<atom("error")>;
using request_atom = atom_constant<atom("request")>;
using response_atom = atom_constant<atom("response")>;
using go_atom = atom_constant<atom("go")>;
using gogo_atom = atom_constant<atom("gogo")>;
using gogogo_atom = atom_constant<atom("gogogo")>;
using no_way_atom = atom_constant<atom("NoWay")>;
using hi_there_atom = atom_constant<atom("HiThere")>;
struct sync_mirror : event_based_actor {
sync_mirror(actor_config& cfg) : event_based_actor(cfg) {
// nop
}
behavior make_behavior() override {
set_default_handler(reflect);
return {
[] {
// nop
}
};
}
};
// replies to 'f' with 0.0f and to 'i' with 0
struct float_or_int : event_based_actor {
float_or_int(actor_config& cfg) : event_based_actor(cfg) {
// nop
}
behavior make_behavior() override {
return {
[](f_atom) {
return 0.0f;
},
[](i_atom) {
return 0;
}
};
}
};
class popular_actor : public event_based_actor { // popular actors have a buddy
public:
explicit popular_actor(actor_config& cfg, actor buddy_arg)
: event_based_actor(cfg),
buddy_(std::move(buddy_arg)) {
// don't pollute unit test output with (provoked) warnings
set_default_handler(drop);
}
inline const actor& buddy() const {
return buddy_;
}
private:
actor buddy_;
};
/******************************************************************************\
* test case 1: *
* *
* A B C *
* | | | *
* | --(delegate)---> | | *
* | | --(forward)----> | *
* | X |---\ *
* | | | *
* | |<--/ *
* | <-------------(reply)-------------- | *
* X X *
\******************************************************************************/
class A : public popular_actor {
public:
explicit A(actor_config& cfg, const actor& buddy_arg)
: popular_actor(cfg, buddy_arg) {
// nop
}
behavior make_behavior() override {
return {
[=](go_atom, const actor& next) {
return delegate(next, gogo_atom::value);
}
};
}
};
class B : public popular_actor {
public:
explicit B(actor_config& cfg, const actor& buddy_arg)
: popular_actor(cfg, buddy_arg) {
// nop
}
behavior make_behavior() override {
return {
[=](gogo_atom x) {
CAF_MESSAGE("forward message to buddy");
quit();
return delegate(buddy(), x);
}
};
}
};
class C : public event_based_actor {
public:
C(actor_config& cfg) : event_based_actor(cfg) {
// don't pollute unit test output with (provoked) warnings
set_default_handler(drop);
}
behavior make_behavior() override {
return {
[=](gogo_atom) -> atom_value {
CAF_MESSAGE("received `gogo_atom`, about to quit");
quit();
return ok_atom::value;
}
};
}
};
/******************************************************************************\
* test case 2: *
* *
* A D C *
* | | | *
* | ---(request)---> | | *
* | | ---(request)---> | *
* | | |---\ *
* | | | | *
* | | |<--/ *
* | | <---(reply)----- | *
* | <---(reply)----- | *
* X X *
\******************************************************************************/
class D : public popular_actor {
public:
explicit D(actor_config& cfg, const actor& buddy_arg)
: popular_actor(cfg, buddy_arg) {
// nop
}
behavior make_behavior() override {
return {
[=](gogo_atom gogo) -> response_promise {
auto rp = make_response_promise();
request(buddy(), infinite, gogo).then(
[=](ok_atom ok) mutable {
rp.deliver(ok);
quit();
}
);
return rp;
}
};
}
};
/******************************************************************************\
* test case 3: *
* *
* Client Server Worker *
* | | | *
* | | <---(idle)------ | *
* | ---(request)---> | | *
* | | ---(request)---> | *
* | | |---\ *
* | X | | *
* | |<--/ *
* | <------------(response)------------ | *
* X *
\******************************************************************************/
behavior server(event_based_actor* self) {
return {
[=](idle_atom, actor worker) {
self->become(
keep_behavior,
[=](request_atom task) {
self->unbecome(); // await next idle message
return self->delegate(worker, task);
},
[](idle_atom) {
return skip();
}
);
},
[](request_atom) {
return skip();
}
};
}
struct fixture {
actor_system_config cfg;
actor_system system;
scoped_actor self;
fixture() : system(cfg), self(system) {
// nop
}
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(request_response_tests1, fixture)
CAF_TEST(test_void_res) {
using testee_a = typed_actor<replies_to<int, int>::with<void>>;
auto buddy = system.spawn([]() -> testee_a::behavior_type {
return [](int, int) {
// nop
};
});
self->request(buddy, infinite, 1, 2).receive(
[] {
CAF_MESSAGE("received void res");
},
ERROR_HANDLER
);
}
CAF_TEST(pending_quit) {
auto mirror = system.spawn([](event_based_actor* ptr) -> behavior {
ptr->set_default_handler(reflect);
return {
[] {
// nop
}
};
});
system.spawn([mirror](event_based_actor* ptr) {
ptr->request(mirror, infinite, 42).then(
[](int) {
CAF_ERROR("received result, should've been terminated already");
},
[](const error& err) {
CAF_CHECK_EQUAL(err, sec::request_receiver_down);
}
);
ptr->quit();
});
}
CAF_TEST(request_float_or_int) {
int invocations = 0;
auto foi = self->spawn<float_or_int, linked>();
self->send(foi, i_atom::value);
self->receive(
[](int i) {
CAF_CHECK_EQUAL(i, 0);
}
);
self->request(foi, infinite, i_atom::value).receive(
[&](int i) {
CAF_CHECK_EQUAL(i, 0);
++invocations;
},
[&](const error& err) {
CAF_ERROR("Error: " << self->system().render(err));
}
);
self->request(foi, infinite, f_atom::value).receive(
[&](float f) {
CAF_CHECK_EQUAL(f, 0.f);
++invocations;
},
[&](const error& err) {
CAF_ERROR("Error: " << self->system().render(err));
}
);
CAF_CHECK_EQUAL(invocations, 2);
CAF_MESSAGE("trigger sync failure");
self->request(foi, infinite, f_atom::value).receive(
[&](int) {
CAF_FAIL("int handler called");
},
[&](error& err) {
CAF_MESSAGE("error received");
CAF_CHECK_EQUAL(err, sec::unexpected_response);
}
);
}
CAF_TEST(request_to_mirror) {
auto mirror = system.spawn<sync_mirror>();
self->request(mirror, infinite, 42).receive(
[&](int value) {
CAF_CHECK_EQUAL(value, 42);
},
ERROR_HANDLER
);
}
CAF_TEST(request_to_a_fwd2_b_fwd2_c) {
self->request(self->spawn<A, monitored>(self), infinite,
go_atom::value, self->spawn<B>(self->spawn<C>())).receive(
[](ok_atom) {
CAF_MESSAGE("received 'ok'");
},
ERROR_HANDLER
);
}
CAF_TEST(request_to_a_fwd2_d_fwd2_c) {
self->request(self->spawn<A, monitored>(self), infinite,
go_atom::value, self->spawn<D>(self->spawn<C>())).receive(
[](ok_atom) {
CAF_MESSAGE("received 'ok'");
},
ERROR_HANDLER
);
}
CAF_TEST(request_to_self) {
self->request(self, milliseconds(50), no_way_atom::value).receive(
[&] {
CAF_ERROR("unexpected empty message");
},
[&](const error& err) {
CAF_MESSAGE("err = " << system.render(err));
CAF_REQUIRE(err == sec::request_timeout);
}
);
}
CAF_TEST(invalid_request) {
self->request(self->spawn<C>(), milliseconds(500),
hi_there_atom::value).receive(
[&](hi_there_atom) {
CAF_ERROR("C did reply to 'HiThere'");
},
[&](const error& err) {
CAF_REQUIRE_EQUAL(err, sec::unexpected_message);
}
);
}
CAF_TEST(client_server_worker_user_case) {
auto serv = self->spawn<linked>(server); // server
auto work = self->spawn<linked>([]() -> behavior { // worker
return {
[](request_atom) {
return response_atom::value;
}
};
});
// first 'idle', then 'request'
anon_send(serv, idle_atom::value, work);
self->request(serv, infinite, request_atom::value).receive(
[&](response_atom) {
CAF_MESSAGE("received 'response'");
CAF_CHECK_EQUAL(self->current_sender(), work);
},
[&](const error& err) {
CAF_ERROR("error: " << self->system().render(err));
}
);
// first 'request', then 'idle'
auto handle = self->request(serv, infinite, request_atom::value);
send_as(work, serv, idle_atom::value, work);
handle.receive(
[&](response_atom) {
CAF_CHECK_EQUAL(self->current_sender(), work.address());
},
[&](const error& err) {
CAF_ERROR("error: " << self->system().render(err));
}
);
}
behavior request_no_then_A(event_based_actor*) {
return [=](int number) {
CAF_MESSAGE("got " << number);
};
}
behavior request_no_then_B(event_based_actor* self) {
return {
[=](int number) {
self->request(self->spawn(request_no_then_A), infinite, number);
}
};
}
CAF_TEST(request_no_then) {
anon_send(system.spawn(request_no_then_B), 8);
}
CAF_TEST(async_request) {
auto foo = system.spawn([](event_based_actor* ptr) -> behavior {
auto receiver = ptr->spawn<linked>([](event_based_actor* ptr2) -> behavior {
return {
[=](int) {
return ptr2->make_response_promise();
}
};
});
ptr->request(receiver, infinite, 1).then(
[=](int) {}
);
return {
[=](int) {
CAF_MESSAGE("int received");
ptr->quit(exit_reason::user_shutdown);
}
};
});
anon_send(foo, 1);
}
CAF_TEST(skip_responses) {
auto mirror = system.spawn<sync_mirror>();
auto future = self->request(mirror, infinite, 42);
self->send(mirror, 42);
self->receive([](int x) {
CAF_CHECK_EQUAL(x, 42);
});
// second receive must time out
self->receive(
[](int) {
CAF_FAIL("received response message as ordinary message");
},
after(std::chrono::milliseconds(20)) >> [] {
CAF_MESSAGE("second receive timed out as expected");
}
);
future.receive(
[](int x) {
CAF_CHECK_EQUAL(x, 42);
},
[&](const error& err) {
CAF_FAIL(system.render(err));
}
);
}
CAF_TEST_FIXTURE_SCOPE_END()
CAF_TEST_FIXTURE_SCOPE(request_response_tests2, test_coordinator_fixture<>)
CAF_TEST(request_response_in_test_coordinator) {
auto mirror = sys.spawn<sync_mirror>();
sched.run();
sched.inline_next_enqueue();
// this block would deadlock without inlining the next enqueue
self->request(mirror, infinite, 23).receive(
[](int x) {
CAF_CHECK_EQUAL(x, 23);
},
[&](const error& err) {
CAF_FAIL("unexpected error: " << sys.render(err));
}
);
}
CAF_TEST_FIXTURE_SCOPE_END()
...@@ -29,7 +29,7 @@ ...@@ -29,7 +29,7 @@
CAF_PUSH_WARNINGS CAF_PUSH_WARNINGS
/// The type of `_`. /// The type of `_`.
struct wildcard { }; struct wildcard {};
/// Allows ignoring individual messages elements in `expect` clauses, e.g. /// Allows ignoring individual messages elements in `expect` clauses, e.g.
/// `expect((int, int), from(foo).to(bar).with(1, _))`. /// `expect((int, int), from(foo).to(bar).with(1, _))`.
...@@ -76,7 +76,7 @@ bool operator==(const message& x, const T& y) { ...@@ -76,7 +76,7 @@ bool operator==(const message& x, const T& y) {
} // namespace caf } // namespace caf
// dummy function to force ADL later on // dummy function to force ADL later on
//int inspect(int, int); // int inspect(int, int);
template <class T> template <class T>
struct has_outer_type { struct has_outer_type {
...@@ -106,7 +106,7 @@ public: ...@@ -106,7 +106,7 @@ public:
template <size_t X> template <size_t X>
using pos = std::integral_constant<size_t, X>; using pos = std::integral_constant<size_t, X>;
elementwise_compare_inspector(const Tup& xs) : xs_(xs) { explicit elementwise_compare_inspector(const Tup& xs) : xs_(xs) {
// nop // nop
} }
...@@ -123,19 +123,13 @@ private: ...@@ -123,19 +123,13 @@ private:
} }
template <size_t X, class T, class... Ts> template <size_t X, class T, class... Ts>
typename std::enable_if< typename std::enable_if<caf::meta::is_annotation<T>::value, bool>::type
caf::meta::is_annotation<T>::value,
bool
>::type
iterate(pos<X> pos, const T&, const Ts&... ys) { iterate(pos<X> pos, const T&, const Ts&... ys) {
return iterate(pos, ys...); return iterate(pos, ys...);
} }
template <size_t X, class T, class... Ts> template <size_t X, class T, class... Ts>
typename std::enable_if< typename std::enable_if<!caf::meta::is_annotation<T>::value, bool>::type
!caf::meta::is_annotation<T>::value,
bool
>::type
iterate(pos<X>, const T& y, const Ts&... ys) { iterate(pos<X>, const T& y, const Ts&... ys) {
std::integral_constant<size_t, X + 1> next; std::integral_constant<size_t, X + 1> next;
check(y, get<X>(xs_)); check(y, get<X>(xs_));
...@@ -286,8 +280,8 @@ std::tuple<T, Ts...> extract(caf_handle x) { ...@@ -286,8 +280,8 @@ std::tuple<T, Ts...> extract(caf_handle x) {
auto ptr = x->peek_at_next_mailbox_element(); auto ptr = x->peek_at_next_mailbox_element();
if (ptr == nullptr) if (ptr == nullptr)
CAF_FAIL("Mailbox is empty"); CAF_FAIL("Mailbox is empty");
CAF_FAIL("Message does not match expected pattern: " CAF_FAIL(
<< to_string(ptr->content())); "Message does not match expected pattern: " << to_string(ptr->content()));
} }
return std::move(*result); return std::move(*result);
} }
...@@ -300,9 +294,8 @@ bool received(caf_handle x) { ...@@ -300,9 +294,8 @@ bool received(caf_handle x) {
template <class... Ts> template <class... Ts>
class expect_clause { class expect_clause {
public: public:
expect_clause(caf::scheduler::test_coordinator& sched) explicit expect_clause(caf::scheduler::test_coordinator& sched)
: sched_(sched), : sched_(sched), dest_(nullptr) {
dest_(nullptr) {
peek_ = [=] { peek_ = [=] {
/// The extractor will call CAF_FAIL on a type mismatch, essentially /// The extractor will call CAF_FAIL on a type mismatch, essentially
/// performing a type check when ignoring the result. /// performing a type check when ignoring the result.
...@@ -350,7 +343,7 @@ public: ...@@ -350,7 +343,7 @@ public:
// TODO: replace this workaround with the make_tuple() line when dropping // TODO: replace this workaround with the make_tuple() line when dropping
// support for GCC 4.8. // support for GCC 4.8.
std::tuple<typename std::decay<Us>::type...> tmp{std::forward<Us>(xs)...}; std::tuple<typename std::decay<Us>::type...> tmp{std::forward<Us>(xs)...};
//auto tmp = std::make_tuple(std::forward<Us>(xs)...); // auto tmp = std::make_tuple(std::forward<Us>(xs)...);
// TODO: move tmp into lambda when switching to C++14 // TODO: move tmp into lambda when switching to C++14
peek_ = [=] { peek_ = [=] {
using namespace caf::detail; using namespace caf::detail;
...@@ -373,13 +366,72 @@ protected: ...@@ -373,13 +366,72 @@ protected:
caf::scheduler::test_coordinator& sched_; caf::scheduler::test_coordinator& sched_;
caf::strong_actor_ptr src_; caf::strong_actor_ptr src_;
caf::abstract_actor* dest_; caf::abstract_actor* dest_;
std::function<void ()> peek_; std::function<void()> peek_;
};
template <>
class expect_clause<void> {
public:
explicit expect_clause(caf::scheduler::test_coordinator& sched)
: sched_(sched), dest_(nullptr) {
// nop
}
expect_clause(expect_clause&& other) = default;
~expect_clause() {
auto ptr = dest_->peek_at_next_mailbox_element();
if (ptr == nullptr)
CAF_FAIL("no message found");
if (!ptr->content().empty())
CAF_FAIL("non-empty message found: " << to_string(ptr->content()));
run_once();
}
expect_clause& from(const wildcard&) {
return *this;
}
template <class Handle>
expect_clause& from(const Handle& whom) {
src_ = caf::actor_cast<caf::strong_actor_ptr>(whom);
return *this;
}
template <class Handle>
expect_clause& to(const Handle& whom) {
CAF_REQUIRE(sched_.prioritize(whom));
dest_ = &sched_.next_job<caf::abstract_actor>();
auto ptr = dest_->peek_at_next_mailbox_element();
CAF_REQUIRE(ptr != nullptr);
if (src_)
CAF_REQUIRE_EQUAL(ptr->sender, src_);
return *this;
}
expect_clause& to(const caf::scoped_actor& whom) {
dest_ = whom.ptr();
return *this;
}
protected:
void run_once() {
auto dptr = dynamic_cast<caf::blocking_actor*>(dest_);
if (dptr == nullptr)
sched_.run_once();
else
dptr->dequeue(); // Drop message.
}
caf::scheduler::test_coordinator& sched_;
caf::strong_actor_ptr src_;
caf::abstract_actor* dest_;
}; };
template <class... Ts> template <class... Ts>
class inject_clause { class inject_clause {
public: public:
inject_clause(caf::scheduler::test_coordinator& sched) explicit inject_clause(caf::scheduler::test_coordinator& sched)
: sched_(sched), dest_(nullptr) { : sched_(sched), dest_(nullptr) {
// nop // nop
} }
...@@ -444,9 +496,8 @@ protected: ...@@ -444,9 +496,8 @@ protected:
template <class... Ts> template <class... Ts>
class allow_clause { class allow_clause {
public: public:
allow_clause(caf::scheduler::test_coordinator& sched) explicit allow_clause(caf::scheduler::test_coordinator& sched)
: sched_(sched), : sched_(sched), dest_(nullptr) {
dest_(nullptr) {
peek_ = [=] { peek_ = [=] {
if (dest_ != nullptr) if (dest_ != nullptr)
return try_extract<Ts...>(dest_) != caf::none; return try_extract<Ts...>(dest_) != caf::none;
...@@ -522,7 +573,7 @@ protected: ...@@ -522,7 +573,7 @@ protected:
caf::scheduler::test_coordinator& sched_; caf::scheduler::test_coordinator& sched_;
caf::strong_actor_ptr src_; caf::strong_actor_ptr src_;
caf::abstract_actor* dest_; caf::abstract_actor* dest_;
std::function<bool ()> peek_; std::function<bool()> peek_;
}; };
template <class... Ts> template <class... Ts>
...@@ -589,7 +640,7 @@ public: ...@@ -589,7 +640,7 @@ public:
protected: protected:
caf_handle src_; caf_handle src_;
caf_handle dest_; caf_handle dest_;
std::function<void ()> check_; std::function<void()> check_;
}; };
template <class... Ts> template <class... Ts>
...@@ -598,9 +649,10 @@ struct test_coordinator_fixture_fetch_helper { ...@@ -598,9 +649,10 @@ struct test_coordinator_fixture_fetch_helper {
std::tuple<Ts...> std::tuple<Ts...>
operator()(caf::response_handle<Self, Interface, true>& from) const { operator()(caf::response_handle<Self, Interface, true>& from) const {
std::tuple<Ts...> result; std::tuple<Ts...> result;
from.receive( from.receive([&](Ts&... xs) { result = std::make_tuple(std::move(xs)...); },
[&](Ts&... xs) { result = std::make_tuple(std::move(xs)...); }, [&](caf::error& err) {
[&](caf::error& err) { FAIL(from.self()->system().render(err)); }); FAIL(from.self()->system().render(err));
});
return result; return result;
} }
}; };
...@@ -610,9 +662,10 @@ struct test_coordinator_fixture_fetch_helper<T> { ...@@ -610,9 +662,10 @@ struct test_coordinator_fixture_fetch_helper<T> {
template <class Self, class Interface> template <class Self, class Interface>
T operator()(caf::response_handle<Self, Interface, true>& from) const { T operator()(caf::response_handle<Self, Interface, true>& from) const {
T result; T result;
from.receive( from.receive([&](T& x) { result = std::move(x); },
[&](T& x) { result = std::move(x); }, [&](caf::error& err) {
[&](caf::error& err) { FAIL(from.self()->system().render(err)); }); FAIL(from.self()->system().render(err));
});
return result; return result;
} }
}; };
...@@ -717,6 +770,13 @@ public: ...@@ -717,6 +770,13 @@ public:
return run_until([] { return false; }); return run_until([] { return false; });
} }
/// Consume ones message or triggers the next timeout.
/// @returns `true` if a message was consumed or timeouts was triggered,
/// `false` otherwise.
bool run_once() {
return run_until([] { return true; }) > 0;
}
/// Consume messages and trigger timeouts until `pred` becomes `true` or /// Consume messages and trigger timeouts until `pred` becomes `true` or
/// until no activity remains. /// until no activity remains.
/// @returns The total number of events, i.e., messages consumed and /// @returns The total number of events, i.e., messages consumed and
......
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