Unverified Commit 0e5a90e1 authored by Noir's avatar Noir Committed by GitHub

Merge branch 'master' into issue/1364

parents 61bf48f2 b3164d5b
...@@ -16,6 +16,14 @@ is based on [Keep a Changelog](https://keepachangelog.com). ...@@ -16,6 +16,14 @@ is based on [Keep a Changelog](https://keepachangelog.com).
unreachable if other actors no longer reference it. unreachable if other actors no longer reference it.
- Typed actors that use a `typed_actor_pointer` can now access the - Typed actors that use a `typed_actor_pointer` can now access the
`run_{delayed,scheduled}` member functions. `run_{delayed,scheduled}` member functions.
- Scheduled and delayed sends now return a disposable (#1362).
- Typed response handles received support for converting them to observable or
single objects.
- Typed actors that use the type-erased pointer-view type received access to the
new flow API functions (e.g., `make_observable`).
- Not initializing the meta objects table now prints a diagnosis message before
aborting the program. Previously, the application would usually just crash due
to a `nullptr`-access inside some CAF function.
### Fixed ### Fixed
...@@ -32,6 +40,7 @@ is based on [Keep a Changelog](https://keepachangelog.com). ...@@ -32,6 +40,7 @@ is based on [Keep a Changelog](https://keepachangelog.com).
operators that use the `ucast` operator internally. operators that use the `ucast` operator internally.
- The `mcast` and `ucast` operators now stop calling `on_next` immediately when - The `mcast` and `ucast` operators now stop calling `on_next` immediately when
disposed. disposed.
- Actors no longer terminate despite having open streams (#1377).
- Actors reading from external sources such as SPSC buffers via a local flow - Actors reading from external sources such as SPSC buffers via a local flow
could end up in a long-running read loop. To avoid potentially starving other could end up in a long-running read loop. To avoid potentially starving other
actors or activities, scheduled actors now limit the amount of actions that actors or activities, scheduled actors now limit the amount of actions that
......
...@@ -332,6 +332,7 @@ caf_add_component( ...@@ -332,6 +332,7 @@ caf_add_component(
message_builder message_builder
message_id message_id
message_lifetime message_lifetime
messaging
metaprogramming metaprogramming
mixin.requester mixin.requester
mixin.sender mixin.sender
...@@ -369,6 +370,7 @@ caf_add_component( ...@@ -369,6 +370,7 @@ caf_add_component(
thread_hook thread_hook
tracing_data tracing_data
type_id_list type_id_list
typed_actor_view
typed_behavior typed_behavior
typed_message_view typed_message_view
typed_response_promise typed_response_promise
......
...@@ -179,7 +179,7 @@ public: ...@@ -179,7 +179,7 @@ public:
} }
error abort_reason() const { error abort_reason() const {
impl_->abort_reason(); return impl_->abort_reason();
} }
private: private:
......
...@@ -64,8 +64,13 @@ CAF_CORE_EXPORT global_meta_objects_guard_type global_meta_objects_guard(); ...@@ -64,8 +64,13 @@ CAF_CORE_EXPORT global_meta_objects_guard_type global_meta_objects_guard();
/// is the index for accessing the corresponding meta object. /// is the index for accessing the corresponding meta object.
CAF_CORE_EXPORT span<const meta_object> global_meta_objects(); CAF_CORE_EXPORT span<const meta_object> global_meta_objects();
/// Returns the global meta object for given type ID. /// Returns the global meta object for given type ID. Aborts the program if no
CAF_CORE_EXPORT const meta_object* global_meta_object(type_id_t id); /// meta object exists for `id`.
CAF_CORE_EXPORT const meta_object& global_meta_object(type_id_t id);
/// Returns the global meta object for given type ID or `nullptr` if no meta
/// object exists for `id`.
CAF_CORE_EXPORT const meta_object* global_meta_object_or_null(type_id_t id);
/// Clears the array for storing global meta objects. /// Clears the array for storing global meta objects.
/// @warning intended for unit testing only! /// @warning intended for unit testing only!
......
...@@ -10,6 +10,7 @@ ...@@ -10,6 +10,7 @@
#include "caf/actor_clock.hpp" #include "caf/actor_clock.hpp"
#include "caf/actor_control_block.hpp" #include "caf/actor_control_block.hpp"
#include "caf/actor_profiler.hpp" #include "caf/actor_profiler.hpp"
#include "caf/disposable.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/mailbox_element.hpp" #include "caf/mailbox_element.hpp"
#include "caf/message_id.hpp" #include "caf/message_id.hpp"
...@@ -33,22 +34,23 @@ void profiled_send(Self* self, SelfHandle&& src, const Handle& dst, ...@@ -33,22 +34,23 @@ void profiled_send(Self* self, SelfHandle&& src, const Handle& dst,
} }
template <class Self, class SelfHandle, class Handle, class... Ts> template <class Self, class SelfHandle, class Handle, class... Ts>
void profiled_send(Self* self, SelfHandle&& src, const Handle& dst, disposable profiled_send(Self* self, SelfHandle&& src, const Handle& dst,
actor_clock& clock, actor_clock::time_point timeout, actor_clock& clock, actor_clock::time_point timeout,
[[maybe_unused]] message_id msg_id, Ts&&... xs) { [[maybe_unused]] message_id msg_id, Ts&&... xs) {
if (dst) { if (dst) {
if constexpr (std::is_same<Handle, group>::value) { if constexpr (std::is_same<Handle, group>::value) {
clock.schedule_message(timeout, dst, std::forward<SelfHandle>(src), return clock.schedule_message(timeout, dst, std::forward<SelfHandle>(src),
make_message(std::forward<Ts>(xs)...)); make_message(std::forward<Ts>(xs)...));
} else { } else {
auto element = make_mailbox_element(std::forward<SelfHandle>(src), msg_id, auto element = make_mailbox_element(std::forward<SelfHandle>(src), msg_id,
no_stages, std::forward<Ts>(xs)...); no_stages, std::forward<Ts>(xs)...);
CAF_BEFORE_SENDING_SCHEDULED(self, timeout, *element); CAF_BEFORE_SENDING_SCHEDULED(self, timeout, *element);
clock.schedule_message(timeout, actor_cast<strong_actor_ptr>(dst), return clock.schedule_message(timeout, actor_cast<strong_actor_ptr>(dst),
std::move(element)); std::move(element));
} }
} else { } else {
self->home_system().base_metrics().rejected_messages->inc(); self->home_system().base_metrics().rejected_messages->inc();
return {};
} }
} }
......
...@@ -249,8 +249,9 @@ public: ...@@ -249,8 +249,9 @@ public:
async::consumer_resource<T> to_resource() { async::consumer_resource<T> to_resource() {
return to_resource(defaults::flow::buffer_size, defaults::flow::min_demand); return to_resource(defaults::flow::buffer_size, defaults::flow::min_demand);
} }
const observable& as_observable() const& noexcept { const observable& as_observable() const& noexcept {
return std::move(*this); return *this;
} }
observable&& as_observable() && noexcept { observable&& as_observable() && noexcept {
......
...@@ -142,8 +142,9 @@ public: ...@@ -142,8 +142,9 @@ public:
void fwd_on_complete(buffer_emit_t) { void fwd_on_complete(buffer_emit_t) {
control_sub_ = nullptr; control_sub_ = nullptr;
err_ = make_error(sec::end_of_stream, if (state_ == state::running)
"buffer: unexpected end of the control stream"); err_ = make_error(sec::end_of_stream,
"buffer: unexpected end of the control stream");
shutdown(); shutdown();
} }
......
...@@ -99,7 +99,7 @@ public: ...@@ -99,7 +99,7 @@ public:
/// passed already). /// passed already).
template <message_priority P = message_priority::normal, class Dest = actor, template <message_priority P = message_priority::normal, class Dest = actor,
class... Ts> class... Ts>
detail::enable_if_t<!std::is_same<Dest, group>::value> detail::enable_if_t<!std::is_same<Dest, group>::value, disposable>
scheduled_send(const Dest& dest, actor_clock::time_point timeout, scheduled_send(const Dest& dest, actor_clock::time_point timeout,
Ts&&... xs) { Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send"); static_assert(sizeof...(Ts) > 0, "no message to send");
...@@ -109,15 +109,16 @@ public: ...@@ -109,15 +109,16 @@ public:
detail::type_list<detail::strip_and_convert_t<Ts>...> args_token; detail::type_list<detail::strip_and_convert_t<Ts>...> args_token;
type_check(dest, args_token); type_check(dest, args_token);
auto self = dptr(); auto self = dptr();
detail::profiled_send(self, self->ctrl(), dest, self->system().clock(), return detail::profiled_send(self, self->ctrl(), dest,
timeout, make_message_id(P), std::forward<Ts>(xs)...); self->system().clock(), timeout,
make_message_id(P), std::forward<Ts>(xs)...);
} }
/// Sends a message at given time point (or immediately if `timeout` has /// Sends a message at given time point (or immediately if `timeout` has
/// passed already). /// passed already).
template <class... Ts> template <class... Ts>
void scheduled_send(const group& dest, actor_clock::time_point timeout, disposable scheduled_send(const group& dest, actor_clock::time_point timeout,
Ts&&... xs) { Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send"); static_assert(sizeof...(Ts) > 0, "no message to send");
static_assert((detail::sendable<Ts> && ...), static_assert((detail::sendable<Ts> && ...),
"at least one type has no ID, " "at least one type has no ID, "
...@@ -128,17 +129,17 @@ public: ...@@ -128,17 +129,17 @@ public:
auto self = dptr(); auto self = dptr();
if (dest) { if (dest) {
auto& clock = self->system().clock(); auto& clock = self->system().clock();
clock.schedule_message(timeout, dest, self->ctrl(), return clock.schedule_message(timeout, dest, self->ctrl(),
make_message(std::forward<Ts>(xs)...)); make_message(std::forward<Ts>(xs)...));
} else {
self->home_system().base_metrics().rejected_messages->inc();
} }
self->home_system().base_metrics().rejected_messages->inc();
return {};
} }
/// Sends a message after a relative timeout. /// Sends a message after a relative timeout.
template <message_priority P = message_priority::normal, class Rep = int, template <message_priority P = message_priority::normal, class Rep = int,
class Period = std::ratio<1>, class Dest = actor, class... Ts> class Period = std::ratio<1>, class Dest = actor, class... Ts>
detail::enable_if_t<!std::is_same<Dest, group>::value> detail::enable_if_t<!std::is_same<Dest, group>::value, disposable>
delayed_send(const Dest& dest, std::chrono::duration<Rep, Period> rel_timeout, delayed_send(const Dest& dest, std::chrono::duration<Rep, Period> rel_timeout,
Ts&&... xs) { Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send"); static_assert(sizeof...(Ts) > 0, "no message to send");
...@@ -150,15 +151,16 @@ public: ...@@ -150,15 +151,16 @@ public:
auto self = dptr(); auto self = dptr();
auto& clock = self->system().clock(); auto& clock = self->system().clock();
auto timeout = clock.now() + rel_timeout; auto timeout = clock.now() + rel_timeout;
detail::profiled_send(self, self->ctrl(), dest, clock, timeout, return detail::profiled_send(self, self->ctrl(), dest, clock, timeout,
make_message_id(P), std::forward<Ts>(xs)...); make_message_id(P), std::forward<Ts>(xs)...);
} }
/// Sends a message after a relative timeout. /// Sends a message after a relative timeout.
template <class Rep = int, class Period = std::ratio<1>, class Dest = actor, template <class Rep = int, class Period = std::ratio<1>, class Dest = actor,
class... Ts> class... Ts>
void delayed_send(const group& dest, std::chrono::duration<Rep, Period> rtime, disposable delayed_send(const group& dest,
Ts&&... xs) { std::chrono::duration<Rep, Period> rtime,
Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send"); static_assert(sizeof...(Ts) > 0, "no message to send");
static_assert((detail::sendable<Ts> && ...), static_assert((detail::sendable<Ts> && ...),
"at least one type has no ID, " "at least one type has no ID, "
...@@ -166,19 +168,21 @@ public: ...@@ -166,19 +168,21 @@ public:
static_assert(!statically_typed<Subtype>(), static_assert(!statically_typed<Subtype>(),
"statically typed actors are not allowed to send to groups"); "statically typed actors are not allowed to send to groups");
// TODO: consider whether it's feasible to track messages to groups // TODO: consider whether it's feasible to track messages to groups
auto self = dptr();
if (dest) { if (dest) {
auto self = dptr();
auto& clock = self->system().clock(); auto& clock = self->system().clock();
auto timeout = clock.now() + rtime; auto timeout = clock.now() + rtime;
clock.schedule_message(timeout, dest, self->ctrl(), return clock.schedule_message(timeout, dest, self->ctrl(),
make_message(std::forward<Ts>(xs)...)); make_message(std::forward<Ts>(xs)...));
} }
self->home_system().base_metrics().rejected_messages->inc();
return {};
} }
template <message_priority P = message_priority::normal, class Dest = actor, template <message_priority P = message_priority::normal, class Dest = actor,
class... Ts> class... Ts>
void scheduled_anon_send(const Dest& dest, actor_clock::time_point timeout, disposable scheduled_anon_send(const Dest& dest,
Ts&&... xs) { actor_clock::time_point timeout, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send"); static_assert(sizeof...(Ts) > 0, "no message to send");
static_assert((detail::sendable<Ts> && ...), static_assert((detail::sendable<Ts> && ...),
"at least one type has no ID, " "at least one type has no ID, "
...@@ -186,15 +190,16 @@ public: ...@@ -186,15 +190,16 @@ public:
detail::type_list<detail::strip_and_convert_t<Ts>...> args_token; detail::type_list<detail::strip_and_convert_t<Ts>...> args_token;
type_check(dest, args_token); type_check(dest, args_token);
auto self = dptr(); auto self = dptr();
detail::profiled_send(self, nullptr, dest, self->system().clock(), timeout, return detail::profiled_send(self, nullptr, dest, self->system().clock(),
make_message_id(P), std::forward<Ts>(xs)...); timeout, make_message_id(P),
std::forward<Ts>(xs)...);
} }
template <message_priority P = message_priority::normal, class Dest = actor, template <message_priority P = message_priority::normal, class Dest = actor,
class Rep = int, class Period = std::ratio<1>, class... Ts> class Rep = int, class Period = std::ratio<1>, class... Ts>
void delayed_anon_send(const Dest& dest, disposable delayed_anon_send(const Dest& dest,
std::chrono::duration<Rep, Period> rel_timeout, std::chrono::duration<Rep, Period> rel_timeout,
Ts&&... xs) { Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send"); static_assert(sizeof...(Ts) > 0, "no message to send");
static_assert((detail::sendable<Ts> && ...), static_assert((detail::sendable<Ts> && ...),
"at least one type has no ID, " "at least one type has no ID, "
...@@ -204,8 +209,8 @@ public: ...@@ -204,8 +209,8 @@ public:
auto self = dptr(); auto self = dptr();
auto& clock = self->system().clock(); auto& clock = self->system().clock();
auto timeout = clock.now() + rel_timeout; auto timeout = clock.now() + rel_timeout;
detail::profiled_send(self, nullptr, dest, clock, timeout, return detail::profiled_send(self, nullptr, dest, clock, timeout,
make_message_id(P), std::forward<Ts>(xs)...); make_message_id(P), std::forward<Ts>(xs)...);
} }
private: private:
......
...@@ -107,13 +107,15 @@ public: ...@@ -107,13 +107,15 @@ public:
template <class T> template <class T>
flow::assert_scheduled_actor_hdr_t<flow::single<T>> as_single() && { 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_); return self_->template single_from_response<T>(policy_);
} }
template <class T> template <class T>
flow::assert_scheduled_actor_hdr_t<flow::observable<T>> as_observable() && { 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(); return self_->template single_from_response<T>(policy_).as_observable();
} }
......
...@@ -709,7 +709,7 @@ public: ...@@ -709,7 +709,7 @@ public:
bool alive() const noexcept { bool alive() const noexcept {
return !bhvr_stack_.empty() || !awaited_responses_.empty() return !bhvr_stack_.empty() || !awaited_responses_.empty()
|| !multiplexed_responses_.empty() || !watched_disposables_.empty() || !multiplexed_responses_.empty() || !watched_disposables_.empty()
|| !stream_sources_.empty(); || !stream_sources_.empty() || !stream_bridges_.empty();
} }
/// Runs all pending actions. /// Runs all pending actions.
......
...@@ -8,12 +8,24 @@ ...@@ -8,12 +8,24 @@
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/mixin/requester.hpp" #include "caf/mixin/requester.hpp"
#include "caf/mixin/sender.hpp" #include "caf/mixin/sender.hpp"
#include "caf/none.hpp"
#include "caf/scheduled_actor.hpp" #include "caf/scheduled_actor.hpp"
#include "caf/stream.hpp"
#include "caf/timespan.hpp" #include "caf/timespan.hpp"
#include "caf/typed_actor_view_base.hpp" #include "caf/typed_actor_view_base.hpp"
#include "caf/typed_stream.hpp"
namespace caf { 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 /// Decorates a pointer to a @ref scheduled_actor with a statically typed actor
/// interface. /// interface.
template <class... Sigs> template <class... Sigs>
...@@ -296,6 +308,101 @@ public: ...@@ -296,6 +308,101 @@ public:
return self_; 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: private:
scheduled_actor* self_; scheduled_actor* self_;
}; };
......
...@@ -35,17 +35,17 @@ bool batch::data::save(Inspector& sink) const { ...@@ -35,17 +35,17 @@ bool batch::data::save(Inspector& sink) const {
sink.emplace_error(sec::unsafe_type); sink.emplace_error(sec::unsafe_type);
return false; return false;
} }
auto meta = detail::global_meta_object(item_type_); auto& meta = detail::global_meta_object(item_type_);
auto ptr = storage_; auto ptr = storage_;
if (!sink.begin_sequence(size_)) if (!sink.begin_sequence(size_))
return false; return false;
auto len = size_; auto len = size_;
do { do {
if constexpr (std::is_same_v<Inspector, binary_serializer>) { if constexpr (std::is_same_v<Inspector, binary_serializer>) {
if (!meta->save_binary(sink, ptr)) if (!meta.save_binary(sink, ptr))
return false; return false;
} else { } else {
if (!meta->save(sink, ptr)) if (!meta.save(sink, ptr))
return false; return false;
} }
ptr += item_size_; ptr += item_size_;
......
...@@ -194,12 +194,12 @@ error_code<sec> config_value::default_construct(type_id_t id) { ...@@ -194,12 +194,12 @@ error_code<sec> config_value::default_construct(type_id_t id) {
set(uri{}); set(uri{});
return sec::none; return sec::none;
default: default:
if (auto meta = detail::global_meta_object(id)) { if (auto meta = detail::global_meta_object_or_null(id)) {
using detail::make_scope_guard;
auto ptr = malloc(meta->padded_size); auto ptr = malloc(meta->padded_size);
auto free_guard = detail::make_scope_guard([ptr] { free(ptr); }); auto free_guard = make_scope_guard([ptr] { free(ptr); });
meta->default_construct(ptr); meta->default_construct(ptr);
auto destroy_guard auto destroy_guard = make_scope_guard([=] { meta->destroy(ptr); });
= detail::make_scope_guard([=] { meta->destroy(ptr); });
config_value_writer writer{this}; config_value_writer writer{this};
if (meta->save(writer, ptr)) { if (meta->save(writer, ptr)) {
return sec::none; return sec::none;
...@@ -519,13 +519,12 @@ config_value::parse_msg_impl(std::string_view str, ...@@ -519,13 +519,12 @@ config_value::parse_msg_impl(std::string_view str,
return false; return false;
auto pos = ptr->storage(); auto pos = ptr->storage();
for (auto type : ls) { for (auto type : ls) {
auto meta = detail::global_meta_object(type); auto& meta = detail::global_meta_object(type);
CAF_ASSERT(meta != nullptr); meta.default_construct(pos);
meta->default_construct(pos);
ptr->inc_constructed_elements(); ptr->inc_constructed_elements();
if (!meta->load(reader, pos)) if (!meta.load(reader, pos))
return false; return false;
pos += meta->padded_size; pos += meta.padded_size;
} }
result.reset(ptr.release(), false); result.reset(ptr.release(), false);
return reader.end_sequence(); return reader.end_sequence();
......
...@@ -51,10 +51,28 @@ span<const meta_object> global_meta_objects() { ...@@ -51,10 +51,28 @@ span<const meta_object> global_meta_objects() {
return {meta_objects, meta_objects_size}; return {meta_objects, meta_objects_size};
} }
const meta_object* global_meta_object(type_id_t id) { const meta_object& global_meta_object(type_id_t id) {
if (id < meta_objects_size) { if (id < meta_objects_size) {
auto& meta = meta_objects[id]; auto& meta = meta_objects[id];
return !meta.type_name.empty() ? &meta : nullptr; if (!meta.type_name.empty())
return meta;
}
CAF_CRITICAL_FMT(
"found no meta object for type ID %d!\n"
" This usually means that run-time type initialization is missing.\n"
" With CAF_MAIN, make sure to pass all custom type ID blocks.\n"
" With a custom main, call (before any other CAF function):\n"
" - caf::core::init_global_meta_objects()\n"
" - <module>::init_global_meta_objects() for all loaded modules\n"
" - caf::init_global_meta_objects<T>() for all custom ID blocks",
static_cast<int>(id));
}
const meta_object* global_meta_object_or_null(type_id_t id) {
if (id < meta_objects_size) {
auto& meta = meta_objects[id];
if (!meta.type_name.empty())
return &meta;
} }
return nullptr; return nullptr;
} }
......
...@@ -64,21 +64,19 @@ int error::compare(uint8_t code, type_id_t category) const noexcept { ...@@ -64,21 +64,19 @@ int error::compare(uint8_t code, type_id_t category) const noexcept {
// -- inspection support ----------------------------------------------------- // -- inspection support -----------------------------------------------------
std::string to_string(const error& x) { std::string to_string(const error& x) {
using const_void_ptr = const void*;
using const_meta_ptr = const detail::meta_object*;
if (!x) if (!x)
return "none"; return "none";
std::string result; std::string result;
auto append = [&result](const_void_ptr ptr, auto append = [&result](const void* ptr,
const_meta_ptr meta) -> const_void_ptr { const detail::meta_object& meta) -> const void* {
meta->stringify(result, ptr); meta.stringify(result, ptr);
return static_cast<const std::byte*>(ptr) + meta->padded_size; return static_cast<const std::byte*>(ptr) + meta.padded_size;
}; };
auto code = x.code(); auto code = x.code();
append(&code, detail::global_meta_object(x.category())); append(&code, detail::global_meta_object(x.category()));
if (auto& ctx = x.context()) { if (auto& ctx = x.context()) {
result += '('; result += '(';
auto ptr = static_cast<const_void_ptr>(ctx.cdata().storage()); auto ptr = static_cast<const void*>(ctx.cdata().storage());
auto types = ctx.types(); auto types = ctx.types();
ptr = append(ptr, detail::global_meta_object(types[0])); ptr = append(ptr, detail::global_meta_object(types[0]));
for (size_t index = 1; index < types.size(); ++index) { for (size_t index = 1; index < types.size(); ++index) {
......
...@@ -71,8 +71,8 @@ bool load_data(Deserializer& source, message::data_ptr& data) { ...@@ -71,8 +71,8 @@ bool load_data(Deserializer& source, message::data_ptr& data) {
CAF_ASSERT(ids.size() == msg_size); CAF_ASSERT(ids.size() == msg_size);
size_t data_size = 0; size_t data_size = 0;
for (auto id : ids) { for (auto id : ids) {
if (auto meta_obj = detail::global_meta_object(id)) if (auto meta = detail::global_meta_object_or_null(id))
data_size += meta_obj->padded_size; data_size += meta->padded_size;
else else
STOP(sec::unknown_type); STOP(sec::unknown_type);
} }
...@@ -155,7 +155,7 @@ bool load_data(Deserializer& source, message::data_ptr& data) { ...@@ -155,7 +155,7 @@ bool load_data(Deserializer& source, message::data_ptr& data) {
for (size_t i = 0; i < msg_size; ++i) { for (size_t i = 0; i < msg_size; ++i) {
auto type = type_id_t{0}; auto type = type_id_t{0};
GUARDED(source.fetch_next_object_type(type)); GUARDED(source.fetch_next_object_type(type));
if (auto meta_obj = detail::global_meta_object(type)) { if (auto meta_obj = detail::global_meta_object_or_null(type)) {
ids.push_back(type); ids.push_back(type);
auto obj_size = meta_obj->padded_size; auto obj_size = meta_obj->padded_size;
data_size += obj_size; data_size += obj_size;
...@@ -289,14 +289,12 @@ std::string to_string(const message& msg) { ...@@ -289,14 +289,12 @@ std::string to_string(const message& msg) {
auto types = msg.types(); auto types = msg.types();
if (!types.empty()) { if (!types.empty()) {
auto ptr = msg.cdata().storage(); auto ptr = msg.cdata().storage();
auto meta = detail::global_meta_object(types[0]); auto* meta = &detail::global_meta_object(types[0]);
CAF_ASSERT(meta != nullptr);
meta->stringify(result, ptr); meta->stringify(result, ptr);
ptr += meta->padded_size; ptr += meta->padded_size;
for (size_t index = 1; index < types.size(); ++index) { for (size_t index = 1; index < types.size(); ++index) {
result += ", "; result += ", ";
meta = detail::global_meta_object(types[index]); meta = &detail::global_meta_object(types[index]);
CAF_ASSERT(meta != nullptr);
meta->stringify(result, ptr); meta->stringify(result, ptr);
ptr += meta->padded_size; ptr += meta->padded_size;
} }
......
...@@ -52,9 +52,9 @@ public: ...@@ -52,9 +52,9 @@ public:
} }
std::byte* copy_init(std::byte* storage) const override { std::byte* copy_init(std::byte* storage) const override {
auto* meta = global_meta_object(src_.type_at(index_)); auto& meta = global_meta_object(src_.type_at(index_));
meta->copy_construct(storage, src_.data().at(index_)); meta.copy_construct(storage, src_.data().at(index_));
return storage + meta->padded_size; return storage + meta.padded_size;
} }
std::byte* move_init(std::byte* storage) override { std::byte* move_init(std::byte* storage) override {
...@@ -75,8 +75,8 @@ message_builder& message_builder::append_from(const caf::message& msg, ...@@ -75,8 +75,8 @@ message_builder& message_builder::append_from(const caf::message& msg,
auto end = std::min(msg.size(), first + n); auto end = std::min(msg.size(), first + n);
for (size_t index = first; index < end; ++index) { for (size_t index = first; index < end; ++index) {
auto tid = msg.type_at(index); auto tid = msg.type_at(index);
auto* meta = detail::global_meta_object(tid); auto& meta = detail::global_meta_object(tid);
storage_size_ += meta->padded_size; storage_size_ += meta.padded_size;
types_.push_back(tid); types_.push_back(tid);
elements_.emplace_back( elements_.emplace_back(
std::make_unique<message_builder_element_adapter>(msg, index)); std::make_unique<message_builder_element_adapter>(msg, index));
......
...@@ -587,21 +587,45 @@ scheduled_actor::categorize(mailbox_element& x) { ...@@ -587,21 +587,45 @@ scheduled_actor::categorize(mailbox_element& x) {
return message_category::internal; return message_category::internal;
} }
case type_id_v<stream_open_msg>: { case type_id_v<stream_open_msg>: {
// Try to subscribe the sink to the observable.
auto& [str_id, ptr, sink_id] = content.get_as<stream_open_msg>(0); auto& [str_id, ptr, sink_id] = content.get_as<stream_open_msg>(0);
if (!ptr) {
CAF_LOG_ERROR("received a stream_open_msg with a null sink");
return message_category::internal;
}
auto sink_hdl = actor_cast<actor>(ptr); auto sink_hdl = actor_cast<actor>(ptr);
if (auto i = stream_sources_.find(str_id); i != stream_sources_.end()) { if (auto i = stream_sources_.find(str_id); i != stream_sources_.end()) {
// Create a forwarder that turns observed items into batches.
auto fwd = make_counted<batch_forwarder_impl>(this, sink_hdl, sink_id); auto fwd = make_counted<batch_forwarder_impl>(this, sink_hdl, sink_id);
auto sub = i->second.obs->subscribe(flow::observer<async::batch>{fwd}); auto sub = i->second.obs->subscribe(flow::observer<async::batch>{fwd});
if (fwd->subscribed()) { if (fwd->subscribed()) {
// Inform the sink that the stream is now open.
auto flow_id = new_u64_id(); auto flow_id = new_u64_id();
stream_subs_.emplace(flow_id, std::move(fwd)); stream_subs_.emplace(flow_id, std::move(fwd));
auto mipb = static_cast<uint32_t>(i->second.max_items_per_batch); auto mipb = static_cast<uint32_t>(i->second.max_items_per_batch);
unsafe_send_as(this, sink_hdl, unsafe_send_as(this, sink_hdl,
stream_ack_msg{ctrl(), sink_id, flow_id, mipb}); stream_ack_msg{ctrl(), sink_id, flow_id, mipb});
if (sink_hdl.node() != node()) {
// Actors cancel any pending streams when they terminate. However,
// remote actors may terminate without sending us a proper goodbye.
// Hence, we add a function object to remote actors to make sure we
// get a cancel in all cases.
auto weak_self = weak_actor_ptr{ctrl()};
sink_hdl->attach_functor([weak_self, flow_id] {
if (auto sptr = weak_self.lock())
anon_send(actor_cast<actor>(sptr), stream_cancel_msg{flow_id});
});
}
} else { } else {
CAF_LOG_ERROR("failed to subscribe a batch forwarder"); CAF_LOG_ERROR("failed to subscribe a batch forwarder");
sub.dispose(); sub.dispose();
} }
} else {
// Abort the flow immediately.
CAF_LOG_DEBUG("requested stream does not exist");
auto err = make_error(sec::invalid_stream);
unsafe_send_as(this, sink_hdl,
stream_abort_msg{sink_id, std::move(err)});
} }
return message_category::internal; return message_category::internal;
} }
...@@ -615,6 +639,7 @@ scheduled_actor::categorize(mailbox_element& x) { ...@@ -615,6 +639,7 @@ scheduled_actor::categorize(mailbox_element& x) {
case type_id_v<stream_cancel_msg>: { case type_id_v<stream_cancel_msg>: {
auto [sub_id] = content.get_as<stream_cancel_msg>(0); auto [sub_id] = content.get_as<stream_cancel_msg>(0);
if (auto i = stream_subs_.find(sub_id); i != stream_subs_.end()) { if (auto i = stream_subs_.find(sub_id); i != stream_subs_.end()) {
CAF_LOG_DEBUG("canceled stream " << sub_id);
i->second->cancel(); i->second->cancel();
stream_subs_.erase(i); stream_subs_.erase(i);
} }
......
...@@ -9,8 +9,8 @@ ...@@ -9,8 +9,8 @@
namespace caf { namespace caf {
std::string_view query_type_name(type_id_t type) { std::string_view query_type_name(type_id_t type) {
if (auto ptr = detail::global_meta_object(type)) if (auto meta = detail::global_meta_object_or_null(type))
return ptr->type_name; return meta->type_name;
return {}; return {};
} }
......
...@@ -12,10 +12,8 @@ namespace caf { ...@@ -12,10 +12,8 @@ namespace caf {
size_t type_id_list::data_size() const noexcept { size_t type_id_list::data_size() const noexcept {
auto result = size_t{0}; auto result = size_t{0};
for (auto type : *this) { for (auto type : *this)
auto meta_obj = detail::global_meta_object(type); result += detail::global_meta_object(type).padded_size;
result += meta_obj->padded_size;
}
return result; return result;
} }
...@@ -25,12 +23,12 @@ std::string to_string(type_id_list xs) { ...@@ -25,12 +23,12 @@ std::string to_string(type_id_list xs) {
std::string result; std::string result;
result += '['; result += '[';
{ {
auto tn = detail::global_meta_object(xs[0])->type_name; auto tn = detail::global_meta_object(xs[0]).type_name;
result.insert(result.end(), tn.begin(), tn.end()); result.insert(result.end(), tn.begin(), tn.end());
} }
for (size_t index = 1; index < xs.size(); ++index) { for (size_t index = 1; index < xs.size(); ++index) {
result += ", "; result += ", ";
auto tn = detail::global_meta_object(xs[index])->type_name; auto tn = detail::global_meta_object(xs[index]).type_name;
result.insert(result.end(), tn.begin(), tn.end()); result.insert(result.end(), tn.begin(), tn.end());
} }
result += ']'; result += ']';
......
// 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 messaging
#include "caf/event_based_actor.hpp"
#include "core-test.hpp"
using namespace caf;
using namespace std::literals;
using self_ptr = event_based_actor*;
namespace {
struct fixture : test_coordinator_fixture<> {
actor uut1;
actor uut2;
disposable dis;
bool had_message = false;
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("send transfers a message from one actor to another") {
GIVEN("two actors: uut1 and uut2") {
WHEN("sending a message from uu2 to uu1") {
THEN("uut1 calls the appropriate message handler") {
uut1 = sys.spawn([this](self_ptr self) -> behavior {
return {
[this, self](int i) {
had_message = true;
CHECK_EQ(i, 42);
CHECK_EQ(self->current_sender(), uut2);
},
[](float) { CAF_FAIL("float handler called"); },
};
});
uut2 = sys.spawn([this](self_ptr self) { self->send(uut1, 42); });
run();
CHECK(had_message);
}
}
}
}
SCENARIO("delayed_send transfers the message after a relative timeout") {
GIVEN("two actors: uut1 and uut2") {
WHEN("sending a message from uu2 to uu1") {
THEN("uut1 calls the appropriate message handler") {
uut1 = sys.spawn([this](self_ptr self) -> behavior {
return {
[this, self](int i) {
had_message = true;
CHECK_EQ(i, 42);
CHECK_EQ(self->current_sender(), uut2);
},
[](float) { CAF_FAIL("float handler called"); },
};
});
uut2 = sys.spawn([this](self_ptr self) { //
self->delayed_send(uut1, 1s, 42);
});
sched.run();
CHECK(!had_message);
advance_time(1s);
sched.run();
CHECK(had_message);
}
}
}
}
SCENARIO("scheduled_send transfers the message after an absolute timeout") {
GIVEN("two actors: uut1 and uut2") {
WHEN("sending a message from uu2 to uu1") {
THEN("uut1 calls the appropriate message handler") {
uut1 = sys.spawn([this](self_ptr self) -> behavior {
return {
[this, self](int i) {
had_message = true;
CHECK_EQ(i, 42);
CHECK_EQ(self->current_sender(), uut2);
},
[](float) { CAF_FAIL("float handler called"); },
};
});
uut2 = sys.spawn([this](self_ptr self) { //
auto timeout = self->clock().now() + 1s;
self->scheduled_send(uut1, timeout, 42);
});
sched.run();
CHECK(!had_message);
advance_time(1s);
sched.run();
CHECK(had_message);
}
}
}
}
SCENARIO("anon_send hides the sender of a message") {
GIVEN("two actors: uut1 and uut2") {
WHEN("sending a message from uu2 to uu1") {
THEN("uut1 calls the appropriate message handler") {
uut1 = sys.spawn([this](self_ptr self) -> behavior {
return {
[this, self](int i) {
had_message = true;
CHECK_EQ(i, 42);
CHECK_EQ(self->current_sender(), nullptr);
},
[](float) { CAF_FAIL("float handler called"); },
};
});
uut2 = sys.spawn([this](self_ptr self) { self->anon_send(uut1, 42); });
run();
CHECK(had_message);
}
}
}
}
SCENARIO("delayed_anon_send hides the sender of a message") {
GIVEN("two actors: uut1 and uut2") {
WHEN("sending a message from uu2 to uu1") {
THEN("uut1 calls the appropriate message handler") {
uut1 = sys.spawn([this](self_ptr self) -> behavior {
return {
[this, self](int i) {
had_message = true;
CHECK_EQ(i, 42);
CHECK_EQ(self->current_sender(), nullptr);
},
[](float) { CAF_FAIL("float handler called"); },
};
});
uut2 = sys.spawn([this](self_ptr self) { //
self->delayed_anon_send(uut1, 1s, 42);
});
sched.run();
CHECK(!had_message);
advance_time(1s);
sched.run();
CHECK(had_message);
}
}
}
}
SCENARIO("scheduled_anon_send hides the sender of a message") {
GIVEN("two actors: uut1 and uut2") {
WHEN("sending a message from uu2 to uu1") {
THEN("uut1 calls the appropriate message handler") {
uut1 = sys.spawn([this](self_ptr self) -> behavior {
return {
[this, self](int i) {
had_message = true;
CHECK_EQ(i, 42);
CHECK_EQ(self->current_sender(), nullptr);
},
[](float) { CAF_FAIL("float handler called"); },
};
});
uut2 = sys.spawn([this](self_ptr self) { //
auto timeout = self->clock().now() + 1s;
self->scheduled_anon_send(uut1, timeout, 42);
});
sched.run();
CHECK(!had_message);
advance_time(1s);
sched.run();
CHECK(had_message);
}
}
}
}
SCENARIO("a delayed message may be canceled before its timeout") {
GIVEN("two actors: uut1 and uut2") {
WHEN("when disposing the message of delayed_send before it arrives") {
THEN("uut1 receives no message") {
uut1 = sys.spawn([this](self_ptr self) -> behavior {
return {
[this, self](int) {
had_message = true;
CHECK_EQ(self->current_sender(), uut2);
},
[](float) { CAF_FAIL("float handler called"); },
};
});
uut2 = sys.spawn([this](self_ptr self) { //
dis = self->delayed_send(uut1, 1s, 42);
});
sched.run();
CHECK(!had_message);
dis.dispose();
advance_time(1s);
run();
CHECK(!had_message);
}
}
WHEN("when disposing the message of delayed_anon_send before it arrives") {
THEN("uut1 receives no message") {
uut1 = sys.spawn([this](self_ptr self) -> behavior {
return {
[this, self](int) {
had_message = true;
CHECK_EQ(self->current_sender(), uut2);
},
[](float) { CAF_FAIL("float handler called"); },
};
});
uut2 = sys.spawn([this](self_ptr self) { //
dis = self->delayed_anon_send(uut1, 1s, 42);
});
sched.run();
CHECK(!had_message);
dis.dispose();
advance_time(1s);
run();
CHECK(!had_message);
}
}
}
}
SCENARIO("a scheduled message may be canceled before its timeout") {
GIVEN("two actors: uut1 and uut2") {
WHEN("when disposing the message of scheduled_send before it arrives") {
THEN("uut1 receives no message") {
uut1 = sys.spawn([this](self_ptr self) -> behavior {
return {
[this, self](int) {
had_message = true;
CHECK_EQ(self->current_sender(), uut2);
},
[](float) { CAF_FAIL("float handler called"); },
};
});
uut2 = sys.spawn([this](self_ptr self) { //
auto timeout = self->clock().now() + 1s;
dis = self->scheduled_send(uut1, timeout, 42);
});
sched.run();
CHECK(!had_message);
dis.dispose();
advance_time(1s);
run();
CHECK(!had_message);
}
}
WHEN(
"when disposing the message of scheduled_anon_send before it arrives") {
THEN("uut1 receives no message") {
uut1 = sys.spawn([this](self_ptr self) -> behavior {
return {
[this, self](int) {
had_message = true;
CHECK_EQ(self->current_sender(), uut2);
},
[](float) { CAF_FAIL("float handler called"); },
};
});
uut2 = sys.spawn([this](self_ptr self) { //
auto timeout = self->clock().now() + 1s;
dis = self->scheduled_anon_send(uut1, timeout, 42);
});
sched.run();
CHECK(!had_message);
dis.dispose();
advance_time(1s);
run();
CHECK(!had_message);
}
}
}
}
END_FIXTURE_SCOPE()
...@@ -14,6 +14,8 @@ using namespace caf; ...@@ -14,6 +14,8 @@ using namespace caf;
namespace { namespace {
using i32_worker = typed_actor<result<int32_t>(int32_t)>;
struct dummy_state { struct dummy_state {
static inline const char* name = "dummy"; static inline const char* name = "dummy";
...@@ -33,9 +35,11 @@ using dummy_actor = stateful_actor<dummy_state>; ...@@ -33,9 +35,11 @@ using dummy_actor = stateful_actor<dummy_state>;
struct fixture : test_coordinator_fixture<> { struct fixture : test_coordinator_fixture<> {
actor dummy; actor dummy;
i32_worker typed_dummy;
fixture() { fixture() {
dummy = sys.spawn<dummy_actor>(); dummy = sys.spawn<dummy_actor>();
typed_dummy = actor_cast<i32_worker>(sys.spawn<dummy_actor>());
sched.run(); sched.run();
} }
}; };
...@@ -45,7 +49,7 @@ struct fixture : test_coordinator_fixture<> { ...@@ -45,7 +49,7 @@ struct fixture : test_coordinator_fixture<> {
BEGIN_FIXTURE_SCOPE(fixture) BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("response handles are convertible to observables and singles") { 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") { WHEN("calling as_single") {
THEN("observers see the result") { THEN("observers see the result") {
using result_t = std::variant<none_t, int32_t, caf::error>; using result_t = std::variant<none_t, int32_t, caf::error>;
...@@ -89,7 +93,51 @@ SCENARIO("response handles are convertible to observables and singles") { ...@@ -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") { WHEN("calling as_single") {
THEN("observers see an error") { THEN("observers see an error") {
using result_t = std::variant<none_t, int32_t, caf::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") { ...@@ -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() END_FIXTURE_SCOPE()
// 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_typed_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