Commit 80ba374f authored by Dominik Charousset's avatar Dominik Charousset

Fix treating typed response handles as observables

parent 236c3078
......@@ -107,13 +107,15 @@ public:
template <class T>
flow::assert_scheduled_actor_hdr_t<flow::single<T>> as_single() && {
static_assert(std::is_same_v<response_type, message>);
static_assert(std::is_same_v<response_type, detail::type_list<T>>
|| std::is_same_v<response_type, message>);
return self_->template single_from_response<T>(policy_);
}
template <class T>
flow::assert_scheduled_actor_hdr_t<flow::observable<T>> as_observable() && {
static_assert(std::is_same_v<response_type, message>);
static_assert(std::is_same_v<response_type, detail::type_list<T>>
|| std::is_same_v<response_type, message>);
return self_->template single_from_response<T>(policy_).as_observable();
}
......
......@@ -14,6 +14,8 @@ using namespace caf;
namespace {
using i32_worker = typed_actor<result<int32_t>(int32_t)>;
struct dummy_state {
static inline const char* name = "dummy";
......@@ -33,9 +35,11 @@ using dummy_actor = stateful_actor<dummy_state>;
struct fixture : test_coordinator_fixture<> {
actor dummy;
i32_worker typed_dummy;
fixture() {
dummy = sys.spawn<dummy_actor>();
typed_dummy = actor_cast<i32_worker>(sys.spawn<dummy_actor>());
sched.run();
}
};
......@@ -45,7 +49,7 @@ struct fixture : test_coordinator_fixture<> {
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("response handles are convertible to observables and singles") {
GIVEN("a response handle that produces a valid result") {
GIVEN("a response handle with dynamic typing that produces a valid result") {
WHEN("calling as_single") {
THEN("observers see the result") {
using result_t = std::variant<none_t, int32_t, caf::error>;
......@@ -89,7 +93,51 @@ SCENARIO("response handles are convertible to observables and singles") {
}
}
}
GIVEN("a response handle that produces an error") {
GIVEN("a response handle with static typing that produces a valid result") {
WHEN("calling as_single") {
THEN("observers see the result") {
using result_t = std::variant<none_t, int32_t, caf::error>;
result_t result;
auto [self, launch] = sys.spawn_inactive<event_based_actor>();
self->request(typed_dummy, infinite, int32_t{42})
.as_single<int32_t>()
.subscribe([&result](int32_t val) { result = val; },
[&result](const error& what) { result = what; });
auto aut = actor{self};
launch();
expect((int32_t), from(aut).to(typed_dummy).with(42));
expect((int32_t), from(typed_dummy).to(aut).with(84));
CHECK(!sched.has_job());
CHECK_EQ(result, result_t{84});
}
}
WHEN("calling as_observable") {
THEN("observers see the result") {
using result_t = std::variant<none_t, int32_t, caf::error>;
size_t on_next_calls = 0;
bool completed = false;
result_t result;
auto [self, launch] = sys.spawn_inactive<event_based_actor>();
self->request(typed_dummy, infinite, int32_t{42})
.as_observable<int32_t>()
.do_on_error([&](const error& what) { result = what; })
.do_on_complete([&] { completed = true; })
.for_each([&](int32_t val) {
result = val;
++on_next_calls;
});
auto aut = actor{self};
launch();
expect((int32_t), from(aut).to(typed_dummy).with(42));
expect((int32_t), from(typed_dummy).to(aut).with(84));
CHECK(!sched.has_job());
CHECK_EQ(result, result_t{84});
CHECK_EQ(on_next_calls, 1u);
CHECK(completed);
}
}
}
GIVEN("a response handle with dynamic typing that produces an error") {
WHEN("calling as_single") {
THEN("observers see an error") {
using result_t = std::variant<none_t, int32_t, caf::error>;
......@@ -133,6 +181,50 @@ SCENARIO("response handles are convertible to observables and singles") {
}
}
}
GIVEN("a response handle with static typing that produces an error") {
WHEN("calling as_single") {
THEN("observers see an error") {
using result_t = std::variant<none_t, int32_t, caf::error>;
result_t result;
auto [self, launch] = sys.spawn_inactive<event_based_actor>();
self->request(typed_dummy, infinite, int32_t{13})
.as_single<int32_t>()
.subscribe([&result](int32_t val) { result = val; },
[&result](const error& what) { result = what; });
auto aut = actor{self};
launch();
expect((int32_t), from(aut).to(typed_dummy).with(13));
expect((error), from(typed_dummy).to(aut));
CHECK(!sched.has_job());
CHECK_EQ(result, result_t{make_error(sec::invalid_argument)});
}
}
WHEN("calling as_observable") {
THEN("observers see an error") {
using result_t = std::variant<none_t, int32_t, caf::error>;
size_t on_next_calls = 0;
bool completed = false;
result_t result;
auto [self, launch] = sys.spawn_inactive<event_based_actor>();
self->request(typed_dummy, infinite, int32_t{13})
.as_observable<int32_t>()
.do_on_error([&](const error& what) { result = what; })
.do_on_complete([&] { completed = true; })
.for_each([&](int32_t val) {
result = val;
++on_next_calls;
});
auto aut = actor{self};
launch();
expect((int32_t), from(aut).to(typed_dummy).with(13));
expect((error), from(typed_dummy).to(aut));
CHECK(!sched.has_job());
CHECK_EQ(result, result_t{make_error(sec::invalid_argument)});
CHECK_EQ(on_next_calls, 0u);
CHECK(!completed);
}
}
}
}
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