Commit ac932d68 authored by Dominik Charousset's avatar Dominik Charousset

Add missing member functions to typed_actor_view

parent 80ba374f
......@@ -369,6 +369,7 @@ caf_add_component(
thread_hook
tracing_data
type_id_list
typed_actor_view
typed_behavior
typed_message_view
typed_response_promise
......
......@@ -8,12 +8,24 @@
#include "caf/config.hpp"
#include "caf/mixin/requester.hpp"
#include "caf/mixin/sender.hpp"
#include "caf/none.hpp"
#include "caf/scheduled_actor.hpp"
#include "caf/stream.hpp"
#include "caf/timespan.hpp"
#include "caf/typed_actor_view_base.hpp"
#include "caf/typed_stream.hpp"
namespace caf {
/// Utility function to force the type of `self` to depend on `T` and to raise a
/// compiler error if the user did not include 'caf/scheduled_actor/flow.hpp'.
/// The function itself does nothing and simply returns `self`.
template <class T>
auto typed_actor_view_flow_access(caf::scheduled_actor* self) {
using Self = flow::assert_scheduled_actor_hdr_t<T, caf::scheduled_actor*>;
return static_cast<Self>(self);
}
/// Decorates a pointer to a @ref scheduled_actor with a statically typed actor
/// interface.
template <class... Sigs>
......@@ -296,6 +308,101 @@ public:
return self_;
}
// -- flow API ---------------------------------------------------------------
/// @copydoc flow::coordinator::make_observable
template <class T = none_t>
auto make_observable() {
// Note: the template parameter T serves no purpose other than forcing the
// compiler to delay evaluation of this function body by having
// *something* to pass to `typed_actor_view_flow_access`.
auto self = typed_actor_view_flow_access<T>(self_);
return self->make_observable();
}
/// @copydoc scheduled_actor::to_stream
template <class Observable>
auto to_stream(cow_string name, timespan max_delay,
size_t max_items_per_batch, Observable&& obs) {
auto self = typed_actor_view_flow_access<Observable>(self_);
return self->to_stream(std::move(name), max_delay, max_items_per_batch,
std::forward<Observable>(obs));
}
/// @copydoc scheduled_actor::to_stream
template <class Observable>
auto to_stream(std::string name, timespan max_delay,
size_t max_items_per_batch, Observable&& obs) {
return to_stream(cow_string{std::move(name)}, max_delay,
max_items_per_batch, std::forward<Observable>(obs));
}
/// Returns a function object for passing it to @c compose.
scheduled_actor::to_stream_t to_stream(cow_string name, timespan max_delay,
size_t max_items_per_batch) {
return {self_, std::move(name), max_delay, max_items_per_batch};
}
/// Returns a function object for passing it to @c compose.
scheduled_actor::to_stream_t to_stream(std::string name, timespan max_delay,
size_t max_items_per_batch) {
return to_stream(cow_string{std::move(name)}, max_delay,
max_items_per_batch);
}
/// @copydoc scheduled_actor::to_typed_stream
template <class Observable>
auto to_typed_stream(cow_string name, timespan max_delay,
size_t max_items_per_batch, Observable obs) {
auto self = typed_actor_view_flow_access<Observable>(self_);
return self->to_typed_stream(std::move(name), max_delay,
max_items_per_batch, std::move(obs));
}
/// @copydoc scheduled_actor::to_typed_stream
template <class Observable>
auto to_typed_stream(std::string name, timespan max_delay,
size_t max_items_per_batch, Observable obs) {
return to_typed_stream(cow_string{std::move(name)}, max_delay,
max_items_per_batch, std::move(obs));
}
/// Returns a function object for passing it to @c compose.
scheduled_actor::to_typed_stream_t
to_typed_stream(cow_string name, timespan max_delay,
size_t max_items_per_batch) {
return {self_, std::move(name), max_delay, max_items_per_batch};
}
/// Returns a function object for passing it to @c compose.
scheduled_actor::to_typed_stream_t
to_typed_stream(std::string name, timespan max_delay,
size_t max_items_per_batch) {
return to_typed_stream(cow_string{std::move(name)}, max_delay,
max_items_per_batch);
}
/// @copydoc scheduled_actor::observe
template <class T>
auto
observe(typed_stream<T> what, size_t buf_capacity, size_t demand_threshold) {
auto self = typed_actor_view_flow_access<T>(self_);
return self->observe(std::move(what), buf_capacity, demand_threshold);
}
/// @copydoc scheduled_actor::observe_as
template <class T>
auto observe_as(stream what, size_t buf_capacity, size_t demand_threshold) {
auto self = typed_actor_view_flow_access<T>(self_);
return self->template observe_as<T>(std::move(what), buf_capacity,
demand_threshold);
}
/// @copydoc scheduled_actor::deregister_stream
void deregister_stream(uint64_t stream_id) {
self_->deregister_stream(stream_id);
}
private:
scheduled_actor* self_;
};
......
// 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 typed_actor_view
#include "caf/typed_actor_view.hpp"
#include "core-test.hpp"
#include "caf/scheduled_actor/flow.hpp"
#include "caf/typed_actor.hpp"
using namespace caf;
using namespace std::literals;
namespace {
using shared_int = std::shared_ptr<int>;
using shared_err = std::shared_ptr<error>;
using int_actor = typed_actor<result<void>(int)>;
using int_actor_ptr = int_actor::pointer_view;
struct int_actor_state {
using init_fn = std::function<void(int_actor_ptr)>;
int_actor_state(int_actor_ptr ptr, init_fn fn)
: self(ptr), init(std::move(fn)) {
// nop
}
int_actor::behavior_type make_behavior() {
init(self);
return {
[](int) {},
};
}
int_actor_ptr self;
init_fn init;
};
using int_actor_impl = int_actor::stateful_impl<int_actor_state>;
void stream_observer(event_based_actor* self, stream str, shared_int res,
shared_err err) {
// Access `self` through the view to check correct forwarding of `observe_as`.
auto view = typed_actor_view<result<void>(int)>{self};
view.observe_as<int>(str, 30, 10)
.do_on_error([err](const error& what) { *err = what; })
.for_each([res](int value) { *res += value; });
}
void typed_stream_observer(event_based_actor* self, typed_stream<int> str,
shared_int res, shared_err err) {
// Access `self` through the view to check correct forwarding of `observe`.
auto view = typed_actor_view<result<void>(int)>{self};
view.observe(str, 30, 10)
.do_on_error([err](const error& what) { *err = what; })
.for_each([res](int value) { *res += value; });
}
struct fixture : test_coordinator_fixture<> {
int_actor spawn_int_actor(int_actor_state::init_fn init) {
return sys.spawn<int_actor_impl>(std::move(init));
}
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("typed actors may use the flow API") {
GIVEN("a typed actor") {
WHEN("the actor calls make_observable") {
THEN("the function returns a flow that lives on the typed actor") {
auto res = std::make_shared<int>(0);
spawn_int_actor([=](int_actor_ptr self) {
self
->make_observable() //
.iota(1)
.take(10)
.for_each([res](int value) { *res += value; });
});
run();
CHECK_EQ(*res, 55);
}
}
WHEN("the actor creates a stream via compose") {
THEN("other actors may observe the values") {
auto res = std::make_shared<int>(0);
auto err = std::make_shared<error>();
spawn_int_actor([=](int_actor_ptr self) {
auto str = self
->make_observable() //
.iota(1)
.take(10)
.compose(self->to_stream("foo", 10ms, 10));
self->spawn(stream_observer, str, res, err);
});
run();
CHECK_EQ(*res, 55);
CHECK_EQ(*err, sec::none);
}
}
WHEN("the actor creates a stream via to_stream") {
THEN("other actors may observe the values") {
auto res = std::make_shared<int>(0);
auto err = std::make_shared<error>();
spawn_int_actor([=](int_actor_ptr self) {
auto obs = self
->make_observable() //
.iota(1)
.take(10)
.as_observable();
auto str = self->to_stream("foo", 10ms, 10, obs);
self->spawn(stream_observer, str, res, err);
});
run();
CHECK_EQ(*res, 55);
CHECK(!*err);
}
}
WHEN("the actor creates a typed stream via compose") {
THEN("other actors may observe the values") {
auto res = std::make_shared<int>(0);
auto err = std::make_shared<error>();
spawn_int_actor([=](int_actor_ptr self) {
auto str = self
->make_observable() //
.iota(1)
.take(10)
.compose(self->to_typed_stream("foo", 10ms, 10));
self->spawn(typed_stream_observer, str, res, err);
});
run();
CHECK_EQ(*res, 55);
CHECK(!*err);
}
}
WHEN("the actor creates a typed stream via to_stream") {
THEN("other actors may observe the values") {
auto res = std::make_shared<int>(0);
auto err = std::make_shared<error>();
spawn_int_actor([=](int_actor_ptr self) {
auto obs = self
->make_observable() //
.iota(1)
.take(10)
.as_observable();
auto str = self->to_typed_stream("foo", 10ms, 10, obs);
self->spawn(typed_stream_observer, str, res, err);
});
run();
CHECK_EQ(*res, 55);
CHECK(!*err);
}
}
}
WHEN("the actor creates a stream and deregisters it immediately") {
THEN("other actors receive an error when observing the stream") {
auto res = std::make_shared<int>(0);
auto err = std::make_shared<error>();
spawn_int_actor([=](int_actor_ptr self) {
auto str = self
->make_observable() //
.iota(1)
.take(10)
.compose(self->to_stream("foo", 10ms, 10));
self->spawn(stream_observer, str, res, err);
self->deregister_stream(str.id());
});
run();
CHECK_EQ(*res, 0);
CHECK_EQ(*err, sec::invalid_stream);
}
}
}
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