Commit ef25ccde authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'issue/1006'

Closes #1006.
parents e44c8110 d6aefd04
......@@ -84,6 +84,20 @@ is based on [Keep a Changelog](https://keepachangelog.com).
+ `illegal_escape_sequence` => `invalid_escape_sequence`
+ `illegal_argument` => `invalid_argument`
+ `illegal_category` => `invalid_category`
- CAF no longer automagically flattens `tuple`, `optional`, or `expected` when
returning these types from message handlers. Users can simply replace
`std::tuple<A, B, C>` with `caf::result<A, B, C>` for returning more than one
value from a message handler.
- A `caf::result` can no longer represent `skip`. Whether a message gets skipped
or not is now only for the default handler to decide. Consequently, default
handlers now return `skippable_result` instead of `result<message>`. A
skippable result is a variant over `delegated<message>`, `message`, `error`,
or `skip_t`. The only good use case for message handlers that skip a message
in their body was in typed actors for getting around the limitation that a
typed behavior always must provide all message handlers (typed behavior assume
a complete implementation of the interface). This use case received direct
support: constructing a typed behavior with `partial_behavior_init` as first
argument suppresses the check for completeness .
### Removed
......@@ -93,6 +107,9 @@ is based on [Keep a Changelog](https://keepachangelog.com).
versions of the standard were never implemented in the first place.
Consequently, we've dropped the `opencl` module.
- The old `duration` type is now superseded by `timespan` (#994).
- The enum `match_result` became obsolete. Individual message handlers can no
longer skip messages. Hence, message handlers can only succeed (match) or not.
Consequently, invoking a message handler or behavior now returns a boolean.
### Fixed
......
......@@ -43,9 +43,9 @@ chopstick::behavior_type taken_chopstick(chopstick::pointer,
// either taken by a philosopher or available
chopstick::behavior_type available_chopstick(chopstick::pointer self) {
return {
[=](take_atom) -> std::tuple<taken_atom, bool> {
[=](take_atom) -> result<taken_atom, bool> {
self->become(taken_chopstick(self, self->current_sender()));
return std::make_tuple(taken_atom_v, true);
return {taken_atom_v, true};
},
[](put_atom) { cerr << "chopstick received unexpected 'put'" << endl; },
};
......@@ -54,8 +54,8 @@ chopstick::behavior_type available_chopstick(chopstick::pointer self) {
chopstick::behavior_type taken_chopstick(chopstick::pointer self,
const strong_actor_ptr& user) {
return {
[](take_atom) -> std::tuple<taken_atom, bool> {
return std::make_tuple(taken_atom_v, false);
[](take_atom) -> result<taken_atom, bool> {
return {taken_atom_v, false};
},
[=](put_atom) {
if (self->current_sender() == user)
......
......@@ -3,17 +3,14 @@
using namespace caf;
behavior server(event_based_actor* self) {
self->set_default_handler(skip);
return {
[=](idle_atom, const actor& worker) {
self->become(
keep_behavior,
[=](ping_atom atm) {
self->become(keep_behavior, [=](ping_atom atm) {
self->delegate(worker, atm);
self->unbecome();
});
},
[=](idle_atom, const actor&) { return skip(); });
},
[=](ping_atom) { return skip(); },
};
}
......
......@@ -23,7 +23,7 @@ behavior ping(event_based_actor* self, actor pong_actor, int n) {
behavior pong() {
return {
[=](ping_atom, int x) { return std::make_tuple(pong_atom_v, x); },
[=](ping_atom, int x) { return make_result(pong_atom_v, x); },
};
}
......
......@@ -6,8 +6,6 @@ file(GLOB_RECURSE CAF_CORE_HEADERS "caf/*.hpp")
add_enum_consistency_check("caf/sec.hpp" "src/sec_strings.cpp")
add_enum_consistency_check("caf/pec.hpp" "src/pec_strings.cpp")
add_enum_consistency_check("caf/match_result.hpp"
"src/match_result_strings.cpp")
add_enum_consistency_check("caf/stream_priority.hpp"
"src/stream_priority_strings.cpp")
add_enum_consistency_check("caf/exit_reason.hpp"
......@@ -111,7 +109,6 @@ set(CAF_CORE_SOURCES
src/logger.cpp
src/mailbox_element.cpp
src/make_config_option.cpp
src/match_result_strings.cpp
src/memory_managed.cpp
src/message.cpp
src/message_builder.cpp
......
......@@ -99,8 +99,8 @@ public:
}
/// Runs this handler with callback.
match_result operator()(detail::invoke_result_visitor& f, message& xs) {
return impl_ ? impl_->invoke(f, xs) : match_result::no_match;
bool operator()(detail::invoke_result_visitor& f, message& xs) {
return impl_ ? impl_->invoke(f, xs) : false;
}
/// Checks whether this behavior is not empty.
......
......@@ -28,7 +28,11 @@ namespace caf {
template <class F>
struct catch_all {
using fun_type = std::function<result<message>(message&)>;
using fun_type = std::function<skippable_result(message&)>;
static_assert(std::is_convertible<F, fun_type>::value,
"catch-all handler must have signature "
"skippable_result (message&)");
F handler;
......@@ -41,10 +45,6 @@ struct catch_all {
// nop
}
static_assert(std::is_convertible<F, fun_type>::value,
"catch-all handler must have signature "
"result<message> (message&)");
fun_type lift() const {
return handler;
}
......
......@@ -31,7 +31,6 @@
#include "caf/detail/type_traits.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/make_counted.hpp"
#include "caf/match_result.hpp"
#include "caf/message.hpp"
#include "caf/none.hpp"
#include "caf/optional.hpp"
......@@ -62,10 +61,9 @@ public:
explicit behavior_impl(timespan tout);
match_result invoke_empty(detail::invoke_result_visitor& f);
bool invoke_empty(detail::invoke_result_visitor& f);
virtual match_result invoke(detail::invoke_result_visitor& f, message& xs)
= 0;
virtual bool invoke(detail::invoke_result_visitor& f, message& xs) = 0;
optional<message> invoke(message&);
......@@ -123,15 +121,13 @@ public:
// nop
}
virtual match_result invoke(detail::invoke_result_visitor& f,
message& xs) override {
virtual bool invoke(detail::invoke_result_visitor& f, message& xs) override {
return invoke_impl(f, xs, std::make_index_sequence<sizeof...(Ts)>{});
}
template <size_t... Is>
match_result invoke_impl(detail::invoke_result_visitor& f, message& msg,
bool invoke_impl(detail::invoke_result_visitor& f, message& msg,
std::index_sequence<Is...>) {
auto result = match_result::no_match;
auto dispatch = [&](auto& fun) {
using fun_type = std::decay_t<decltype(fun)>;
using trait = get_callable_trait_t<fun_type>;
......@@ -141,18 +137,16 @@ public:
using fun_result = decltype(detail::apply_args(fun, xs));
if constexpr (std::is_same<void, fun_result>::value) {
detail::apply_args(fun, xs);
result = f.visit(unit) ? match_result::match : match_result::skip;
f(unit);
} else {
auto invoke_res = detail::apply_args(fun, xs);
result = f.visit(invoke_res) ? match_result::match
: match_result::skip;
f(invoke_res);
}
return true;
}
return false;
};
static_cast<void>((dispatch(std::get<Is>(cases_)) || ...));
return result;
return (dispatch(std::get<Is>(cases_)) || ...);
}
void handle_timeout() override {
......
......@@ -35,7 +35,7 @@ public:
virtual ~blocking_behavior();
virtual result<message> fallback(message&);
virtual skippable_result fallback(message&);
virtual timespan timeout();
......@@ -54,7 +54,7 @@ public:
blocking_behavior_v2(blocking_behavior_v2&&) = default;
result<message> fallback(message& x) override {
skippable_result fallback(message& x) override {
return f.handler(x);
}
};
......@@ -93,7 +93,7 @@ public:
blocking_behavior_v4(blocking_behavior_v4&&) = default;
result<message> fallback(message& x) override {
skippable_result fallback(message& x) override {
return f1.handler(x);
}
......
......@@ -27,7 +27,9 @@ namespace caf::detail {
template <class Self>
class default_invoke_result_visitor : public invoke_result_visitor {
public:
inline default_invoke_result_visitor(Self* ptr) : self_(ptr) {
using super = invoke_result_visitor;
default_invoke_result_visitor(Self* ptr) : self_(ptr) {
// nop
}
......@@ -35,9 +37,7 @@ public:
// nop
}
void operator()() override {
// nop
}
using super::operator();
void operator()(error& x) override {
CAF_LOG_TRACE(CAF_ARG(x));
......@@ -49,11 +49,6 @@ public:
delegate(x);
}
void operator()(const none_t& x) override {
CAF_LOG_TRACE(CAF_ARG(x));
delegate(x);
}
private:
void deliver(response_promise& rp, error& x) {
CAF_LOG_DEBUG("report error back to requesting actor");
......@@ -68,11 +63,6 @@ private:
rp.deliver(std::move(x));
}
void deliver(response_promise& rp, const none_t&) {
error err = sec::unexpected_response;
deliver(rp, err);
}
template <class T>
void delegate(T& x) {
auto rp = self_->make_response_promise();
......
......@@ -25,7 +25,6 @@
#include "caf/detail/int_list.hpp"
#include "caf/detail/overload.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/expected.hpp"
#include "caf/fwd.hpp"
#include "caf/make_sink_result.hpp"
#include "caf/make_source_result.hpp"
......@@ -51,47 +50,13 @@ public:
// -- virtual handlers -------------------------------------------------------
/// Called whenever no result messages gets produced, e.g., when returning a
/// `response_promise`.
virtual void operator()() = 0;
/// Called if the message handler returned an error.
virtual void operator()(error&) = 0;
/// Called if the message handler returned any "ordinary" value.
virtual void operator()(message&) = 0;
/// Called if the message handler returns "nothing", for example a
/// default-constructed `optional<T>`.
virtual void operator()(const none_t&) = 0;
// -- on-the-fly type conversions --------------------------------------------
/// Called if the message handler returns `void` or `unit_t`.
inline void operator()(const unit_t&) {
message empty_msg;
(*this)(empty_msg);
}
/// Unwraps an `optional<T>` by recursively calling the visitor with either
/// `none_t` or `T`.
template <class T>
void operator()(optional<T>& x) {
if (x)
(*this)(*x);
else
(*this)(none);
}
/// Unwraps an `expected<T>` by recursively calling the visitor with either
/// `error` or `T`.
template <class T>
void operator()(expected<T>& x) {
if (x)
(*this)(*x);
else
(*this)(x.error());
}
// -- extraction and conversions ---------------------------------------------
/// Wraps arbitrary values into a `message` and calls the visitor recursively.
template <class... Ts>
......@@ -103,114 +68,62 @@ public:
(*this)(tmp);
}
/// Wraps the tuple into a `message` and calls the visitor recursively with
/// its contents.
template <class... Ts>
void operator()(std::tuple<Ts...>& xs) {
apply_args(*this, get_indices(xs), xs);
/// Called if the message handler returns `void` or `unit_t`.
void operator()(const unit_t&) {
message empty_msg;
(*this)(empty_msg);
}
/// Disambiguates the variadic `operator<Ts...>()`.
inline void operator()(none_t& x) {
(*this)(const_cast<const none_t&>(x));
void operator()(unit_t& x) {
(*this)(const_cast<const unit_t&>(x));
}
/// Disambiguates the variadic `operator<Ts...>()`.
inline void operator()(unit_t& x) {
(*this)(const_cast<const unit_t&>(x));
/// Dispatches on the runtime-type of `x`.
template <class... Ts>
void operator()(result<Ts...>& res) {
caf::visit([this](auto& x) { (*this)(x); }, res);
}
// -- special-purpose handlers that don't produce results --------------------
/// Calls `(*this)()`.
inline void operator()(response_promise&) {
(*this)();
void operator()(response_promise&) {
// nop
}
/// Calls `(*this)()`.
template <class... Ts>
void operator()(typed_response_promise<Ts...>&) {
(*this)();
// nop
}
/// Calls `(*this)()`.
template <class... Ts>
void operator()(delegated<Ts...>&) {
(*this)();
// nop
}
/// Calls `(*this)()`.
template <class Out, class... Ts>
void operator()(outbound_stream_slot<Out, Ts...>&) {
(*this)();
// nop
}
/// Calls `(*this)()`.
template <class In>
void operator()(inbound_stream_slot<In>&) {
(*this)();
// nop
}
/// Calls `(*this)()`.
template <class In>
void operator()(make_sink_result<In>&) {
(*this)();
// nop
}
/// Calls `(*this)()`.
template <class DownstreamManager, class... Ts>
void operator()(make_source_result<DownstreamManager, Ts...>&) {
(*this)();
// nop
}
/// Calls `(*this)()`.
template <class In, class DownstreamManager, class... Ts>
void operator()(make_stage_result<In, DownstreamManager, Ts...>&) {
(*this)();
}
// -- visit API: return true if T was visited, false if T was skipped --------
/// Delegates `x` to the appropriate handler and returns `true`.
template <class T>
bool visit(T& x) {
(*this)(x);
return true;
}
/// Returns `false`.
inline bool visit(skip_t&) {
return false;
}
/// Returns `false`.
inline bool visit(const skip_t&) {
return false;
}
/// Returns `false` if `x != none`, otherwise calls the void handler and
/// returns `true`..
inline bool visit(optional<skip_t>& x) {
if (x)
return false;
(*this)();
return true;
}
/// Dispatches on the runtime-type of `x`.
template <class... Ts>
bool visit(result<Ts...>& res) {
auto f = detail::make_overload(
[this](auto& x) {
(*this)(x);
return true;
},
[this](delegated<Ts...>&) {
(*this)();
return true;
},
[this](skip_t&) { return false; });
return caf::visit(f, res);
// nop
}
};
......
......@@ -87,7 +87,7 @@ struct CAF_CORE_EXPORT function_view_storage_catch_all {
// nop
}
result<message> operator()(message& msg) {
skippable_result operator()(message& msg) {
*storage_ = std::move(msg);
return message{};
}
......
......@@ -135,6 +135,7 @@ class resumable;
class scheduled_actor;
class scoped_actor;
class serializer;
class skip_t;
class stream_manager;
class string_view;
class tracing_data;
......@@ -193,6 +194,7 @@ using ip_address = ipv6_address;
using ip_endpoint = ipv6_endpoint;
using ip_subnet = ipv6_subnet;
using settings = dictionary<config_value>;
using skippable_result = variant<delegated<message>, message, error, skip_t>;
using stream_slot = uint16_t;
using type_id_t = uint16_t;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 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. *
******************************************************************************/
#pragma once
#include <string>
#include "caf/detail/core_export.hpp"
namespace caf {
/// Denotes the invoke result of a ::behavior or ::message_handler.
enum class match_result {
no_match,
match,
skip,
};
/// @relates match_result
CAF_CORE_EXPORT std::string to_string(match_result);
} // namespace caf
......@@ -91,8 +91,8 @@ public:
}
/// Runs this handler with callback.
match_result operator()(detail::invoke_result_visitor& f, message& xs) {
return impl_ ? impl_->invoke(f, xs) : match_result::no_match;
bool operator()(detail::invoke_result_visitor& f, message& xs) {
return impl_ ? impl_->invoke(f, xs) : false;
}
/// Returns a new handler that concatenates this handler
......
......@@ -50,7 +50,7 @@ class result_base {
public:
static_assert(sizeof...(Ts) > 0);
using types = detail::type_list<delegated<Ts...>, message, error, skip_t>;
using types = detail::type_list<delegated<Ts...>, message, error>;
result_base() = default;
......@@ -71,10 +71,6 @@ public:
// nop
}
result_base(skip_t) : content_(skip) {
// nop
}
result_base(delegated<Ts...> x) : content_(x) {
// nop
}
......@@ -109,7 +105,7 @@ protected:
// nop
}
variant<delegated<Ts...>, message, error, skip_t> content_;
variant<delegated<Ts...>, message, error> content_;
};
// -- result<Ts...> and its specializations ------------------------------------
......@@ -236,6 +232,20 @@ public:
}
};
// -- free functions -----------------------------------------------------------
/// Convenience function for wrapping the parameter pack `xs...` into a
/// `result`.
template <class... Ts>
auto make_result(Ts&&... xs) {
return result<std::decay_t<Ts>...>(std::forward<Ts>(xs)...);
}
// -- special type alias for a skippable result<message> -----------------------
/// Similar to `result<message>`, but also allows to *skip* a message.
using skippable_result = variant<delegated<message>, message, error, skip_t>;
// -- sum type access to result<Ts...> -----------------------------------------
template <class... Ts>
......
......@@ -84,21 +84,21 @@ namespace caf {
/// @relates scheduled_actor
/// Default handler function that sends the message back to the sender.
CAF_CORE_EXPORT result<message> reflect(scheduled_actor*, message&);
CAF_CORE_EXPORT skippable_result reflect(scheduled_actor*, message&);
/// @relates scheduled_actor
/// Default handler function that sends
/// the message back to the sender and then quits.
CAF_CORE_EXPORT result<message> reflect_and_quit(scheduled_actor*, message&);
CAF_CORE_EXPORT skippable_result reflect_and_quit(scheduled_actor*, message&);
/// @relates scheduled_actor
/// Default handler function that prints messages
/// message via `aout` and drops them afterwards.
CAF_CORE_EXPORT result<message> print_and_drop(scheduled_actor*, message&);
CAF_CORE_EXPORT skippable_result print_and_drop(scheduled_actor*, message&);
/// @relates scheduled_actor
/// Default handler function that simply drops messages.
CAF_CORE_EXPORT result<message> drop(scheduled_actor*, message&);
CAF_CORE_EXPORT skippable_result drop(scheduled_actor*, message&);
/// A cooperatively scheduled, event-based actor implementation.
class CAF_CORE_EXPORT scheduled_actor : public local_actor,
......@@ -189,7 +189,7 @@ public:
using pointer = scheduled_actor*;
/// Function object for handling unmatched messages.
using default_handler = std::function<result<message>(pointer, message&)>;
using default_handler = std::function<skippable_result(pointer, message&)>;
/// Function object for handling error messages.
using error_handler = std::function<void(pointer, error&)>;
......@@ -215,8 +215,8 @@ public:
size_t max_throughput;
/// Consumes upstream messages.
intrusive::task_result
operator()(size_t, upstream_queue&, mailbox_element&);
intrusive::task_result operator()(size_t, upstream_queue&,
mailbox_element&);
/// Consumes downstream messages.
intrusive::task_result
......@@ -730,8 +730,8 @@ public:
/// Closes all incoming stream traffic for a manager. Emits a drop message on
/// each path if `reason == none` and a `forced_drop` message on each path
/// otherwise.
virtual void
erase_inbound_paths_later(const stream_manager* mgr, error reason);
virtual void erase_inbound_paths_later(const stream_manager* mgr,
error reason);
// -- handling of stream messages --------------------------------------------
......
......@@ -31,20 +31,13 @@ namespace caf {
/// skipping to the runtime.
class CAF_CORE_EXPORT skip_t {
public:
using fun = std::function<result<message>(scheduled_actor* self, message&)>;
using fun = std::function<skippable_result(scheduled_actor* self, message&)>;
constexpr skip_t() {
// nop
}
constexpr skip_t operator()() const {
return *this;
}
operator fun() const;
private:
static result<message> skip_fun_impl(scheduled_actor*, message&);
};
/// Tells the runtime system to skip a message when used as message
......
......@@ -27,9 +27,7 @@
#include "caf/detail/typed_actor_util.hpp"
namespace caf {
namespace detail {
namespace caf ::detail {
// converts a list of replies_to<...>::with<...> elements to a list of
// lists containing the replies_to<...> half only
......@@ -127,15 +125,30 @@ void static_check_typed_behavior_input() {
"typed behavior (exact match needed)");
}
} // namespace detail
} // namespace caf::detail
template <class... Sigs>
class typed_actor;
namespace caf::mixin {
namespace mixin {
template <class, class, class>
class behavior_stack_based_impl;
}
} // namespace caf::mixin
namespace caf {
/// Tag type for constructing a `typed_behavior` with an incomplete list of
/// message handlers, delegating to the default handler for all unmatched
/// inputs.
struct partial_behavior_init_t {};
constexpr partial_behavior_init_t partial_behavior_init
= partial_behavior_init_t{};
/// Empty struct tag for constructing from an untyped behavior.
struct unsafe_behavior_init_t {};
constexpr unsafe_behavior_init_t unsafe_behavior_init
= unsafe_behavior_init_t{};
template <class... Sigs>
class typed_behavior {
......@@ -156,9 +169,6 @@ public:
/// Stores the template parameter pack in a type list.
using signatures = detail::type_list<Sigs...>;
/// Empty struct tag for constructing from an untyped behavior.
struct unsafe_init {};
// -- constructors, destructors, and assignment operators --------------------
typed_behavior(typed_behavior&&) = default;
......@@ -182,11 +192,18 @@ public:
set(detail::make_behavior(std::move(x), std::move(xs)...));
}
typed_behavior(unsafe_init, behavior x) : bhvr_(std::move(x)) {
template <class... Ts>
typed_behavior(partial_behavior_init_t, Ts... xs)
: typed_behavior(unsafe_behavior_init, behavior{std::move(xs)...}) {
// TODO: implement type checking.
}
typed_behavior(unsafe_behavior_init_t, behavior x) : bhvr_(std::move(x)) {
// nop
}
typed_behavior(unsafe_init, message_handler x) : bhvr_(std::move(x)) {
typed_behavior(unsafe_behavior_init_t, message_handler x)
: bhvr_(std::move(x)) {
// nop
}
......
......@@ -158,7 +158,7 @@ behavior spawn_serv_impl(stateful_actor<spawn_serv_state>* self) {
CAF_LOG_TRACE("");
return {
[=](spawn_atom, const std::string& name, message& args,
actor_system::mpi& xs) -> expected<strong_actor_ptr> {
actor_system::mpi& xs) -> result<strong_actor_ptr> {
CAF_LOG_TRACE(CAF_ARG(name) << CAF_ARG(args));
return self->system().spawn<strong_actor_ptr>(name, std::move(args),
self->context(), true, &xs);
......
......@@ -190,19 +190,14 @@ blocking_actor::mailbox_visitor::operator()(mailbox_element& x) {
[&] { self->current_element_ = prev_element; });
// Dispatch on x.
detail::default_invoke_result_visitor<blocking_actor> visitor{self};
switch (bhvr.nested(visitor, x.content())) {
default:
if (bhvr.nested(visitor, x.content()))
return check_if_done();
case match_result::no_match: { // Blocking actors can have fallback
// handlers for catch-all rules.
// Blocking actors can have fallback handlers for catch-all rules.
auto sres = bhvr.fallback(self->current_element_->payload);
if (!holds_alternative<skip_t>(sres)) {
visitor.visit(sres);
return check_if_done();
}
}
// Response handlers must get re-invoked with an error when receiving an
// unexpected message.
auto f = detail::make_overload(
[&](skip_t&) {
// Response handlers must get re-invoked with an error when
// receiving an unexpected message.
if (mid.is_response()) {
auto err = make_error(sec::unexpected_response, std::move(x.payload));
mailbox_element tmp{std::move(x.sender), x.mid, std::move(x.stages),
......@@ -211,10 +206,13 @@ blocking_actor::mailbox_visitor::operator()(mailbox_element& x) {
bhvr.nested(tmp.content());
return check_if_done();
}
CAF_ANNOTATE_FALLTHROUGH;
case match_result::skip:
return intrusive::task_result::skip;
}
},
[&](auto& res) {
visitor(res);
return check_if_done();
});
return visit(f, sres);
};
// Post-process the returned value from the function body.
auto result = body();
......
......@@ -28,9 +28,8 @@ namespace {
class combinator final : public behavior_impl {
public:
match_result invoke(detail::invoke_result_visitor& f, message& xs) override {
auto x = first->invoke(f, xs);
return x == match_result::no_match ? second->invoke(f, xs) : x;
bool invoke(detail::invoke_result_visitor& f, message& xs) override {
return first->invoke(f, xs) || second->invoke(f, xs);
}
void handle_timeout() override {
......@@ -53,10 +52,6 @@ class maybe_message_visitor : public detail::invoke_result_visitor {
public:
optional<message> value;
void operator()() override {
value = message{};
}
void operator()(error& x) override {
value = make_message(std::move(x));
}
......@@ -64,10 +59,6 @@ public:
void operator()(message& x) override {
value = std::move(x);
}
void operator()(const none_t&) override {
(*this)();
}
};
} // namespace
......@@ -84,27 +75,20 @@ behavior_impl::behavior_impl(timespan tout) : timeout_(tout) {
// nop
}
match_result behavior_impl::invoke_empty(detail::invoke_result_visitor& f) {
bool behavior_impl::invoke_empty(detail::invoke_result_visitor& f) {
message xs;
return invoke(f, xs);
}
optional<message> behavior_impl::invoke(message& xs) {
maybe_message_visitor f;
// the following const-cast is safe, because invoke() is aware of
// copy-on-write and does not modify x if it's shared
if (!xs.empty())
invoke(f, xs);
else
invoke_empty(f);
if (invoke(f, xs))
return std::move(f.value);
return none;
}
match_result behavior_impl::invoke(detail::invoke_result_visitor& f,
message& xs) {
if (!xs.empty())
bool behavior_impl::invoke(detail::invoke_result_visitor& f, message& xs) {
return invoke(f, xs);
return invoke_empty(f);
}
void behavior_impl::handle_timeout() {
......
......@@ -28,7 +28,7 @@ blocking_behavior::blocking_behavior(behavior& x) : nested(x) {
// nop
}
result<message> blocking_behavior::fallback(message&) {
skippable_result blocking_behavior::fallback(message&) {
return skip;
}
......
......@@ -161,9 +161,9 @@ public:
CAF_LOG_TRACE("");
// instead of dropping "unexpected" messages,
// we simply forward them to our acquaintances
auto fwd = [=](scheduled_actor*, message& msg) -> result<message> {
auto fwd = [=](scheduled_actor*, message& msg) -> skippable_result {
send_to_acquaintances(std::move(msg));
return message{};
return delegated<message>{};
};
set_default_handler(fwd);
set_down_handler([=](down_msg& dm) {
......@@ -313,7 +313,7 @@ behavior proxy_broker::make_behavior() {
CAF_LOG_TRACE("");
// instead of dropping "unexpected" messages,
// we simply forward them to our acquaintances
auto fwd = [=](local_actor*, message& msg) -> result<message> {
auto fwd = [=](local_actor*, message& msg) -> skippable_result {
group_->send_all_subscribers(current_element_->sender, std::move(msg),
context());
return message{};
......
// clang-format off
// DO NOT EDIT: this file is auto-generated by caf-generate-enum-strings.
// Run the target update-enum-strings if this file is out of sync.
#include "caf/match_result.hpp"
#include <string>
namespace caf {
std::string to_string(match_result x) {
switch(x) {
default:
return "???";
case match_result::no_match:
return "no_match";
case match_result::match:
return "match";
case match_result::skip:
return "skip";
};
}
} // namespace caf
......@@ -82,38 +82,27 @@ invoke_message_result raw_event_based_actor::consume(mailbox_element& x) {
}
// handle everything else as ordinary message
detail::default_invoke_result_visitor<event_based_actor> visitor{this};
bool skipped = false;
auto had_timeout = getf(has_timeout_flag);
if (had_timeout)
unsetf(has_timeout_flag);
// restore timeout at scope exit if message was skipped
auto timeout_guard = detail::make_scope_guard([&] {
if (skipped && had_timeout)
setf(has_timeout_flag);
});
auto call_default_handler = [&] {
auto sres = call_handler(default_handler_, this, x.payload);
auto f = detail::make_overload([&](auto& x) { visitor.visit(x); },
[&](skip_t& x) { skipped = true; });
visit(f, sres);
};
if (bhvr_stack_.empty()) {
call_default_handler();
return !skipped ? invoke_message_result::consumed
: invoke_message_result::skipped;
}
if (!bhvr_stack_.empty()) {
auto& bhvr = bhvr_stack_.back();
switch (bhvr(visitor, x.content())) {
default:
break;
case match_result::skip:
skipped = true;
break;
case match_result::no_match:
call_default_handler();
if (bhvr(visitor, x.content()))
return invoke_message_result::consumed;
}
return !skipped ? invoke_message_result::consumed
: invoke_message_result::skipped;
auto sres = call_handler(default_handler_, this, x.payload);
auto f = detail::make_overload(
[&](auto& x) {
visitor(x);
return invoke_message_result::consumed;
},
[&](skip_t& x) {
// Restore timeout if message was skipped.
if (had_timeout)
setf(has_timeout_flag);
return invoke_message_result::skipped;
});
return visit(f, sres);
};
// Post-process the returned value from the function body.
auto result = body();
......
......@@ -36,26 +36,26 @@ namespace caf {
// -- related free functions ---------------------------------------------------
result<message> reflect(scheduled_actor*, message& msg) {
skippable_result reflect(scheduled_actor*, message& msg) {
return std::move(msg);
}
result<message> reflect_and_quit(scheduled_actor* ptr, message& msg) {
skippable_result reflect_and_quit(scheduled_actor* ptr, message& msg) {
error err = exit_reason::normal;
scheduled_actor::default_error_handler(ptr, err);
return reflect(ptr, msg);
}
result<message> print_and_drop(scheduled_actor* ptr, message& msg) {
skippable_result print_and_drop(scheduled_actor* ptr, message& msg) {
CAF_LOG_WARNING("unexpected message:" << msg);
aout(ptr) << "*** unexpected message [id: " << ptr->id()
<< ", name: " << ptr->name() << "]: " << to_string(msg)
<< std::endl;
return sec::unexpected_message;
return make_error(sec::unexpected_message);
}
result<message> drop(scheduled_actor*, message&) {
return sec::unexpected_message;
skippable_result drop(scheduled_actor*, message&) {
return make_error(sec::unexpected_message);
}
// -- implementation details ---------------------------------------------------
......@@ -67,7 +67,7 @@ void silently_ignore(scheduled_actor*, T&) {
// nop
}
result<message> drop_after_quit(scheduled_actor* self, message&) {
skippable_result drop_after_quit(scheduled_actor* self, message&) {
if (self->current_message_id().is_request())
return make_error(sec::request_receiver_down);
return make_message();
......@@ -691,38 +691,26 @@ invoke_message_result scheduled_actor::consume(mailbox_element& x) {
return invoke_message_result::consumed;
case message_category::ordinary: {
detail::default_invoke_result_visitor<scheduled_actor> visitor{this};
bool skipped = false;
auto had_timeout = getf(has_timeout_flag);
if (had_timeout)
unsetf(has_timeout_flag);
// restore timeout at scope exit if message was skipped
auto timeout_guard = detail::make_scope_guard([&] {
if (skipped && had_timeout)
setf(has_timeout_flag);
});
auto call_default_handler = [&] {
auto f = detail::make_overload([&](auto& val) { visitor.visit(val); },
[&](skip_t&) { skipped = true; });
auto sres = call_handler(default_handler_, this, x.payload);
visit(f, sres);
};
if (bhvr_stack_.empty()) {
call_default_handler();
return !skipped ? invoke_message_result::consumed
: invoke_message_result::skipped;
}
if (!bhvr_stack_.empty()) {
auto& bhvr = bhvr_stack_.back();
switch (bhvr(visitor, x.content())) {
default:
break;
case match_result::skip:
skipped = true;
break;
case match_result::no_match:
call_default_handler();
if (bhvr(visitor, x.content()))
return invoke_message_result::consumed;
}
return !skipped ? invoke_message_result::consumed
: invoke_message_result::skipped;
auto sres = call_handler(default_handler_, this, x.payload);
auto f = detail::make_overload(
[&](auto& x) {
visitor(x);
return invoke_message_result::consumed;
},
[&](skip_t&) {
if (had_timeout)
setf(has_timeout_flag);
return invoke_message_result::skipped;
});
return visit(f, sres);
}
}
// Unreachable.
......@@ -1100,10 +1088,6 @@ scheduled_actor::handle_open_stream_msg(mailbox_element& x) {
CAF_LOG_TRACE(CAF_ARG(x));
// Fetches a stream manager from a behavior.
struct visitor : detail::invoke_result_visitor {
void operator()() override {
// nop
}
void operator()(error&) override {
// nop
}
......@@ -1111,10 +1095,6 @@ scheduled_actor::handle_open_stream_msg(mailbox_element& x) {
void operator()(message&) override {
// nop
}
void operator()(const none_t&) override {
// nop
}
};
// Extract the handshake part of the message.
CAF_ASSERT(x.content().match_elements<open_stream_msg>());
......@@ -1128,35 +1108,20 @@ scheduled_actor::handle_open_stream_msg(mailbox_element& x) {
auto rp = make_response_promise();
rp.deliver(sec::stream_init_failed);
};
// Utility for invoking the default handler.
auto fallback = [&] {
// Invoke behavior and dispatch on the result.
auto& bs = bhvr_stack();
if (!bs.empty() && bs.back()(f, osm.msg))
return invoke_message_result::consumed;
CAF_LOG_DEBUG("no match in behavior, fall back to default handler");
auto sres = call_handler(default_handler_, this, x.payload);
if (holds_alternative<skip_t>(sres)) {
CAF_LOG_DEBUG("default handler skipped open_stream_msg:" << osm.msg);
return invoke_message_result::skipped;
} else {
CAF_LOG_DEBUG(
"default handler was called for open_stream_msg:" << osm.msg);
CAF_LOG_DEBUG("default handler was called for open_stream_msg:" << osm.msg);
fail(sec::stream_init_failed, "dropped open_stream_msg (no match)");
return invoke_message_result::dropped;
}
};
// Invoke behavior and dispatch on the result.
auto& bs = bhvr_stack();
if (bs.empty())
return fallback();
auto res = (bs.back())(f, osm.msg);
switch (res) {
case match_result::no_match:
CAF_LOG_DEBUG("no match in behavior, fall back to default handler");
return fallback();
case match_result::match: {
return invoke_message_result::consumed;
}
default:
CAF_LOG_DEBUG("behavior skipped open_stream_msg:" << osm.msg);
return invoke_message_result::skipped; // nop
}
}
actor_clock::time_point
......
......@@ -23,10 +23,14 @@
namespace caf {
result<message> skip_t::skip_fun_impl(scheduled_actor*, message&) {
return skip();
namespace {
skippable_result skip_fun_impl(scheduled_actor*, message&) {
return skip;
}
} // namespace
skip_t::operator fun() const {
return skip_fun_impl;
}
......
......@@ -43,9 +43,9 @@ CAF_TEST_FIXTURE_SCOPE(blocking_actor_tests, fixture)
CAF_TEST(catch_all) {
self->send(self, 42);
self->receive([](float) { CAF_FAIL("received unexpected float"); },
others >> [](message& msg) -> result<message> {
others >> [](message& msg) -> skippable_result {
CAF_CHECK_EQUAL(to_string(msg), "(42)");
return sec::unexpected_message;
return make_error(sec::unexpected_message);
});
self->receive(
[](const error& err) {
......@@ -75,110 +75,4 @@ CAF_TEST(timeout_in_scoped_actor) {
CAF_CHECK(timeout_called);
}
// -- scoped_actors using skip -------------------------------------------------
using msg_t = int;
// send_order_t contains messages which are send to an actor in the same order
// as in vector
using send_order_t = std::vector<msg_t>;
// sequence_t contains a number of messages for processing by an actor with
// the information to skip the current message for later processing
using sequence_t = std::vector<std::pair<msg_t, bool>>;
using check_order_t = std::pair<send_order_t, sequence_t>;
behavior check_order_behavior_factory(local_actor*,
sequence_t::const_iterator* seq_it_ptr) {
return {
[=](int i) -> result<void> {
auto& seq_it = *seq_it_ptr;
CAF_CHECK_EQUAL(i, seq_it->first);
if (seq_it->second) {
CAF_MESSAGE("current: " << i << "; awaiting: " << seq_it->first
<< " SKIPPED");
++seq_it;
return skip();
} else {
CAF_MESSAGE("current: " << i << "; awaiting: " << seq_it->first
<< " OK");
++seq_it;
return unit;
}
}
};
}
void check_order_event_based_actor(const check_order_t& corder) {
actor_system_config cfg;
actor_system system{cfg};
auto& send_order = corder.first;
auto& sequence = corder.second;
auto seq_it = sequence.cbegin();
{
auto tmp = system.spawn(
[&](event_based_actor* self) {
self->set_default_handler(skip);
for(auto i : send_order) {
self->send(self, i);
}
self->become(
check_order_behavior_factory(self, &seq_it));
}
);
}
system.await_all_actors_done();
}
void check_order_scoped_actor(const check_order_t& corder) {
actor_system_config cfg;
actor_system system{cfg};
auto& send_order = corder.first;
auto& sequence = corder.second;
auto seq_it = begin(sequence);
scoped_actor self{system};
auto check_order_behavior =
check_order_behavior_factory(actor_cast<local_actor*>(self), &seq_it);
for(auto i : send_order) {
self->send(self, i);
}
while (seq_it != end(sequence)) {
self->receive(check_order_behavior);
}
}
CAF_TEST(skip_message) {
check_order_t a = {
{0, 1, 2, 3}, //recv_order = 0,1,2,3
{{0, false}, {1, false}, {2, false}, {3, false}}
};
check_order_t b = {
{3, 2, 1, 0}, //recv_order = 0,1,2,3
{{3, true}, {2, true}, {1, true}, {0, false},
{3, true}, {2, true}, {1, false},
{3, true}, {2, false},
{3, false}}
};
check_order_t c = {
{1, 0, 2}, //recv_order = 0,1,2
{{1, true}, {0, false},
{1, false},
{2, false}}
};
check_order_t d = {
{3, 1, 2, 0}, //recv_order = 0,1,2,3
{{3, true}, {1, true}, {2, true}, {0, false},
{3, true}, {1, false},
{3, true}, {2, false},
{3, false}}
};
check_order_event_based_actor(a);
check_order_event_based_actor(b);
check_order_event_based_actor(c);
check_order_event_based_actor(d);
check_order_scoped_actor(a);
check_order_scoped_actor(b);
check_order_scoped_actor(c);
check_order_scoped_actor(d);
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -42,7 +42,9 @@ using second_stage = typed_actor<replies_to<double, double>::with<double>>;
first_stage::behavior_type typed_first_stage() {
return {
[](int i) { return std::make_tuple(i * 2.0, i * 4.0); },
[](int i) -> result<double, double> {
return {i * 2.0, i * 4.0};
},
};
}
......
......@@ -370,7 +370,7 @@ CAF_TEST(spawn_event_testee2_test) {
CAF_TEST(function_spawn) {
scoped_actor self{system};
auto f = [](const std::string& name) -> behavior {
return ([name](get_atom) { return std::make_tuple(name_atom_v, name); });
return ([name](get_atom) { return make_result(name_atom_v, name); });
};
auto a1 = system.spawn(f, "alice");
auto a2 = system.spawn(f, "bob");
......
......@@ -45,9 +45,9 @@ calculator::behavior_type multiplier() {
calculator::behavior_type divider() {
return {
[](int x, int y) -> optional<int> {
[](int x, int y) -> result<int> {
if (y == 0)
return none;
return make_error(sec::runtime_error, "division by zero");
return x / y;
},
};
......@@ -57,7 +57,9 @@ using doubler = typed_actor<replies_to<int>::with<int, int>>;
doubler::behavior_type simple_doubler() {
return {
[](int x) { return std::make_tuple(x, x); },
[](int x) -> result<int, int> {
return {x, x};
},
};
}
......
......@@ -165,34 +165,18 @@ CAF_TEST(typed_behavior_assignment) {
replies_to<double, double>::with<int, int>>;
// compatible handlers resulting in perfect match
auto f1 = [=](int) { return 0.; };
auto f2 = [=](double, double) { return std::make_tuple(0, 0); };
// compatible handlers using skip
auto g1 = [=](int) { return skip(); };
auto g2 = [=](double, double) { return skip(); };
auto f2 = [=](double, double) -> result<int, int> { return {0, 0}; };
// incompatbile handlers
auto e1 = [=](int) { return 0.f; };
auto e2 = [=](double, double) { return std::make_tuple(0.f, 0.f); };
// omit one handler
CAF_CHECK_EQUAL(bi_pair(false, -1), tb_assign<bh1>(f1));
CAF_CHECK_EQUAL(bi_pair(false, -1), tb_assign<bh1>(f2));
CAF_CHECK_EQUAL(bi_pair(false, -1), tb_assign<bh1>(g1));
CAF_CHECK_EQUAL(bi_pair(false, -1), tb_assign<bh1>(g2));
CAF_CHECK_EQUAL(bi_pair(false, -1), tb_assign<bh1>(e1));
CAF_CHECK_EQUAL(bi_pair(false, -1), tb_assign<bh1>(e2));
// any valid alteration of (f1, f2, g1, g2)
// any valid alteration of (f1, f2)
CAF_CHECK_EQUAL(bi_pair(true, 2), tb_assign<bh1>(f1, f2));
CAF_CHECK_EQUAL(bi_pair(true, 2), tb_assign<bh1>(f2, f1));
CAF_CHECK_EQUAL(bi_pair(true, 2), tb_assign<bh1>(g1, g2));
CAF_CHECK_EQUAL(bi_pair(true, 2), tb_assign<bh1>(g2, g1));
CAF_CHECK_EQUAL(bi_pair(true, 2), tb_assign<bh1>(g1, f2));
CAF_CHECK_EQUAL(bi_pair(true, 2), tb_assign<bh1>(f2, g1));
CAF_CHECK_EQUAL(bi_pair(true, 2), tb_assign<bh1>(f1, g2));
CAF_CHECK_EQUAL(bi_pair(true, 2), tb_assign<bh1>(g2, f1));
// any invalid alteration of (f1, f2, g1, g2)
CAF_CHECK_EQUAL(bi_pair(false, 1), tb_assign<bh1>(f1, g1));
CAF_CHECK_EQUAL(bi_pair(false, 1), tb_assign<bh1>(g1, f1));
CAF_CHECK_EQUAL(bi_pair(false, 1), tb_assign<bh1>(f2, g2));
CAF_CHECK_EQUAL(bi_pair(false, 1), tb_assign<bh1>(g2, g2));
// any invalid alteration of (f1, f2, e1, e2)
CAF_CHECK_EQUAL(bi_pair(false, 1), tb_assign<bh1>(f1, e1));
CAF_CHECK_EQUAL(bi_pair(false, 1), tb_assign<bh1>(f1, e2));
......
......@@ -32,17 +32,10 @@ template<class T>
void test_unit_void() {
auto x = result<T>{};
CAF_CHECK(holds_alternative<message>(x));
x = skip;
CAF_CHECK(holds_alternative<skip_t>(x));
}
} // namespace anonymous
CAF_TEST(skip) {
auto x = result<int>{skip};
CAF_CHECK(holds_alternative<skip_t>(x));
}
CAF_TEST(value) {
auto x = result<int>{42};
CAF_REQUIRE(holds_alternative<message>(x));
......
......@@ -108,36 +108,33 @@ using event_testee_type
class event_testee : public event_testee_type::base {
public:
event_testee(actor_config& cfg) : event_testee_type::base(cfg) {
// nop
set_default_handler(skip);
}
behavior_type wait4string() {
return {
partial_behavior_init,
[=](get_state_atom) { return "wait4string"; },
[=](const string&) { become(wait4int()); },
[=](float) { return skip(); },
[=](int) { return skip(); },
};
}
behavior_type wait4int() {
return {
partial_behavior_init,
[=](get_state_atom) { return "wait4int"; },
[=](int) -> int {
become(wait4float());
return 42;
},
[=](float) { return skip(); },
[=](const string&) { return skip(); },
};
}
behavior_type wait4float() {
return {
partial_behavior_init,
[=](get_state_atom) { return "wait4float"; },
[=](float) { become(wait4string()); },
[=](const string&) { return skip(); },
[=](int) { return skip(); },
};
}
......@@ -365,7 +362,7 @@ CAF_TEST(sending_typed_actors_and_down_msg) {
CAF_TEST(check_signature) {
using foo_type = typed_actor<replies_to<put_atom>::with<ok_atom>>;
using foo_result_type = optional<ok_atom>;
using foo_result_type = result<ok_atom>;
using bar_type = typed_actor<reacts_to<ok_atom>>;
auto foo_action = [](foo_type::pointer ptr) -> foo_type::behavior_type {
return {
......
......@@ -366,8 +366,7 @@ behavior basp_broker::make_behavior() {
return unit;
return sec::cannot_close_invalid_port;
},
[=](get_atom,
const node_id& x) -> std::tuple<node_id, std::string, uint16_t> {
[=](get_atom, const node_id& x) -> result<node_id, std::string, uint16_t> {
std::string addr;
uint16_t port = 0;
auto hdl = instance.tbl().lookup_direct(x);
......@@ -375,7 +374,7 @@ behavior basp_broker::make_behavior() {
addr = remote_addr(*hdl);
port = remote_port(*hdl);
}
return std::make_tuple(x, std::move(addr), port);
return {x, std::move(addr), port};
},
[=](tick_atom, size_t interval) {
instance.handle_heartbeat(context());
......
......@@ -126,11 +126,11 @@ behavior peer_acceptor_fun(broker* self, const actor& buddy) {
self->fork(peer_fun, msg.handle, buddy);
self->quit();
},
[=](publish_atom) -> expected<uint16_t> {
auto res = self->add_tcp_doorman(8080);
if (!res)
return std::move(res.error());
[=](publish_atom) -> result<uint16_t> {
if (auto res = self->add_tcp_doorman(8080))
return res->second;
else
return std::move(res.error());
},
};
}
......
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