Commit a441111d authored by Dominik Charousset's avatar Dominik Charousset

Simplify pattern matching, remove `others`

Add new unexpected message handler to actors that centralizes how actors deal
with messages that are not handled in their current behavior. This obsoletes
the previous approach of cluttering the code base with `others >>` handlers.
Relates #444. Also relates #446, since the new interface uses
`type_erased_tuple` and is a first step towards removing `message` from the
interface of actors entirely.

Removing `others` as well as the "advanced pattern matching syntax" from CAF
opens up design space, allows for several optimizations, and improves
compatibility to MSVC.
parent 19c50ef2
Subproject commit 0a3aa77dc06ff89d360a36a3c89d788869cc5d2b Subproject commit 19a1c08d4e2c247997ac63dd4353c5b90f02088e
Subproject commit 61a80ed346cf06d50b46e1aa75cc5d826b2677b8 Subproject commit cef7a924ba08a742f581388dd214c30d09d0a51f
...@@ -152,9 +152,6 @@ behavior broker_impl(broker* self, connection_handle hdl, const actor& buddy) { ...@@ -152,9 +152,6 @@ behavior broker_impl(broker* self, connection_handle hdl, const actor& buddy) {
<< endl; << endl;
// send composed message to our buddy // send composed message to our buddy
self->send(buddy, atm, ival); self->send(buddy, atm, ival);
},
others >> [=](const message& msg) {
aout(self) << "unexpected: " << to_string(msg) << endl;
} }
}; };
} }
...@@ -170,9 +167,6 @@ behavior server(broker* self, const actor& buddy) { ...@@ -170,9 +167,6 @@ behavior server(broker* self, const actor& buddy) {
print_on_exit(impl, "broker_impl"); print_on_exit(impl, "broker_impl");
aout(self) << "quit server (only accept 1 connection)" << endl; aout(self) << "quit server (only accept 1 connection)" << endl;
self->quit(); self->quit();
},
others >> [=](const message& msg) {
aout(self) << "unexpected: " << to_string(msg) << endl;
} }
}; };
} }
......
...@@ -60,9 +60,6 @@ behavior server(broker* self) { ...@@ -60,9 +60,6 @@ behavior server(broker* self) {
aout(self) << "Finished " << *counter << " requests per second." << endl; aout(self) << "Finished " << *counter << " requests per second." << endl;
*counter = 0; *counter = 0;
self->delayed_send(self, std::chrono::seconds(1), tick_atom::value); self->delayed_send(self, std::chrono::seconds(1), tick_atom::value);
},
others >> [=](const message& msg) {
aout(self) << "unexpected: " << to_string(msg) << endl;
} }
}; };
} }
......
...@@ -5,7 +5,7 @@ ...@@ -5,7 +5,7 @@
// This example is partially included in the manual, do not modify // This example is partially included in the manual, do not modify
// without updating the references in the *.tex files! // without updating the references in the *.tex files!
// Manual references: lines 17-21, 24-29, 31-71, 73-107, and 140-146 (Actor.tex) // Manual references: lines 17-21, 24-26, 28-68, 70-104, and 137-143 (Actor.tex)
#include <iostream> #include <iostream>
...@@ -29,16 +29,13 @@ class blocking_calculator; ...@@ -29,16 +29,13 @@ class blocking_calculator;
class typed_calculator; class typed_calculator;
// function-based, dynamically typed, event-based API // function-based, dynamically typed, event-based API
behavior calculator_fun(event_based_actor* self) { behavior calculator_fun(event_based_actor*) {
return behavior{ return behavior{
[](add_atom, int a, int b) { [](add_atom, int a, int b) {
return a + b; return a + b;
}, },
[](sub_atom, int a, int b) { [](sub_atom, int a, int b) {
return a - b; return a - b;
},
others >> [=](const message& msg) {
aout(self) << "received: " << to_string(msg) << endl;
} }
}; };
} }
...@@ -51,9 +48,6 @@ void blocking_calculator_fun(blocking_actor* self) { ...@@ -51,9 +48,6 @@ void blocking_calculator_fun(blocking_actor* self) {
}, },
[](sub_atom, int a, int b) { [](sub_atom, int a, int b) {
return a - b; return a - b;
},
others >> [=](const message& msg) {
aout(self) << "received: " << to_string(msg) << endl;
} }
); );
} }
......
...@@ -105,6 +105,8 @@ private: ...@@ -105,6 +105,8 @@ private:
} }
behavior reconnecting(std::function<void()> continuation = nullptr) { behavior reconnecting(std::function<void()> continuation = nullptr) {
return {};
/* TODO: implement me
using std::chrono::seconds; using std::chrono::seconds;
auto mm = system().middleman().actor_handle(); auto mm = system().middleman().actor_handle();
send(mm, connect_atom::value, host_, port_); send(mm, connect_atom::value, host_, port_);
...@@ -153,6 +155,7 @@ private: ...@@ -153,6 +155,7 @@ private:
// simply ignore all requests until we have a connection // simply ignore all requests until we have a connection
others >> skip_message others >> skip_message
}; };
*/
} }
actor server_; actor server_;
...@@ -235,8 +238,7 @@ void client_repl(actor_system& system, string host, uint16_t port) { ...@@ -235,8 +238,7 @@ void client_repl(actor_system& system, string host, uint16_t port) {
if (x && y && op) if (x && y && op)
anon_send(client, *op, *x, *y); anon_send(client, *op, *x, *y);
} }
}, }
others >> usage
}; };
// read next line, split it, and feed to the eval handler // read next line, split it, and feed to the eval handler
string line; string line;
...@@ -244,7 +246,8 @@ void client_repl(actor_system& system, string host, uint16_t port) { ...@@ -244,7 +246,8 @@ void client_repl(actor_system& system, string host, uint16_t port) {
line = trim(std::move(line)); // ignore leading and trailing whitespaces line = trim(std::move(line)); // ignore leading and trailing whitespaces
std::vector<string> words; std::vector<string> words;
split(words, line, is_any_of(" "), token_compress_on); split(words, line, is_any_of(" "), token_compress_on);
message_builder(words.begin(), words.end()).apply(eval); if (! message_builder(words.begin(), words.end()).apply(eval))
usage();
} }
} }
......
...@@ -57,9 +57,6 @@ void client(event_based_actor* self, const string& name) { ...@@ -57,9 +57,6 @@ void client(event_based_actor* self, const string& name) {
}, },
[=](const group_down_msg& g) { [=](const group_down_msg& g) {
cout << "*** chatroom offline: " << to_string(g.source) << endl; cout << "*** chatroom offline: " << to_string(g.source) << endl;
},
others >> [=](const message& msg) {
cout << "unexpected: " << to_string(msg) << endl;
} }
); );
} }
...@@ -115,13 +112,12 @@ int main(int argc, char** argv) { ...@@ -115,13 +112,12 @@ int main(int argc, char** argv) {
vector<string> words; vector<string> words;
for (istream_iterator<line> i(cin); i != eof; ++i) { for (istream_iterator<line> i(cin); i != eof; ++i) {
auto send_input = [&] { auto send_input = [&] {
if (!i->str.empty()) { if (! i->str.empty())
anon_send(client_actor, broadcast_atom::value, i->str); anon_send(client_actor, broadcast_atom::value, i->str);
}
}; };
words.clear(); words.clear();
split(words, i->str, is_any_of(" ")); split(words, i->str, is_any_of(" "));
message_builder(words.begin(), words.end()).apply({ auto res = message_builder(words.begin(), words.end()).apply({
[&](const string& cmd, const string& mod, const string& id) { [&](const string& cmd, const string& mod, const string& id) {
if (cmd == "/join") { if (cmd == "/join") {
try { try {
...@@ -150,9 +146,10 @@ int main(int argc, char** argv) { ...@@ -150,9 +146,10 @@ int main(int argc, char** argv) {
else { else {
send_input(); send_input();
} }
}, }
others >> send_input
}); });
if (! res)
send_input();
} }
// force actor to quit // force actor to quit
anon_send_exit(client_actor, exit_reason::user_shutdown); anon_send_exit(client_actor, exit_reason::user_shutdown);
......
...@@ -28,6 +28,7 @@ ...@@ -28,6 +28,7 @@
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/message.hpp"
#include "caf/actor_marker.hpp" #include "caf/actor_marker.hpp"
#include "caf/abstract_actor.hpp" #include "caf/abstract_actor.hpp"
#include "caf/actor_control_block.hpp" #include "caf/actor_control_block.hpp"
......
...@@ -54,7 +54,7 @@ public: ...@@ -54,7 +54,7 @@ public:
using portable_names = std::unordered_map<std::type_index, std::string>; using portable_names = std::unordered_map<std::type_index, std::string>;
using error_renderer = std::function<std::string (uint8_t, const std::string&)>; using error_renderer = std::function<std::string (uint8_t)>;
using error_renderers = std::unordered_map<atom_value, error_renderer>; using error_renderers = std::unordered_map<atom_value, error_renderer>;
......
...@@ -17,36 +17,38 @@ ...@@ -17,36 +17,38 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_ANYTHING_HPP #ifndef CAF_AFTER_HPP
#define CAF_ANYTHING_HPP #define CAF_AFTER_HPP
#include <tuple>
#include <type_traits> #include <type_traits>
#include "caf/timeout_definition.hpp"
namespace caf { namespace caf {
/// Acts as wildcard expression in patterns. class timeout_definition_builder {
struct anything { public:
constexpr anything() { constexpr timeout_definition_builder(const duration& d) : tout_(d) {
// nop // nop
} }
};
/// @relates anything
inline bool operator==(const anything&, const anything&) {
return true;
}
/// @relates anything template <class F>
inline bool operator!=(const anything&, const anything&) { timeout_definition<F> operator>>(F f) const {
return false; return {tout_, std::move(f)};
} }
/// @relates anything private:
template <class T> duration tout_;
struct is_anything : std::is_same<T, anything> {
// no content
}; };
/// Returns a generator for timeouts.
template <class Rep, class Period>
constexpr timeout_definition_builder
after(const std::chrono::duration<Rep, Period>& d) {
return {duration(d)};
}
} // namespace caf } // namespace caf
#endif // CAF_ANYTHING_HPP #endif // CAF_AFTER_HPP
...@@ -22,12 +22,12 @@ ...@@ -22,12 +22,12 @@
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/on.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/atom.hpp" #include "caf/atom.hpp"
#include "caf/send.hpp" #include "caf/send.hpp"
#include "caf/unit.hpp" #include "caf/unit.hpp"
#include "caf/actor.hpp" #include "caf/actor.hpp"
#include "caf/after.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/group.hpp" #include "caf/group.hpp"
#include "caf/extend.hpp" #include "caf/extend.hpp"
...@@ -35,7 +35,6 @@ ...@@ -35,7 +35,6 @@
#include "caf/result.hpp" #include "caf/result.hpp"
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/node_id.hpp" #include "caf/node_id.hpp"
#include "caf/anything.hpp"
#include "caf/behavior.hpp" #include "caf/behavior.hpp"
#include "caf/duration.hpp" #include "caf/duration.hpp"
#include "caf/exception.hpp" #include "caf/exception.hpp"
......
...@@ -102,8 +102,9 @@ public: ...@@ -102,8 +102,9 @@ public:
} }
/// Runs this handler with callback. /// Runs this handler with callback.
inline bool operator()(detail::invoke_result_visitor& f, message& x) { inline match_case::result operator()(detail::invoke_result_visitor& f,
return impl_ ? impl_->invoke(f, x) : false; message& x) {
return impl_ ? impl_->invoke(f, x) : match_case::no_match;
} }
/// Checks whether this behavior is not empty. /// Checks whether this behavior is not empty.
......
...@@ -25,7 +25,6 @@ ...@@ -25,7 +25,6 @@
#include "caf/none.hpp" #include "caf/none.hpp"
#include "caf/on.hpp"
#include "caf/extend.hpp" #include "caf/extend.hpp"
#include "caf/behavior.hpp" #include "caf/behavior.hpp"
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
...@@ -192,6 +191,8 @@ public: ...@@ -192,6 +191,8 @@ public:
void dequeue(behavior& bhvr, message_id mid = invalid_message_id); void dequeue(behavior& bhvr, message_id mid = invalid_message_id);
using local_actor::await_data;
/// @endcond /// @endcond
protected: protected:
......
...@@ -42,8 +42,8 @@ ...@@ -42,8 +42,8 @@
#include "caf/detail/int_list.hpp" #include "caf/detail/int_list.hpp"
#include "caf/detail/apply_args.hpp" #include "caf/detail/apply_args.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
#include "caf/detail/invoke_result_visitor.hpp"
#include "caf/detail/tail_argument_token.hpp" #include "caf/detail/tail_argument_token.hpp"
#include "caf/detail/invoke_result_visitor.hpp"
namespace caf { namespace caf {
...@@ -68,9 +68,10 @@ public: ...@@ -68,9 +68,10 @@ public:
behavior_impl(duration tout = duration{}); behavior_impl(duration tout = duration{});
virtual bool invoke(detail::invoke_result_visitor& f, message&); virtual match_case::result invoke(detail::invoke_result_visitor& f, message&);
inline bool invoke(detail::invoke_result_visitor& f, message&& arg) { inline match_case::result invoke(detail::invoke_result_visitor& f,
message&& arg) {
message tmp(std::move(arg)); message tmp(std::move(arg));
return invoke(f, tmp); return invoke(f, tmp);
} }
...@@ -99,7 +100,6 @@ struct defaut_bhvr_impl_init { ...@@ -99,7 +100,6 @@ struct defaut_bhvr_impl_init {
static void init(Array& arr, Tuple& tup) { static void init(Array& arr, Tuple& tup) {
auto& x = arr[Pos]; auto& x = arr[Pos];
x.ptr = &std::get<Pos>(tup); x.ptr = &std::get<Pos>(tup);
x.has_wildcard = x.ptr->has_wildcard();
x.type_token = x.ptr->type_token(); x.type_token = x.ptr->type_token();
defaut_bhvr_impl_init<Pos + 1, Size>::init(arr, tup); defaut_bhvr_impl_init<Pos + 1, Size>::init(arr, tup);
} }
......
...@@ -20,7 +20,6 @@ ...@@ -20,7 +20,6 @@
#ifndef CAF_DETAIL_BOXED_HPP #ifndef CAF_DETAIL_BOXED_HPP
#define CAF_DETAIL_BOXED_HPP #define CAF_DETAIL_BOXED_HPP
#include "caf/anything.hpp"
#include "caf/detail/wrapped.hpp" #include "caf/detail/wrapped.hpp"
namespace caf { namespace caf {
...@@ -42,14 +41,6 @@ struct boxed<detail::wrapped<T>> { ...@@ -42,14 +41,6 @@ struct boxed<detail::wrapped<T>> {
using type = detail::wrapped<T>; using type = detail::wrapped<T>;
}; };
template <>
struct boxed<anything> {
constexpr boxed() {
// nop
}
using type = anything;
};
template <class T> template <class T>
struct is_boxed { struct is_boxed {
static constexpr bool value = false; static constexpr bool value = false;
......
...@@ -30,39 +30,55 @@ namespace detail { ...@@ -30,39 +30,55 @@ namespace detail {
class concatenated_tuple : public message_data { class concatenated_tuple : public message_data {
public: public:
concatenated_tuple& operator=(const concatenated_tuple&) = delete; // -- member types -----------------------------------------------------------
using message_data::cow_ptr; using message_data::cow_ptr;
using vector_type = std::vector<cow_ptr>; using vector_type = std::vector<cow_ptr>;
// -- constructors, destructors, and assignment operators --------------------
concatenated_tuple(std::initializer_list<cow_ptr> xs);
static cow_ptr make(std::initializer_list<cow_ptr> xs); static cow_ptr make(std::initializer_list<cow_ptr> xs);
void* get_mutable(size_t pos) override; concatenated_tuple(const concatenated_tuple&) = default;
void load(size_t pos, deserializer& source) override; concatenated_tuple& operator=(const concatenated_tuple&) = delete;
size_t size() const override; // -- overridden observers of message_data -----------------------------------
cow_ptr copy() const override; cow_ptr copy() const override;
const void* get(size_t pos) const override; // -- overridden modifiers of type_erased_tuple ------------------------------
void* get_mutable(size_t pos) override;
void load(size_t pos, deserializer& source) override;
// -- overridden observers of type_erased_tuple ------------------------------
size_t size() const override;
uint32_t type_token() const override; uint32_t type_token() const override;
rtti_pair type(size_t pos) const override; rtti_pair type(size_t pos) const override;
void save(size_t pos, serializer& sink) const override; const void* get(size_t pos) const override;
std::string stringify(size_t pos) const override; std::string stringify(size_t pos) const override;
concatenated_tuple(std::initializer_list<cow_ptr> xs); type_erased_value_ptr copy(size_t pos) const override;
concatenated_tuple(const concatenated_tuple&) = default; void save(size_t pos, serializer& sink) const override;
// -- observers --------------------------------------------------------------
std::pair<message_data*, size_t> select(size_t pos) const; std::pair<message_data*, size_t> select(size_t pos) const;
private: private:
// -- data members -----------------------------------------------------------
vector_type data_; vector_type data_;
uint32_t type_token_; uint32_t type_token_;
size_t size_; size_t size_;
......
...@@ -36,34 +36,47 @@ namespace detail { ...@@ -36,34 +36,47 @@ namespace detail {
class decorated_tuple : public message_data { class decorated_tuple : public message_data {
public: public:
decorated_tuple& operator=(const decorated_tuple&) = delete; // -- member types -----------------------------------------------------------
using message_data::cow_ptr;
using vector_type = std::vector<size_t>; using vector_type = std::vector<size_t>;
using message_data::cow_ptr; // -- constructors, destructors, and assignment operators --------------------
decorated_tuple(cow_ptr&&, vector_type&&); decorated_tuple(cow_ptr&&, vector_type&&);
// creates a typed subtuple from `d` with mapping `v`
static cow_ptr make(cow_ptr d, vector_type v); static cow_ptr make(cow_ptr d, vector_type v);
decorated_tuple& operator=(const decorated_tuple&) = delete;
// -- overridden observers of message_data -----------------------------------
cow_ptr copy() const override;
// -- overridden modifiers of type_erased_tuple ------------------------------
void* get_mutable(size_t pos) override; void* get_mutable(size_t pos) override;
void load(size_t pos, deserializer& source) override; void load(size_t pos, deserializer& source) override;
// -- overridden observers of type_erased_tuple ------------------------------
size_t size() const override; size_t size() const override;
cow_ptr copy() const override; uint32_t type_token() const override;
rtti_pair type(size_t pos) const override;
const void* get(size_t pos) const override; const void* get(size_t pos) const override;
uint32_t type_token() const override; std::string stringify(size_t pos) const override;
rtti_pair type(size_t pos) const override; type_erased_value_ptr copy(size_t pos) const override;
void save(size_t pos, serializer& sink) const override; void save(size_t pos, serializer& sink) const override;
std::string stringify(size_t pos) const override; // -- inline observers -------------------------------------------------------
inline const cow_ptr& decorated() const { inline const cow_ptr& decorated() const {
return decorated_; return decorated_;
...@@ -74,8 +87,12 @@ public: ...@@ -74,8 +87,12 @@ public:
} }
private: private:
// -- constructors, destructors, and assignment operators --------------------
decorated_tuple(const decorated_tuple&) = default; decorated_tuple(const decorated_tuple&) = default;
// -- data members -----------------------------------------------------------
cow_ptr decorated_; cow_ptr decorated_;
vector_type mapping_; vector_type mapping_;
uint32_t type_token_; uint32_t type_token_;
......
...@@ -31,41 +31,57 @@ namespace detail { ...@@ -31,41 +31,57 @@ namespace detail {
class dynamic_message_data : public message_data { class dynamic_message_data : public message_data {
public: public:
// -- member types -----------------------------------------------------------
using elements = std::vector<type_erased_value_ptr>; using elements = std::vector<type_erased_value_ptr>;
dynamic_message_data(); // -- constructors, destructors, and assignment operators --------------------
dynamic_message_data(const dynamic_message_data& other); dynamic_message_data();
dynamic_message_data(elements&& data); dynamic_message_data(elements&& data);
dynamic_message_data(const dynamic_message_data& other);
~dynamic_message_data(); ~dynamic_message_data();
const void* get(size_t pos) const override; // -- overridden observers of message_data -----------------------------------
void save(size_t pos, serializer& sink) const override; cow_ptr copy() const override;
void load(size_t pos, deserializer& source) override; // -- overridden modifiers of type_erased_tuple ------------------------------
void* get_mutable(size_t pos) override; void* get_mutable(size_t pos) override;
void load(size_t pos, deserializer& source) override;
// -- overridden observers of type_erased_tuple ------------------------------
size_t size() const override; size_t size() const override;
cow_ptr copy() const override; uint32_t type_token() const override;
rtti_pair type(size_t pos) const override; rtti_pair type(size_t pos) const override;
const void* get(size_t pos) const override;
std::string stringify(size_t pos) const override; std::string stringify(size_t pos) const override;
uint32_t type_token() const override; type_erased_value_ptr copy(size_t pos) const override;
void append(type_erased_value_ptr x); void save(size_t pos, serializer& sink) const override;
void add_to_type_token(uint16_t typenr); // -- modifiers --------------------------------------------------------------
void clear(); void clear();
void append(type_erased_value_ptr x);
void add_to_type_token(uint16_t typenr);
private: private:
// -- data members -----------------------------------------------------------
elements elements_; elements elements_;
uint32_t type_token_; uint32_t type_token_;
}; };
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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. *
******************************************************************************/
#ifndef CAF_DETAIL_MATCH_CASE_BUILDER_HPP
#define CAF_DETAIL_MATCH_CASE_BUILDER_HPP
#include <type_traits>
#include "caf/duration.hpp"
#include "caf/match_case.hpp"
#include "caf/timeout_definition.hpp"
#include "caf/detail/tail_argument_token.hpp"
namespace caf {
namespace detail {
class timeout_definition_builder {
public:
constexpr timeout_definition_builder(const duration& d) : tout_(d) {
// nop
}
template <class F>
timeout_definition<F> operator>>(F f) const {
return {tout_, std::move(f)};
}
private:
duration tout_;
};
class message_case_builder { };
class trivial_match_case_builder : public message_case_builder {
public:
constexpr trivial_match_case_builder() {
// nop
}
template <class F>
trivial_match_case<F> operator>>(F f) const {
return {std::move(f)};
}
};
class catch_all_match_case_builder : public message_case_builder {
public:
constexpr catch_all_match_case_builder() {
// nop
}
const catch_all_match_case_builder& operator()() const {
return *this;
}
template <class F>
catch_all_match_case<F> operator>>(F f) const {
return {std::move(f)};
}
};
template <class Left, class Right>
class message_case_pair_builder : public message_case_builder {
public:
message_case_pair_builder(Left l, Right r)
: left_(std::move(l)),
right_(std::move(r)) {
// nop
}
template <class F>
auto operator>>(F f) const
-> std::tuple<decltype(*static_cast<Left*>(nullptr) >> f),
decltype(*static_cast<Right*>(nullptr) >> f)> {
return std::make_tuple(left_ >> f, right_ >> f);
}
private:
Left left_;
Right right_;
};
struct tuple_maker {
template <class... Ts>
inline auto operator()(Ts&&... xs)
-> decltype(std::make_tuple(std::forward<Ts>(xs)...)) {
return std::make_tuple(std::forward<Ts>(xs)...);
}
};
struct variadic_ctor {};
template <class F, class Projections, class Pattern>
struct advanced_match_case_factory {
static constexpr bool uses_arg_match =
std::is_same<
arg_match_t,
typename detail::tl_back<Pattern>::type
>::value;
using fargs = typename get_callable_trait<F>::arg_types;
static constexpr size_t fargs_size =
detail::tl_size<
fargs
>::value;
using decayed_fargs =
typename tl_map<
fargs,
std::decay
>::type;
using concatenated_pattern =
typename detail::tl_concat<
typename detail::tl_pop_back<Pattern>::type,
decayed_fargs
>::type;
using full_pattern =
typename std::conditional<
uses_arg_match,
concatenated_pattern,
Pattern
>::type;
using filtered_pattern =
typename detail::tl_filter_not_type<
full_pattern,
anything
>::type;
using padded_projections =
typename detail::tl_pad_right<
Projections,
detail::tl_size<filtered_pattern>::value
>::type;
using projection_funs =
typename detail::tl_apply<
padded_projections,
std::tuple
>::type;
using tuple_type =
typename detail::tl_apply<
typename detail::tl_zip_right<
typename detail::tl_map<
padded_projections,
projection_result
>::type,
fargs,
detail::left_or_right,
fargs_size
>::type,
std::tuple
>::type;
using padding =
typename detail::tl_apply<
typename detail::tl_replicate<fargs_size, unit_t>::type,
std::tuple
>::type;
static projection_funs pad(projection_funs res, std::false_type) {
return std::move(res);
}
template <class Guards>
static projection_funs pad(Guards gs, std::true_type) {
return std::tuple_cat(std::move(gs), padding{});
}
using case_type = advanced_match_case_impl<F, tuple_type,
full_pattern, projection_funs>;
template <class Guards>
static case_type create(F f, Guards gs) {
std::integral_constant<bool, uses_arg_match> tk;
return {tl_exists<full_pattern, is_anything>::value,
make_type_token_from_list<full_pattern>(), std::move(f),
pad(std::move(gs), tk)};
}
};
template <class X, class Y>
struct pattern_projection_zipper {
using type = Y;
};
template <class Y>
struct pattern_projection_zipper<anything, Y> {
using type = none_t;
};
template <class Projections, class Pattern>
class advanced_match_case_builder : public message_case_builder {
public:
using filtered_projections =
typename tl_filter_not_type<
typename tl_zip<
Pattern,
Projections,
pattern_projection_zipper
>::type,
none_t
>::type;
using guards_tuple =
typename detail::tl_apply<
typename std::conditional<
std::is_same<
arg_match_t,
typename tl_back<Pattern>::type
>::value,
typename tl_pop_back<filtered_projections>::type,
filtered_projections
>::type,
std::tuple
>::type;
template <class... Fs>
advanced_match_case_builder(variadic_ctor, Fs... fs)
: g_(make_guards(Pattern{}, fs...)) {
// nop
}
template <class F>
typename advanced_match_case_factory<F, Projections, Pattern>::case_type
operator>>(F f) const {
return advanced_match_case_factory<F, Projections, Pattern>::create(f, g_);
}
private:
template <class... Fs>
static guards_tuple make_guards(tail_argument_token&, Fs&... fs) {
return std::make_tuple(std::move(fs)...);
}
static guards_tuple make_guards(tail_argument_token&) {
return {};
}
template <class F, class... Fs>
static guards_tuple make_guards(std::pair<wrapped<arg_match_t>, F>,
tail_argument_token&, Fs&... fs) {
return std::make_tuple(std::move(fs)...);
}
template <class T, class F, class... Fs>
static guards_tuple make_guards(std::pair<wrapped<T>, F> x, Fs&&... fs) {
return make_guards(std::forward<Fs>(fs)..., x.second);
}
template <class F, class... Fs>
static guards_tuple make_guards(std::pair<wrapped<anything>, F>, Fs&&... fs) {
return make_guards(std::forward<Fs>(fs)...);
}
template <class... Ts, class... Fs>
static guards_tuple make_guards(type_list<Ts...>, Fs&... fs) {
tail_argument_token eoa;
return make_guards(std::make_pair(wrapped<Ts>(), std::ref(fs))..., eoa);
}
guards_tuple g_;
};
template <class Left, class Right>
typename std::enable_if<
std::is_base_of<message_case_builder, Left>::value
&& std::is_base_of<message_case_builder, Right>::value,
message_case_pair_builder<Left, Right>
>::type
operator||(Left l, Right r) {
return {l, r};
}
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_MATCH_CASE_BUILDER_HPP
...@@ -30,42 +30,59 @@ namespace detail { ...@@ -30,42 +30,59 @@ namespace detail {
class merged_tuple : public message_data { class merged_tuple : public message_data {
public: public:
merged_tuple& operator=(const merged_tuple&) = delete; // -- member types -----------------------------------------------------------
using mapping_type = std::vector<std::pair<size_t, size_t>>;
using message_data::cow_ptr; using message_data::cow_ptr;
using data_type = std::vector<cow_ptr>; using data_type = std::vector<cow_ptr>;
merged_tuple(data_type xs, mapping_type ys); using mapping_type = std::vector<std::pair<size_t, size_t>>;
// -- constructors, destructors, and assignment operators --------------------
// creates a typed subtuple from `d` with mapping `v`
static cow_ptr make(message x, message y); static cow_ptr make(message x, message y);
merged_tuple(data_type xs, mapping_type ys);
merged_tuple& operator=(const merged_tuple&) = delete;
// -- overridden observers of message_data -----------------------------------
cow_ptr copy() const override;
// -- overridden modifiers of type_erased_tuple ------------------------------
void* get_mutable(size_t pos) override; void* get_mutable(size_t pos) override;
void load(size_t pos, deserializer& source) override; void load(size_t pos, deserializer& source) override;
// -- overridden observers of type_erased_tuple ------------------------------
size_t size() const override; size_t size() const override;
cow_ptr copy() const override; uint32_t type_token() const override;
rtti_pair type(size_t pos) const override;
const void* get(size_t pos) const override; const void* get(size_t pos) const override;
uint32_t type_token() const override; std::string stringify(size_t pos) const override;
rtti_pair type(size_t pos) const override; type_erased_value_ptr copy(size_t pos) const override;
void save(size_t pos, serializer& sink) const override; void save(size_t pos, serializer& sink) const override;
std::string stringify(size_t pos) const override; // -- observers --------------------------------------------------------------
const mapping_type& mapping() const; const mapping_type& mapping() const;
private: private:
// -- constructors, destructors, and assignment operators --------------------
merged_tuple(const merged_tuple&) = default; merged_tuple(const merged_tuple&) = default;
// -- data members -----------------------------------------------------------
data_type data_; data_type data_;
uint32_t type_token_; uint32_t type_token_;
mapping_type mapping_; mapping_type mapping_;
......
...@@ -111,6 +111,8 @@ public: ...@@ -111,6 +111,8 @@ public:
intrusive_ptr<message_data> ptr_; intrusive_ptr<message_data> ptr_;
}; };
using type_erased_tuple::copy;
virtual cow_ptr copy() const = 0; virtual cow_ptr copy() const = 0;
}; };
......
...@@ -49,25 +49,29 @@ public: ...@@ -49,25 +49,29 @@ public:
} }
behavior make_behavior() override { behavior make_behavior() override {
auto f = [=](local_actor*, const type_erased_tuple* x) -> result<message> {
auto msg = message::from(x);
auto rp = this->make_response_promise();
split_(workset_, msg);
for (auto& x : workset_)
this->send(x.first, std::move(x.second));
auto g = [=](local_actor*, const type_erased_tuple* x) mutable
-> result<message> {
auto res = message::from(x);
join_(value_, res);
if (--awaited_results_ == 0) {
rp.deliver(value_);
quit();
}
return delegated<message>{};
};
set_unexpected_handler(g);
return delegated<message>{};
};
set_unexpected_handler(f);
return { return {
// first message is the forwarded request [] {
others >> [=](message& msg) { // nop
auto rp = this->make_response_promise();
split_(workset_, msg);
for (auto& x : workset_)
this->send(x.first, std::move(x.second));
this->become(
// collect results
others >> [=](message& res) mutable {
join_(value_, res);
if (--awaited_results_ == 0) {
rp.deliver(make_message(value_));
quit();
}
}
);
// no longer needed
workset_.clear();
} }
}; };
} }
......
...@@ -67,13 +67,6 @@ struct meta_element_factory<atom_constant<V>, type_nr<atom_value>::value> { ...@@ -67,13 +67,6 @@ struct meta_element_factory<atom_constant<V>, type_nr<atom_value>::value> {
} }
}; };
template <>
struct meta_element_factory<anything, 0> {
static meta_element create() {
return {static_cast<atom_value>(0), 0, nullptr, nullptr};
}
};
template <class TypeList> template <class TypeList>
struct meta_elements; struct meta_elements;
......
...@@ -71,6 +71,14 @@ struct tup_ptr_access { ...@@ -71,6 +71,14 @@ struct tup_ptr_access {
return deep_to_string(std::get<Pos>(tup)); return deep_to_string(std::get<Pos>(tup));
return tup_ptr_access<Pos + 1, Max>::stringify(pos, tup); return tup_ptr_access<Pos + 1, Max>::stringify(pos, tup);
} }
template <class T>
static type_erased_value_ptr copy(size_t pos, const T& tup) {
using value_type = typename std::tuple_element<Pos, T>::type;
if (pos == Pos)
return make_type_erased_value<value_type>(std::get<Pos>(tup));
return tup_ptr_access<Pos + 1, Max>::copy(pos, tup);;
}
}; };
template <size_t Pos, size_t Max> template <size_t Pos, size_t Max>
...@@ -100,6 +108,11 @@ struct tup_ptr_access<Pos, Max, false> { ...@@ -100,6 +108,11 @@ struct tup_ptr_access<Pos, Max, false> {
static std::string stringify(size_t, const T&) { static std::string stringify(size_t, const T&) {
return "<unprintable>"; return "<unprintable>";
} }
template <class T>
static type_erased_value_ptr copy(size_t, const T&) {
return nullptr;
}
}; };
template <class T, uint16_t N = detail::type_nr<T>::value> template <class T, uint16_t N = detail::type_nr<T>::value>
...@@ -166,6 +179,10 @@ public: ...@@ -166,6 +179,10 @@ public:
return tup_ptr_access<0, sizeof...(Ts)>::stringify(pos, data_); return tup_ptr_access<0, sizeof...(Ts)>::stringify(pos, data_);
} }
type_erased_value_ptr copy(size_t pos) const override {
return tup_ptr_access<0, sizeof...(Ts)>::copy(pos, data_);
}
void load(size_t pos, deserializer& source) override { void load(size_t pos, deserializer& source) override {
return tup_ptr_access<0, sizeof...(Ts)>::serialize(pos, data_, source); return tup_ptr_access<0, sizeof...(Ts)>::serialize(pos, data_, source);
} }
......
...@@ -67,10 +67,6 @@ struct disjunction<X, Xs...> { ...@@ -67,10 +67,6 @@ struct disjunction<X, Xs...> {
static constexpr bool value = X || disjunction<Xs...>::value; static constexpr bool value = X || disjunction<Xs...>::value;
}; };
/// Equal to std::is_same<T, anything>.
template <class T>
struct is_anything : std::is_same<T, anything> {};
/// Checks whether `T` is an array of type `U`. /// Checks whether `T` is an array of type `U`.
template <class T, typename U> template <class T, typename U>
struct is_array_of { struct is_array_of {
...@@ -117,9 +113,9 @@ template <class T> ...@@ -117,9 +113,9 @@ template <class T>
struct is_builtin { struct is_builtin {
// all arithmetic types are considered builtin // all arithmetic types are considered builtin
static constexpr bool value = std::is_arithmetic<T>::value static constexpr bool value = std::is_arithmetic<T>::value
|| is_one_of<T, anything, std::string, || is_one_of<T, std::string, std::u16string,
std::u16string, std::u32string, std::u32string, atom_value,
atom_value, message, actor, group, message, actor, group,
node_id>::value; node_id>::value;
}; };
......
...@@ -24,6 +24,7 @@ ...@@ -24,6 +24,7 @@
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/atom.hpp" #include "caf/atom.hpp"
#include "caf/message.hpp"
#include "caf/detail/comparable.hpp" #include "caf/detail/comparable.hpp"
...@@ -73,7 +74,7 @@ public: ...@@ -73,7 +74,7 @@ public:
error(const error&) = default; error(const error&) = default;
error& operator=(error&&) = default; error& operator=(error&&) = default;
error& operator=(const error&) = default; error& operator=(const error&) = default;
error(uint8_t code, atom_value category, std::string context = std::string{}); error(uint8_t code, atom_value category, message context = message{});
template <class E, template <class E,
class = typename std::enable_if< class = typename std::enable_if<
...@@ -93,10 +94,10 @@ public: ...@@ -93,10 +94,10 @@ public:
atom_value category() const; atom_value category() const;
/// Returns optional context information to this error. /// Returns optional context information to this error.
std::string& context(); message& context();
/// Returns optional context information to this error. /// Returns optional context information to this error.
const std::string& context() const; const message& context() const;
/// Returns `code() != 0`. /// Returns `code() != 0`.
explicit operator bool() const; explicit operator bool() const;
...@@ -118,7 +119,7 @@ public: ...@@ -118,7 +119,7 @@ public:
private: private:
uint8_t code_; uint8_t code_;
atom_value category_; atom_value category_;
std::string context_; message context_;
}; };
/// @relates error /// @relates error
......
...@@ -22,7 +22,6 @@ ...@@ -22,7 +22,6 @@
#include <type_traits> #include <type_traits>
#include "caf/on.hpp"
#include "caf/extend.hpp" #include "caf/extend.hpp"
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/response_handle.hpp" #include "caf/response_handle.hpp"
......
...@@ -87,7 +87,6 @@ class forwarding_actor_proxy; ...@@ -87,7 +87,6 @@ class forwarding_actor_proxy;
// -- structs ------------------------------------------------------------------ // -- structs ------------------------------------------------------------------
struct unit_t; struct unit_t;
struct anything;
struct exit_msg; struct exit_msg;
struct down_msg; struct down_msg;
struct timeout_msg; struct timeout_msg;
......
...@@ -84,18 +84,52 @@ struct make_response_promise_helper<response_promise> { ...@@ -84,18 +84,52 @@ struct make_response_promise_helper<response_promise> {
} // namespace detail } // namespace detail
/// @relates local_actor
/// Default handler function for unexpected messages
/// that sends the message back to the sender.
result<message> mirror_unexpected(local_actor*, const type_erased_tuple*);
/// @relates local_actor
/// Default handler function for unexpected messages
/// that sends the message back to the sender and then quits.
result<message> mirror_unexpected_once(local_actor*, const type_erased_tuple*);
/// @relates local_actor
/// Default handler function for unexpected messages that
/// skips an unexpected message in order to handle it again
/// after a behavior change.
result<message> skip_unexpected(local_actor*, const type_erased_tuple*);
/// @relates local_actor
/// Default handler function for unexpected messages that prints
/// an unexpected message via `aout` and drops it afterwards.
result<message> print_and_drop_unexpected(local_actor*,
const type_erased_tuple*);
/// @relates local_actor
/// Default handler function for unexpected messages that
/// drops unexpected messages without printing them.
result<message> silently_drop_unexpected(local_actor*,
const type_erased_tuple*);
/// Base class for actors running on this node, either /// Base class for actors running on this node, either
/// living in an own thread or cooperatively scheduled. /// living in an own thread or cooperatively scheduled.
class local_actor : public monitorable_actor, public resumable { class local_actor : public monitorable_actor, public resumable {
public: public:
// -- member types -----------------------------------------------------------
using mailbox_type = detail::single_reader_queue<mailbox_element, using mailbox_type = detail::single_reader_queue<mailbox_element,
detail::disposer>; detail::disposer>;
using unexpected_handler
= std::function<result<message>(local_actor* self,
const type_erased_tuple*)>;
// -- constructors, destructors, and assignment operators --------------------
~local_actor(); ~local_actor();
/**************************************************************************** // -- spawn functions --------------------------------------------------------
* spawn actors *
****************************************************************************/
template <class T, spawn_options Os = no_spawn_options, class... Ts> template <class T, spawn_options Os = no_spawn_options, class... Ts>
typename infer_handle_from_class<T>::type typename infer_handle_from_class<T>::type
...@@ -150,9 +184,7 @@ public: ...@@ -150,9 +184,7 @@ public:
return eval_opts(Os, system().spawn_in_groups_impl<make_unbound(Os)>(cfg, first, first + 1, fun, std::forward<Ts>(xs)...)); return eval_opts(Os, system().spawn_in_groups_impl<make_unbound(Os)>(cfg, first, first + 1, fun, std::forward<Ts>(xs)...));
} }
/**************************************************************************** // -- sending asynchronous messages ------------------------------------------
* send asynchronous messages *
****************************************************************************/
/// Sends `{xs...}` to `dest` using the priority `mp`. /// Sends `{xs...}` to `dest` using the priority `mp`.
template <class... Ts> template <class... Ts>
...@@ -281,12 +313,14 @@ public: ...@@ -281,12 +313,14 @@ public:
delayed_send(typed_actor<Sigs...>{dest}, rtime, std::forward<Ts>(xs)...); delayed_send(typed_actor<Sigs...>{dest}, rtime, std::forward<Ts>(xs)...);
} }
/**************************************************************************** // -- miscellaneous actor operations -----------------------------------------
* miscellaneous actor operations *
****************************************************************************/ /// Sets a custom handler for unexpected messages.
inline void set_unexpected_handler(unexpected_handler fun) {
unexpected_handler_ = std::move(fun);
}
/// Returns the execution unit currently used by this actor. /// Returns the execution unit currently used by this actor.
/// @warning not thread safe
inline execution_unit* context() const { inline execution_unit* context() const {
return context_; return context_;
} }
...@@ -444,17 +478,13 @@ public: ...@@ -444,17 +478,13 @@ public:
/// The default implementation throws a `std::logic_error`. /// The default implementation throws a `std::logic_error`.
virtual void load_state(deserializer& source, const unsigned int version); virtual void load_state(deserializer& source, const unsigned int version);
/**************************************************************************** // -- overridden member functions of resumable -------------------------------
* override pure virtual member functions of resumable *
****************************************************************************/
subtype_t subtype() const override; subtype_t subtype() const override;
resume_result resume(execution_unit*, size_t) override; resume_result resume(execution_unit*, size_t) override;
/**************************************************************************** // -- here be dragons: end of public interface -------------------------------
* here be dragons: end of public interface *
****************************************************************************/
/// @cond PRIVATE /// @cond PRIVATE
...@@ -523,7 +553,10 @@ public: ...@@ -523,7 +553,10 @@ public:
current_element_->mid = mp == message_priority::high current_element_->mid = mp == message_priority::high
? mid.with_high_priority() ? mid.with_high_priority()
: mid.with_normal_priority(); : mid.with_normal_priority();
current_element_->msg = make_message(std::forward<Ts>(xs)...); // make sure our current message is not
// destroyed before the end of the scope
auto next = make_message(std::forward<Ts>(xs)...);
next.swap(current_element_->msg);
dest->enqueue(std::move(current_element_), context()); dest->enqueue(std::move(current_element_), context());
} }
...@@ -709,6 +742,9 @@ protected: ...@@ -709,6 +742,9 @@ protected:
// used for group management // used for group management
std::set<group> subscriptions_; std::set<group> subscriptions_;
// used for setting custom functions for handling unexpected messages
unexpected_handler unexpected_handler_;
/// @endcond /// @endcond
private: private:
......
...@@ -46,7 +46,7 @@ public: ...@@ -46,7 +46,7 @@ public:
skip skip
}; };
match_case(bool has_wildcard, uint32_t token); match_case(uint32_t token);
match_case(match_case&&) = default; match_case(match_case&&) = default;
match_case(const match_case&) = default; match_case(const match_case&) = default;
...@@ -59,12 +59,7 @@ public: ...@@ -59,12 +59,7 @@ public:
return token_; return token_;
} }
inline bool has_wildcard() const {
return has_wildcard_;
}
private: private:
bool has_wildcard_;
uint32_t token_; uint32_t token_;
}; };
...@@ -139,16 +134,6 @@ private: ...@@ -139,16 +134,6 @@ private:
F& fun_; F& fun_;
}; };
template <class T>
struct projection_result {
using type = typename detail::get_callable_trait<T>::result_type;
};
template <>
struct projection_result<unit_t> {
using type = unit_t;
};
template <class F> template <class F>
class trivial_match_case : public match_case { class trivial_match_case : public match_case {
public: public:
...@@ -187,7 +172,7 @@ public: ...@@ -187,7 +172,7 @@ public:
trivial_match_case& operator=(const trivial_match_case&) = default; trivial_match_case& operator=(const trivial_match_case&) = default;
trivial_match_case(F f) trivial_match_case(F f)
: match_case(false, detail::make_type_token_from_list<pattern>()), : match_case(detail::make_type_token_from_list<pattern>()),
fun_(std::move(f)) { fun_(std::move(f)) {
// nop // nop
} }
...@@ -218,215 +203,7 @@ protected: ...@@ -218,215 +203,7 @@ protected:
F fun_; F fun_;
}; };
template <class F>
class catch_all_match_case : public match_case {
public:
using ctrait = typename detail::get_callable_trait<F>::type;
using plain_result_type = typename ctrait::result_type;
using result_type =
typename std::conditional<
std::is_reference<plain_result_type>::value,
plain_result_type,
typename std::remove_const<plain_result_type>::type
>::type;
using arg_types = typename ctrait::arg_types;
catch_all_match_case(F f)
: match_case(true, detail::make_type_token<>()),
fun_(std::move(f)) {
// nop
}
match_case::result invoke(detail::invoke_result_visitor& f, message& msg) override {
lfinvoker<std::is_same<result_type, void>::value, F> fun{fun_};
arg_types token;
auto fun_res = call_fun(fun, msg, token);
return f.visit(fun_res) ? match_case::match : match_case::skip;
}
protected:
template <class T>
static auto call_fun(T& f, message&, detail::type_list<>) -> decltype(f()) {
return f();
}
template <class T, class U>
static auto call_fun(T& f, message& x, detail::type_list<U>)
-> decltype(f(x)) {
static_assert(std::is_same<typename std::decay<U>::type, message>::value,
"catch-all match case callback must take either no "
"arguments or one argument of type `message`, "
"`const message&`, or `message&`");
return f(x);
}
F fun_;
};
template <class F, class Tuple>
class advanced_match_case : public match_case {
public:
using tuple_type = Tuple;
using base_type = advanced_match_case;
using result_type = typename detail::get_callable_trait<F>::result_type;
advanced_match_case(bool hw, uint32_t tt, F f)
: match_case(hw, tt),
fun_(std::move(f)) {
// nop
}
virtual bool prepare_invoke(message& msg, tuple_type*) = 0;
// this function could as well be implemented in `match_case_impl`;
// however, dealing with all the template parameters in a debugger
// is just dreadful; this "hack" essentially hides all the ugly
// template boilterplate types when debugging CAF applications
match_case::result invoke(detail::invoke_result_visitor& f, message& msg) override {
struct storage {
storage() : valid(false) {
// nop
}
~storage() {
if (valid) {
data.~tuple_type();
}
}
union { tuple_type data; };
bool valid;
};
storage st;
if (prepare_invoke(msg, &st.data)) {
st.valid = true;
lfinvoker<std::is_same<result_type, void>::value, F> fun{fun_};
auto fun_res = apply_args(fun, detail::get_indices(st.data), st.data);
return f.visit(fun_res) ? match_case::match : match_case::skip;
}
return match_case::no_match;
}
protected:
F fun_;
};
template <class Pattern>
struct pattern_has_wildcard {
static constexpr bool value =
detail::tl_index_of<
Pattern,
anything
>::value != -1;
};
template <class Projections>
struct projection_is_trivial {
static constexpr bool value =
detail::tl_count_not<
Projections,
detail::tbind<std::is_same, unit_t>::template type
>::value == 0;
};
/// @tparam F Function or function object denoting the callback.
/// @tparam Tuple Type of the storage for intermediate results during matching.
/// @tparam Pattern Input types for this match case.
template <class F, class Tuple, class Pattern, class Projections>
class advanced_match_case_impl : public advanced_match_case<F, Tuple> {
public:
using plain_result_type = typename detail::get_callable_trait<F>::result_type;
using result_type =
typename std::conditional<
std::is_reference<plain_result_type>::value,
plain_result_type,
typename std::remove_const<plain_result_type>::type
>::type;
static constexpr uint32_t static_type_token =
detail::make_type_token_from_list<Pattern>();
// Let F be "R (Ts...)" then match_case<F...> returns optional<R>
// unless R is void in which case bool is returned
using optional_result_type =
typename std::conditional<
std::is_same<result_type, void>::value,
optional<unit_t>,
optional<result_type>
>::type;
// Needed for static type checking when assigning to a typed behavior.
using arg_types = Pattern;
using fargs = typename detail::get_callable_trait<F>::arg_types;
static constexpr size_t fargs_size = detail::tl_size<fargs>::value;
using super = advanced_match_case<F, Tuple>;
advanced_match_case_impl(advanced_match_case_impl&&) = default;
advanced_match_case_impl(const advanced_match_case_impl&) = default;
advanced_match_case_impl(bool has_wcard, uint32_t ttoken, F f, Projections ps)
: super(has_wcard, ttoken, std::move(f)),
ps_(std::move(ps)) {
// nop
}
bool prepare_invoke(message& msg, Tuple* out) {
// detach msg before invoking fun_ if needed
if (detail::tl_exists<fargs, detail::is_mutable_ref>::value) {
msg.force_unshare();
}
using filtered_pattern =
typename detail::tl_filter_not_type<
Pattern,
anything
>::type;
using intermediate_tuple =
typename detail::tl_apply<
filtered_pattern,
detail::pseudo_tuple
>::type;
intermediate_tuple it;
detail::meta_elements<Pattern> ms;
// check if try_match() reports success
if (! detail::try_match(msg, ms.arr.data(), ms.arr.size(), it.data)) {
return false;
}
match_case_zipper zip;
using indices_type = typename detail::il_indices<intermediate_tuple>::type;
//indices_type indices;
typename detail::il_take<indices_type, std::tuple_size<Projections>::value - fargs_size>::type lefts;
typename detail::il_right<indices_type, fargs_size>::type rights;
has_none hn;
// check if guards of discarded arguments are fulfilled
auto lhs_tup = tuple_zip(zip, lefts, ps_, it);
if (detail::apply_args(hn, detail::get_indices(lhs_tup), lhs_tup)) {
return false;
}
// zip remaining arguments into output tuple
new (out) Tuple (tuple_zip(zip, rights, ps_, it));
//tuple_type rhs_tup = tuple_zip(zip, rights, ps_, it);
// check if remaining guards are fulfilled
if (detail::apply_args(hn, detail::get_indices(*out), *out)) {
out->~Tuple();
return false;
}
return true;
}
private:
Projections ps_;
};
struct match_case_info { struct match_case_info {
bool has_wildcard;
uint32_t type_token; uint32_t type_token;
match_case* ptr; match_case* ptr;
}; };
......
...@@ -294,7 +294,8 @@ public: ...@@ -294,7 +294,8 @@ public:
if (valid()) if (valid())
return {}; return {};
if (has_error_context()) if (has_error_context())
return {error_code(), extended_error_->first, extended_error_->second}; return {error_code(), extended_error_->first,
make_message(extended_error_->second)};
return {error_code(), error_category_}; return {error_code(), error_category_};
} }
...@@ -371,7 +372,8 @@ private: ...@@ -371,7 +372,8 @@ private:
void cr_error(const error_type& x) { void cr_error(const error_type& x) {
flag_ = x.compress_code_and_size(); flag_ = x.compress_code_and_size();
if (has_error_context()) if (has_error_context())
extended_error_ = new extended_error(x.category(), x.context()); extended_error_ = new extended_error(x.category(),
to_string(x.context()));
else else
error_category_ = x.category(); error_category_ = x.category();
} }
...@@ -380,7 +382,7 @@ private: ...@@ -380,7 +382,7 @@ private:
flag_ = x.compress_code_and_size(); flag_ = x.compress_code_and_size();
if (has_error_context()) if (has_error_context())
extended_error_ = new extended_error(x.category(), extended_error_ = new extended_error(x.category(),
std::move(x.context())); to_string(x.context()));
else else
error_category_ = x.category(); error_category_ = x.category();
} }
......
...@@ -267,6 +267,9 @@ public: ...@@ -267,6 +267,9 @@ public:
message& operator+=(const message& x); message& operator+=(const message& x);
/// Creates a message object from `ptr`.
static message from(const type_erased_tuple* ptr);
/// @cond PRIVATE /// @cond PRIVATE
using raw_ptr = detail::message_data*; using raw_ptr = detail::message_data*;
......
...@@ -33,6 +33,8 @@ namespace caf { ...@@ -33,6 +33,8 @@ namespace caf {
/// from a series of values using the member function `append`. /// from a series of values using the member function `append`.
class message_builder { class message_builder {
public: public:
friend class message;
message_builder(const message_builder&) = delete; message_builder(const message_builder&) = delete;
message_builder& operator=(const message_builder&) = delete; message_builder& operator=(const message_builder&) = delete;
......
...@@ -30,7 +30,6 @@ ...@@ -30,7 +30,6 @@
#include "caf/none.hpp" #include "caf/none.hpp"
#include "caf/intrusive_ptr.hpp" #include "caf/intrusive_ptr.hpp"
#include "caf/on.hpp"
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/duration.hpp" #include "caf/duration.hpp"
#include "caf/behavior.hpp" #include "caf/behavior.hpp"
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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. *
******************************************************************************/
#ifndef CAF_ON_HPP
#define CAF_ON_HPP
#include <chrono>
#include <memory>
#include <functional>
#include <type_traits>
#include "caf/unit.hpp"
#include "caf/atom.hpp"
#include "caf/anything.hpp"
#include "caf/message.hpp"
#include "caf/duration.hpp"
#include "caf/skip_message.hpp"
#include "caf/may_have_timeout.hpp"
#include "caf/timeout_definition.hpp"
#include "caf/detail/boxed.hpp"
#include "caf/detail/unboxed.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/arg_match_t.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/detail/match_case_builder.hpp"
#include "caf/detail/tail_argument_token.hpp"
#include "caf/detail/implicit_conversions.hpp"
namespace caf {
namespace detail {
template <class T, bool IsFun = detail::is_callable<T>::value
&& ! detail::is_boxed<T>::value>
struct pattern_type {
using type =
typename implicit_conversions<
typename std::decay<
typename detail::unboxed<T>::type
>::type
>::type;
};
template <class T>
struct pattern_type<T, true> {
using ctrait = detail::get_callable_trait<T>;
using args = typename ctrait::arg_types;
static_assert(detail::tl_size<args>::value == 1,
"only unary functions are allowed as projections");
using type =
typename std::decay<
typename detail::tl_head<args>::type
>::type;
};
} // namespace detail
} // namespace caf
namespace caf {
/// A wildcard that matches any number of any values.
constexpr auto any_vals = anything();
/// A wildcard that matches any value of type `T`.
template <class T>
constexpr typename detail::boxed<T>::type val() {
return typename detail::boxed<T>::type();
}
/// A wildcard that matches the argument types
/// of a given callback, must be the last argument to `on()`.
constexpr auto arg_match = detail::boxed<detail::arg_match_t>::type();
/// Generates function objects from a binary predicate and a value.
template <class T, typename BinaryPredicate>
std::function<optional<T>(const T&)> guarded(BinaryPredicate p, T value) {
return [=](const T& other) -> optional<T> {
if (p(other, value)) {
return value;
}
return none;
};
}
// special case covering arg_match as argument to guarded()
template <class T, typename Predicate>
unit_t guarded(Predicate, const detail::wrapped<T>&) {
return unit;
}
inline unit_t to_guard(const anything&) {
return unit;
}
template <class T>
unit_t to_guard(detail::wrapped<T> (*)()) {
return unit;
}
template <class T>
unit_t to_guard(const detail::wrapped<T>&) {
return unit;
}
template <class T>
std::function<optional<typename detail::strip_and_convert<T>::type>(
const typename detail::strip_and_convert<T>::type&)>
to_guard(const T& value,
typename std::enable_if<! detail::is_callable<T>::value>::type* = 0) {
using type = typename detail::strip_and_convert<T>::type;
return guarded<type>(std::equal_to<type>{}, value);
}
template <class F>
F to_guard(F fun,
typename std::enable_if<detail::is_callable<F>::value>::type* = 0) {
return fun;
}
template <atom_value V>
auto to_guard(const atom_constant<V>&) -> decltype(to_guard(V)) {
return to_guard(V);
}
/// Returns a generator for `match_case` objects.
template <class... Ts>
auto on(const Ts&... xs)
-> detail::advanced_match_case_builder<
detail::type_list<
decltype(to_guard(xs))...
>,
detail::type_list<
typename detail::pattern_type<typename std::decay<Ts>::type>::type...>
> {
return {detail::variadic_ctor{}, to_guard(xs)...};
}
/// Returns a generator for `match_case` objects.
template <class T, class... Ts>
decltype(on(val<T>(), val<Ts>()...)) on() {
return on(val<T>(), val<Ts>()...);
}
/// Returns a generator for `match_case` objects.
template <atom_value A0, class... Ts>
decltype(on(A0, val<Ts>()...)) on() {
return on(A0, val<Ts>()...);
}
/// Returns a generator for `match_case` objects.
template <atom_value A0, atom_value A1, class... Ts>
decltype(on(A0, A1, val<Ts>()...)) on() {
return on(A0, A1, val<Ts>()...);
}
/// Returns a generator for `match_case` objects.
template <atom_value A0, atom_value A1, atom_value A2, class... Ts>
decltype(on(A0, A1, A2, val<Ts>()...)) on() {
return on(A0, A1, A2, val<Ts>()...);
}
/// Returns a generator for `match_case` objects.
template <atom_value A0, atom_value A1, atom_value A2, atom_value A3,
class... Ts>
decltype(on(A0, A1, A2, A3, val<Ts>()...)) on() {
return on(A0, A1, A2, A3, val<Ts>()...);
}
/// Returns a generator for timeouts.
template <class Rep, class Period>
constexpr detail::timeout_definition_builder
after(const std::chrono::duration<Rep, Period>& d) {
return {duration(d)};
}
/// Generates catch-all `match_case` objects.
constexpr auto others = detail::catch_all_match_case_builder();
/// Semantically equal to `on(arg_match)`, but uses a (faster)
/// special-purpose `match_case` implementation.
constexpr auto on_arg_match = detail::trivial_match_case_builder();
} // namespace caf
#endif // CAF_ON_HPP
...@@ -75,20 +75,6 @@ public: ...@@ -75,20 +75,6 @@ public:
await_impl(f, e); await_impl(f, e);
} }
template <class F, class E = detail::is_callable_t<F>>
void generic_await(F f) {
behavior tmp{others >> f};
self_->set_awaited_response_handler(mid_, std::move(tmp));
}
template <class F, class OnError,
class E1 = detail::is_callable_t<F>,
class E2 = detail::is_handler_for_ef<OnError, error>>
void generic_await(F f, OnError ef) {
behavior tmp{ef, others >> f};
self_->set_awaited_response_handler(mid_, std::move(tmp));
}
template <class F, class E = detail::is_callable_t<F>> template <class F, class E = detail::is_callable_t<F>>
void then(F f) const { void then(F f) const {
then_impl(f); then_impl(f);
...@@ -101,20 +87,6 @@ public: ...@@ -101,20 +87,6 @@ public:
then_impl(f, e); then_impl(f, e);
} }
template <class F, class E = detail::is_callable_t<F>>
void generic_then(F f) {
behavior tmp{others >> f};
self_->set_multiplexed_response_handler(mid_, std::move(tmp));
}
template <class F, class OnError,
class E1 = detail::is_callable_t<F>,
class E2 = detail::is_handler_for_ef<OnError, error>>
void generic_then(F f, OnError ef) {
behavior tmp{ef, others >> f};
self_->set_multiplexed_response_handler(mid_, std::move(tmp));
}
private: private:
template <class F> template <class F>
void await_impl(F& f) const { void await_impl(F& f) const {
...@@ -125,7 +97,7 @@ private: ...@@ -125,7 +97,7 @@ private:
"response handlers are not allowed to have a return " "response handlers are not allowed to have a return "
"type other than void"); "type other than void");
detail::type_checker<Output, F>::check(); detail::type_checker<Output, F>::check();
self_->set_awaited_response_handler(mid_, behavior{std::move(f)}); self_->set_awaited_response_handler(mid_, message_handler{std::move(f)});
} }
template <class F, class OnError> template <class F, class OnError>
...@@ -137,13 +109,8 @@ private: ...@@ -137,13 +109,8 @@ private:
"response handlers are not allowed to have a return " "response handlers are not allowed to have a return "
"type other than void"); "type other than void");
detail::type_checker<Output, F>::check(); detail::type_checker<Output, F>::check();
auto fallback = others >> [=] {
auto err = make_error(sec::unexpected_response);
ef(err);
};
self_->set_awaited_response_handler(mid_, behavior{std::move(f), self_->set_awaited_response_handler(mid_, behavior{std::move(f),
std::move(ef), std::move(ef)});
std::move(fallback)});
} }
template <class F> template <class F>
...@@ -155,8 +122,7 @@ private: ...@@ -155,8 +122,7 @@ private:
"response handlers are not allowed to have a return " "response handlers are not allowed to have a return "
"type other than void"); "type other than void");
detail::type_checker<Output, F>::check(); detail::type_checker<Output, F>::check();
self_->set_multiplexed_response_handler(mid_, self_->set_multiplexed_response_handler(mid_, behavior{std::move(f)});
behavior{std::move(f)});
} }
template <class F, class OnError> template <class F, class OnError>
...@@ -168,14 +134,8 @@ private: ...@@ -168,14 +134,8 @@ private:
"response handlers are not allowed to have a return " "response handlers are not allowed to have a return "
"type other than void"); "type other than void");
detail::type_checker<Output, F>::check(); detail::type_checker<Output, F>::check();
auto fallback = others >> [=] { self_->set_multiplexed_response_handler(mid_, behavior{std::move(f),
auto err = make_error(sec::unexpected_response); std::move(ef)});
ef(err);
};
self_->set_multiplexed_response_handler(mid_,
behavior{std::move(f),
std::move(ef),
std::move(fallback)});
} }
message_id mid_; message_id mid_;
...@@ -233,11 +193,7 @@ private: ...@@ -233,11 +193,7 @@ private:
"response handlers are not allowed to have a return " "response handlers are not allowed to have a return "
"type other than void"); "type other than void");
detail::type_checker<Output, F>::check(); detail::type_checker<Output, F>::check();
auto fallback = others >> [=] { behavior tmp{std::move(f), std::move(ef)};
auto err = make_error(sec::unexpected_response);
ef(err);
};
behavior tmp{std::move(f), std::move(ef), std::move(fallback)};
self_->dequeue(tmp, mid_); self_->dequeue(tmp, mid_);
} }
......
...@@ -73,6 +73,9 @@ const char* to_string(sec); ...@@ -73,6 +73,9 @@ const char* to_string(sec);
/// @relates sec /// @relates sec
error make_error(sec); error make_error(sec);
/// @relates sec
error make_error(sec, message context);
} // namespace caf } // namespace caf
#endif // CAF_SEC_HPP #endif // CAF_SEC_HPP
...@@ -74,6 +74,9 @@ public: ...@@ -74,6 +74,9 @@ public:
/// Returns a string representation of the element at position `pos`. /// Returns a string representation of the element at position `pos`.
virtual std::string stringify(size_t pos) const = 0; virtual std::string stringify(size_t pos) const = 0;
/// Returns a copy of the element at position `pos`.
virtual type_erased_value_ptr copy(size_t pos) const = 0;
/// Saves the element at position `pos` to `sink`. /// Saves the element at position `pos` to `sink`.
virtual void save(size_t pos, serializer& sink) const = 0; virtual void save(size_t pos, serializer& sink) const = 0;
...@@ -171,6 +174,10 @@ public: ...@@ -171,6 +174,10 @@ public:
return ptrs_[pos]->stringify(); return ptrs_[pos]->stringify();
} }
type_erased_value_ptr copy(size_t pos) const override {
return ptrs_[pos]->copy();
}
void save(size_t pos, serializer& sink) const override { void save(size_t pos, serializer& sink) const override {
return ptrs_[pos]->save(sink); return ptrs_[pos]->save(sink);
} }
......
...@@ -65,7 +65,7 @@ public: ...@@ -65,7 +65,7 @@ public:
using portable_names = std::unordered_map<std::type_index, std::string>; using portable_names = std::unordered_map<std::type_index, std::string>;
using error_renderer = std::function<std::string (uint8_t, const std::string&)>; using error_renderer = std::function<std::string (uint8_t)>;
using error_renderers = std::unordered_map<atom_value, error_renderer>; using error_renderers = std::unordered_map<atom_value, error_renderer>;
......
...@@ -27,9 +27,7 @@ ...@@ -27,9 +27,7 @@
#include <iostream> #include <iostream>
#include <condition_variable> #include <condition_variable>
#include "caf/on.hpp"
#include "caf/send.hpp" #include "caf/send.hpp"
#include "caf/anything.hpp"
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/scoped_actor.hpp" #include "caf/scoped_actor.hpp"
...@@ -113,9 +111,6 @@ public: ...@@ -113,9 +111,6 @@ public:
}, },
[&](const exit_msg&) { [&](const exit_msg&) {
received_exit = true; received_exit = true;
},
others >> [&] {
CAF_LOG_ERROR("unexpected:" << CAF_ARG(msg_ptr->msg));
} }
}; };
// loop // loop
...@@ -305,9 +300,6 @@ void printer_loop(blocking_actor* self) { ...@@ -305,9 +300,6 @@ void printer_loop(blocking_actor* self) {
auto d = get_data(src, true); auto d = get_data(src, true);
if (d) if (d)
d->redirect = get_sink_handle(self->system(), fcache, fn, flag); d->redirect = get_sink_handle(self->system(), fcache, fn, flag);
},
others >> [&] {
std::cerr << "*** unexpected message in printer loop" << std::endl;
} }
); );
} }
......
...@@ -251,11 +251,6 @@ void actor_registry::start() { ...@@ -251,11 +251,6 @@ void actor_registry::start() {
[=](const down_msg& dm) { [=](const down_msg& dm) {
CAF_LOG_TRACE(CAF_ARG(dm)); CAF_LOG_TRACE(CAF_ARG(dm));
unsubscribe_all(actor_cast<actor>(dm.source)); unsubscribe_all(actor_cast<actor>(dm.source));
},
others >> [=](const message& msg) -> error {
CAF_IGNORE_UNUSED(msg);
CAF_LOG_WARNING("unexpected:" << CAF_ARG(msg));
return sec::unexpected_message;
} }
}; };
}; };
...@@ -270,10 +265,6 @@ void actor_registry::start() { ...@@ -270,10 +265,6 @@ void actor_registry::start() {
if (! res.first) if (! res.first)
return sec::cannot_spawn_actor_from_arguments; return sec::cannot_spawn_actor_from_arguments;
return {ok_atom::value, res.first, res.second}; return {ok_atom::value, res.first, res.second};
},
others >> [=](const message& msg) {
CAF_IGNORE_UNUSED(msg);
CAF_LOG_WARNING("unexpected:" << CAF_ARG(msg));
} }
}; };
}; };
......
...@@ -174,14 +174,15 @@ const uniform_type_info_map& actor_system::types() const { ...@@ -174,14 +174,15 @@ const uniform_type_info_map& actor_system::types() const {
std::string actor_system::render(const error& x) const { std::string actor_system::render(const error& x) const {
auto f = types().renderer(x.category()); auto f = types().renderer(x.category());
if (f) std::string result = "error(";
return f(x.code(), x.context()); result += to_string(x.category());
std::string result = "unregistered error category "; result += ", ";
result += deep_to_string(x.category()); result += f ? f(x.code()) : std::to_string(static_cast<int>(x.code()));
result += ", error code "; if (! x.context().empty()) {
result += std::to_string(static_cast<int>(x.code())); result += ", ";
result += ", context: "; result += to_string(x.context());
result += x.context(); }
result += ")";
return result; return result;
} }
......
...@@ -243,6 +243,10 @@ actor_system_config::actor_system_config() ...@@ -243,6 +243,10 @@ actor_system_config::actor_system_config()
opt_group(options_, "opencl") opt_group(options_, "opencl")
.add(opencl_device_ids, "device-ids", .add(opencl_device_ids, "device-ids",
"restricts which OpenCL devices are accessed by CAF"); "restricts which OpenCL devices are accessed by CAF");
// add renderers for default error categories
error_renderers_.emplace(atom("system"), [](uint8_t x) -> std::string {
return to_string(static_cast<sec>(x));
});
} }
actor_system_config::actor_system_config(int argc, char** argv) actor_system_config::actor_system_config(int argc, char** argv)
......
...@@ -28,8 +28,9 @@ namespace { ...@@ -28,8 +28,9 @@ namespace {
class combinator final : public behavior_impl { class combinator final : public behavior_impl {
public: public:
bool invoke(detail::invoke_result_visitor& f, message& arg) override { match_case::result invoke(detail::invoke_result_visitor& f, message& arg) override {
return first->invoke(f, arg) || second->invoke(f, arg); auto x = first->invoke(f, arg);
return x == match_case::no_match ? second->invoke(f, arg) : x;
} }
void handle_timeout() override { void handle_timeout() override {
...@@ -88,19 +89,20 @@ behavior_impl::behavior_impl(duration tout) ...@@ -88,19 +89,20 @@ behavior_impl::behavior_impl(duration tout)
// nop // nop
} }
bool behavior_impl::invoke(detail::invoke_result_visitor& f, message& msg) { match_case::result behavior_impl::invoke(detail::invoke_result_visitor& f,
message& msg) {
auto msg_token = msg.type_token(); auto msg_token = msg.type_token();
for (auto i = begin_; i != end_; ++i) for (auto i = begin_; i != end_; ++i)
if (i->has_wildcard || i->type_token == msg_token) if (i->type_token == msg_token)
switch (i->ptr->invoke(f, msg)) { switch (i->ptr->invoke(f, msg)) {
case match_case::no_match:
break;
case match_case::match: case match_case::match:
return true; return match_case::match;
case match_case::skip: case match_case::skip:
return false; return match_case::skip;
case match_case::no_match:
static_cast<void>(0); // nop
}; };
return false; return match_case::no_match;
} }
optional<message> behavior_impl::invoke(message& x) { optional<message> behavior_impl::invoke(message& x) {
......
...@@ -28,6 +28,7 @@ namespace caf { ...@@ -28,6 +28,7 @@ namespace caf {
blocking_actor::blocking_actor(actor_config& sys) : super(sys) { blocking_actor::blocking_actor(actor_config& sys) : super(sys) {
is_blocking(true); is_blocking(true);
set_unexpected_handler(skip_unexpected);
} }
blocking_actor::~blocking_actor() { blocking_actor::~blocking_actor() {
......
...@@ -27,10 +27,36 @@ ...@@ -27,10 +27,36 @@
namespace caf { namespace caf {
namespace detail { namespace detail {
concatenated_tuple::concatenated_tuple(std::initializer_list<cow_ptr> xs) {
for (auto& x : xs) {
if (x) {
auto dptr = dynamic_cast<const concatenated_tuple*>(x.get());
if (dptr) {
auto& vec = dptr->data_;
data_.insert(data_.end(), vec.begin(), vec.end());
} else {
data_.push_back(std::move(x));
}
}
}
type_token_ = make_type_token();
for (const auto& m : data_)
for (size_t i = 0; i < m->size(); ++i)
type_token_ = add_to_type_token(type_token_, m->type_nr(i));
auto acc_size = [](size_t tmp, const cow_ptr& val) {
return tmp + val->size();
};
size_ = std::accumulate(data_.begin(), data_.end(), size_t{0}, acc_size);
}
auto concatenated_tuple::make(std::initializer_list<cow_ptr> xs) -> cow_ptr { auto concatenated_tuple::make(std::initializer_list<cow_ptr> xs) -> cow_ptr {
return cow_ptr{make_counted<concatenated_tuple>(xs)}; return cow_ptr{make_counted<concatenated_tuple>(xs)};
} }
message_data::cow_ptr concatenated_tuple::copy() const {
return cow_ptr(new concatenated_tuple(*this), false);
}
void* concatenated_tuple::get_mutable(size_t pos) { void* concatenated_tuple::get_mutable(size_t pos) {
CAF_ASSERT(pos < size()); CAF_ASSERT(pos < size());
auto selected = select(pos); auto selected = select(pos);
...@@ -47,16 +73,6 @@ size_t concatenated_tuple::size() const { ...@@ -47,16 +73,6 @@ size_t concatenated_tuple::size() const {
return size_; return size_;
} }
message_data::cow_ptr concatenated_tuple::copy() const {
return cow_ptr(new concatenated_tuple(*this), false);
}
const void* concatenated_tuple::get(size_t pos) const {
CAF_ASSERT(pos < size());
auto selected = select(pos);
return selected.first->get(selected.second);
}
uint32_t concatenated_tuple::type_token() const { uint32_t concatenated_tuple::type_token() const {
return type_token_; return type_token_;
} }
...@@ -67,10 +83,10 @@ message_data::rtti_pair concatenated_tuple::type(size_t pos) const { ...@@ -67,10 +83,10 @@ message_data::rtti_pair concatenated_tuple::type(size_t pos) const {
return selected.first->type(selected.second); return selected.first->type(selected.second);
} }
void concatenated_tuple::save(size_t pos, serializer& sink) const { const void* concatenated_tuple::get(size_t pos) const {
CAF_ASSERT(pos < size()); CAF_ASSERT(pos < size());
auto selected = select(pos); auto selected = select(pos);
selected.first->save(selected.second, sink); return selected.first->get(selected.second);
} }
std::string concatenated_tuple::stringify(size_t pos) const { std::string concatenated_tuple::stringify(size_t pos) const {
...@@ -79,6 +95,18 @@ std::string concatenated_tuple::stringify(size_t pos) const { ...@@ -79,6 +95,18 @@ std::string concatenated_tuple::stringify(size_t pos) const {
return selected.first->stringify(selected.second); return selected.first->stringify(selected.second);
} }
type_erased_value_ptr concatenated_tuple::copy(size_t pos) const {
CAF_ASSERT(pos < size());
auto selected = select(pos);
return selected.first->copy(selected.second);
}
void concatenated_tuple::save(size_t pos, serializer& sink) const {
CAF_ASSERT(pos < size());
auto selected = select(pos);
selected.first->save(selected.second, sink);
}
std::pair<message_data*, size_t> concatenated_tuple::select(size_t pos) const { std::pair<message_data*, size_t> concatenated_tuple::select(size_t pos) const {
auto idx = pos; auto idx = pos;
for (const auto& m : data_) { for (const auto& m : data_) {
...@@ -91,27 +119,5 @@ std::pair<message_data*, size_t> concatenated_tuple::select(size_t pos) const { ...@@ -91,27 +119,5 @@ std::pair<message_data*, size_t> concatenated_tuple::select(size_t pos) const {
throw std::out_of_range("out of range: concatenated_tuple::select"); throw std::out_of_range("out of range: concatenated_tuple::select");
} }
concatenated_tuple::concatenated_tuple(std::initializer_list<cow_ptr> xs) {
for (auto& x : xs) {
if (x) {
auto dptr = dynamic_cast<const concatenated_tuple*>(x.get());
if (dptr) {
auto& vec = dptr->data_;
data_.insert(data_.end(), vec.begin(), vec.end());
} else {
data_.push_back(std::move(x));
}
}
}
type_token_ = make_type_token();
for (const auto& m : data_)
for (size_t i = 0; i < m->size(); ++i)
type_token_ = add_to_type_token(type_token_, m->type_nr(i));
auto acc_size = [](size_t tmp, const cow_ptr& val) {
return tmp + val->size();
};
size_ = std::accumulate(data_.begin(), data_.end(), size_t{0}, acc_size);
}
} // namespace detail } // namespace detail
} // namespace caf } // namespace caf
...@@ -24,6 +24,20 @@ ...@@ -24,6 +24,20 @@
namespace caf { namespace caf {
namespace detail { namespace detail {
decorated_tuple::decorated_tuple(cow_ptr&& d, vector_type&& v)
: decorated_(std::move(d)),
mapping_(std::move(v)),
type_token_(0xFFFFFFFF) {
CAF_ASSERT(mapping_.empty()
|| *(std::max_element(mapping_.begin(), mapping_.end()))
< static_cast<const cow_ptr&>(decorated_)->size());
// calculate type token
for (size_t i = 0; i < mapping_.size(); ++i) {
type_token_ <<= 6;
type_token_ |= decorated_->type_nr(mapping_[i]);
}
}
decorated_tuple::cow_ptr decorated_tuple::make(cow_ptr d, vector_type v) { decorated_tuple::cow_ptr decorated_tuple::make(cow_ptr d, vector_type v) {
auto ptr = dynamic_cast<const decorated_tuple*>(d.get()); auto ptr = dynamic_cast<const decorated_tuple*>(d.get());
if (ptr) { if (ptr) {
...@@ -36,6 +50,10 @@ decorated_tuple::cow_ptr decorated_tuple::make(cow_ptr d, vector_type v) { ...@@ -36,6 +50,10 @@ decorated_tuple::cow_ptr decorated_tuple::make(cow_ptr d, vector_type v) {
return make_counted<decorated_tuple>(std::move(d), std::move(v)); return make_counted<decorated_tuple>(std::move(d), std::move(v));
} }
message_data::cow_ptr decorated_tuple::copy() const {
return cow_ptr(new decorated_tuple(*this), false);
}
void* decorated_tuple::get_mutable(size_t pos) { void* decorated_tuple::get_mutable(size_t pos) {
CAF_ASSERT(pos < size()); CAF_ASSERT(pos < size());
return decorated_->get_mutable(mapping_[pos]); return decorated_->get_mutable(mapping_[pos]);
...@@ -50,15 +68,6 @@ size_t decorated_tuple::size() const { ...@@ -50,15 +68,6 @@ size_t decorated_tuple::size() const {
return mapping_.size(); return mapping_.size();
} }
message_data::cow_ptr decorated_tuple::copy() const {
return cow_ptr(new decorated_tuple(*this), false);
}
const void* decorated_tuple::get(size_t pos) const {
CAF_ASSERT(pos < size());
return decorated_->get(mapping_[pos]);
}
uint32_t decorated_tuple::type_token() const { uint32_t decorated_tuple::type_token() const {
return type_token_; return type_token_;
} }
...@@ -67,9 +76,9 @@ message_data::rtti_pair decorated_tuple::type(size_t pos) const { ...@@ -67,9 +76,9 @@ message_data::rtti_pair decorated_tuple::type(size_t pos) const {
return decorated_->type(mapping_[pos]); return decorated_->type(mapping_[pos]);
} }
void decorated_tuple::save(size_t pos, serializer& sink) const { const void* decorated_tuple::get(size_t pos) const {
CAF_ASSERT(pos < size()); CAF_ASSERT(pos < size());
return decorated_->save(mapping_[pos], sink); return decorated_->get(mapping_[pos]);
} }
std::string decorated_tuple::stringify(size_t pos) const { std::string decorated_tuple::stringify(size_t pos) const {
...@@ -77,18 +86,14 @@ std::string decorated_tuple::stringify(size_t pos) const { ...@@ -77,18 +86,14 @@ std::string decorated_tuple::stringify(size_t pos) const {
return decorated_->stringify(mapping_[pos]); return decorated_->stringify(mapping_[pos]);
} }
decorated_tuple::decorated_tuple(cow_ptr&& d, vector_type&& v) type_erased_value_ptr decorated_tuple::copy(size_t pos) const {
: decorated_(std::move(d)), CAF_ASSERT(pos < size());
mapping_(std::move(v)), return decorated_->copy(mapping_[pos]);
type_token_(0xFFFFFFFF) { }
CAF_ASSERT(mapping_.empty()
|| *(std::max_element(mapping_.begin(), mapping_.end())) void decorated_tuple::save(size_t pos, serializer& sink) const {
< static_cast<const cow_ptr&>(decorated_)->size()); CAF_ASSERT(pos < size());
// calculate type token return decorated_->save(mapping_[pos], sink);
for (size_t i = 0; i < mapping_.size(); ++i) {
type_token_ <<= 6;
type_token_ |= decorated_->type_nr(mapping_[i]);
}
} }
} // namespace detail } // namespace detail
......
...@@ -28,6 +28,13 @@ dynamic_message_data::dynamic_message_data() : type_token_(0xFFFFFFFF) { ...@@ -28,6 +28,13 @@ dynamic_message_data::dynamic_message_data() : type_token_(0xFFFFFFFF) {
// nop // nop
} }
dynamic_message_data::dynamic_message_data(elements&& data)
: elements_(std::move(data)),
type_token_(0xFFFFFFFF) {
for (auto& e : elements_)
add_to_type_token(e->type().first);
}
dynamic_message_data::dynamic_message_data(const dynamic_message_data& other) dynamic_message_data::dynamic_message_data(const dynamic_message_data& other)
: detail::message_data(other), : detail::message_data(other),
type_token_(0xFFFFFFFF) { type_token_(0xFFFFFFFF) {
...@@ -37,53 +44,61 @@ dynamic_message_data::dynamic_message_data(const dynamic_message_data& other) ...@@ -37,53 +44,61 @@ dynamic_message_data::dynamic_message_data(const dynamic_message_data& other)
} }
} }
dynamic_message_data::dynamic_message_data(elements&& data)
: elements_(std::move(data)),
type_token_(0xFFFFFFFF) {
for (auto& e : elements_)
add_to_type_token(e->type().first);
}
dynamic_message_data::~dynamic_message_data() { dynamic_message_data::~dynamic_message_data() {
// nop // nop
} }
const void* dynamic_message_data::get(size_t pos) const { message_data::cow_ptr dynamic_message_data::copy() const {
CAF_ASSERT(pos < size()); return make_counted<dynamic_message_data>(*this);
return elements_[pos]->get();
} }
void dynamic_message_data::save(size_t pos, serializer& sink) const { void* dynamic_message_data::get_mutable(size_t pos) {
elements_[pos]->save(sink); CAF_ASSERT(pos < size());
return elements_[pos]->get_mutable();
} }
void dynamic_message_data::load(size_t pos, deserializer& source) { void dynamic_message_data::load(size_t pos, deserializer& source) {
elements_[pos]->load(source);
}
void* dynamic_message_data::get_mutable(size_t pos) {
CAF_ASSERT(pos < size()); CAF_ASSERT(pos < size());
return elements_[pos]->get_mutable(); elements_[pos]->load(source);
} }
size_t dynamic_message_data::size() const { size_t dynamic_message_data::size() const {
return elements_.size(); return elements_.size();
} }
message_data::cow_ptr dynamic_message_data::copy() const { uint32_t dynamic_message_data::type_token() const {
return make_counted<dynamic_message_data>(*this); return type_token_;
} }
auto dynamic_message_data::type(size_t pos) const -> rtti_pair { auto dynamic_message_data::type(size_t pos) const -> rtti_pair {
CAF_ASSERT(pos < size());
return elements_[pos]->type(); return elements_[pos]->type();
} }
const void* dynamic_message_data::get(size_t pos) const {
CAF_ASSERT(pos < size());
CAF_ASSERT(pos < size());
return elements_[pos]->get();
}
std::string dynamic_message_data::stringify(size_t pos) const { std::string dynamic_message_data::stringify(size_t pos) const {
CAF_ASSERT(pos < size());
return elements_[pos]->stringify(); return elements_[pos]->stringify();
} }
uint32_t dynamic_message_data::type_token() const { type_erased_value_ptr dynamic_message_data::copy(size_t pos) const {
return type_token_; CAF_ASSERT(pos < size());
return elements_[pos]->copy();
}
void dynamic_message_data::save(size_t pos, serializer& sink) const {
CAF_ASSERT(pos < size());
elements_[pos]->save(sink);
}
void dynamic_message_data::clear() {
elements_.clear();
type_token_ = 0xFFFFFFFF;
} }
void dynamic_message_data::append(type_erased_value_ptr x) { void dynamic_message_data::append(type_erased_value_ptr x) {
...@@ -95,10 +110,5 @@ void dynamic_message_data::add_to_type_token(uint16_t typenr) { ...@@ -95,10 +110,5 @@ void dynamic_message_data::add_to_type_token(uint16_t typenr) {
type_token_ = (type_token_ << 6) | typenr; type_token_ = (type_token_ << 6) | typenr;
} }
void dynamic_message_data::clear() {
elements_.clear();
type_token_ = 0xFFFFFFFF;
}
} // namespace detail } // namespace detail
} // namespace caf } // namespace caf
...@@ -29,7 +29,7 @@ error::error() : code_(0), category_(atom("")) { ...@@ -29,7 +29,7 @@ error::error() : code_(0), category_(atom("")) {
// nop // nop
} }
error::error(uint8_t x, atom_value y, std::string z) error::error(uint8_t x, atom_value y, message z)
: code_(x), : code_(x),
category_(y), category_(y),
context_(std::move(z)) { context_(std::move(z)) {
...@@ -44,11 +44,11 @@ atom_value error::category() const { ...@@ -44,11 +44,11 @@ atom_value error::category() const {
return category_; return category_;
} }
std::string& error::context() { message& error::context() {
return context_; return context_;
} }
const std::string& error::context() const { const message& error::context() const {
return context_; return context_;
} }
...@@ -68,9 +68,6 @@ uint32_t error::compress_code_and_size() const { ...@@ -68,9 +68,6 @@ uint32_t error::compress_code_and_size() const {
} }
void serialize(serializer& sink, error& x, const unsigned int) { void serialize(serializer& sink, error& x, const unsigned int) {
// truncat oversized strings
if (x.context_.size() > 0xFFFFFF)
x.context_.resize(0xFFFFFF);
auto flag_size_and_code = x.compress_code_and_size(); auto flag_size_and_code = x.compress_code_and_size();
if (x.category_ == atom("") || x.code_ == 0) { if (x.category_ == atom("") || x.code_ == 0) {
flag_size_and_code |= 0x80000000; flag_size_and_code |= 0x80000000;
...@@ -80,11 +77,12 @@ void serialize(serializer& sink, error& x, const unsigned int) { ...@@ -80,11 +77,12 @@ void serialize(serializer& sink, error& x, const unsigned int) {
sink << flag_size_and_code; sink << flag_size_and_code;
sink << x.category_; sink << x.category_;
if (! x.context_.empty()) if (! x.context_.empty())
sink.apply_raw(x.context_.size(), const_cast<char*>(x.context_.data())); sink << x.context_;
//sink.apply_raw(x.context_.size(), const_cast<char*>(x.context_.data()));
} }
void serialize(deserializer& source, error& x, const unsigned int) { void serialize(deserializer& source, error& x, const unsigned int) {
x.context_.clear(); x.context_.reset();
uint32_t flag_size_and_code; uint32_t flag_size_and_code;
source >> flag_size_and_code; source >> flag_size_and_code;
if (flag_size_and_code & 0x80000000) { if (flag_size_and_code & 0x80000000) {
...@@ -95,10 +93,8 @@ void serialize(deserializer& source, error& x, const unsigned int) { ...@@ -95,10 +93,8 @@ void serialize(deserializer& source, error& x, const unsigned int) {
source >> x.category_; source >> x.category_;
x.code_ = static_cast<uint8_t>(flag_size_and_code & 0xFF); x.code_ = static_cast<uint8_t>(flag_size_and_code & 0xFF);
auto size = flag_size_and_code >> 8; auto size = flag_size_and_code >> 8;
if (size > 0) { if (size > 0)
x.context_.resize(size); source >> x.context_;
source.apply_raw(size, &x.context_[0]);
}
} }
int error::compare(uint8_t x, atom_value y) const { int error::compare(uint8_t x, atom_value y) const {
...@@ -114,11 +110,15 @@ int error::compare(const error& x) const { ...@@ -114,11 +110,15 @@ int error::compare(const error& x) const {
} }
std::string to_string(const error& x) { std::string to_string(const error& x) {
std::string result = "<error: ("; std::string result = "error(";
result += to_string(x.category()); result += to_string(x.category());
result += ", "; result += ", ";
result += std::to_string(static_cast<int>(x.code())); result += std::to_string(static_cast<int>(x.code()));
result += ")>"; if (! x.context().empty()) {
result += ", ";
result += to_string(x.context());
}
result += ")";
return result; return result;
} }
......
...@@ -163,6 +163,15 @@ public: ...@@ -163,6 +163,15 @@ public:
behavior make_behavior() override { behavior make_behavior() override {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
// instead of dropping "unexpected" messages,
// we simply forward them to our acquaintances
auto fwd = [=](local_actor*, const type_erased_tuple* ptr) -> result<message> {
CAF_LOG_TRACE(CAF_ARG(msg));
send_to_acquaintances(message::from(ptr));
return message{};
};
set_unexpected_handler(fwd);
// return behavior
return { return {
[=](join_atom, const actor& other) { [=](join_atom, const actor& other) {
CAF_LOG_TRACE(CAF_ARG(other)); CAF_LOG_TRACE(CAF_ARG(other));
...@@ -173,11 +182,8 @@ public: ...@@ -173,11 +182,8 @@ public:
[=](leave_atom, const actor& other) { [=](leave_atom, const actor& other) {
CAF_LOG_TRACE(CAF_ARG(other)); CAF_LOG_TRACE(CAF_ARG(other));
acquaintances_.erase(other); acquaintances_.erase(other);
// TODO
/*
if (other && acquaintances_.erase(other) > 0) if (other && acquaintances_.erase(other) > 0)
demonitor(other); demonitor(other);
*/
}, },
[=](forward_atom, const message& what) { [=](forward_atom, const message& what) {
CAF_LOG_TRACE(CAF_ARG(what)); CAF_LOG_TRACE(CAF_ARG(what));
...@@ -197,10 +203,6 @@ public: ...@@ -197,10 +203,6 @@ public:
if (i != last) if (i != last)
acquaintances_.erase(i); acquaintances_.erase(i);
} }
},
others >> [=](const message& msg) {
CAF_LOG_TRACE(CAF_ARG(msg));
send_to_acquaintances(msg);
} }
}; };
} }
...@@ -321,10 +323,19 @@ private: ...@@ -321,10 +323,19 @@ private:
behavior proxy_broker::make_behavior() { behavior proxy_broker::make_behavior() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
// instead of dropping "unexpected" messages,
// we simply forward them to our acquaintances
auto fwd = [=](local_actor*, const type_erased_tuple* x) -> result<message> {
CAF_LOG_TRACE(CAF_ARG(msg));
group_->send_all_subscribers(current_element_->sender, message::from(x),
context());
return message{};
};
set_unexpected_handler(fwd);
// return dummy behavior
return { return {
others >> [=](const message& msg) { [](const down_msg&) {
CAF_LOG_TRACE(CAF_ARG(msg)); // nop
group_->send_all_subscribers(current_element_->sender, msg, context());
} }
}; };
} }
......
...@@ -29,6 +29,7 @@ ...@@ -29,6 +29,7 @@
#include "caf/exit_reason.hpp" #include "caf/exit_reason.hpp"
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/actor_ostream.hpp"
#include "caf/blocking_actor.hpp" #include "caf/blocking_actor.hpp"
#include "caf/binary_serializer.hpp" #include "caf/binary_serializer.hpp"
#include "caf/default_attachable.hpp" #include "caf/default_attachable.hpp"
...@@ -38,6 +39,33 @@ ...@@ -38,6 +39,33 @@
namespace caf { namespace caf {
result<message> mirror_unexpected(local_actor*, const type_erased_tuple* x) {
return message::from(x);
}
result<message> mirror_unexpected_once(local_actor* self,
const type_erased_tuple* x) {
self->quit();
return mirror_unexpected(self, x);
}
result<message> skip_unexpected(local_actor*, const type_erased_tuple*) {
return skip_message();
}
result<message> print_and_drop_unexpected(local_actor* self,
const type_erased_tuple* x) {
aout(self) << "*** unexpected message [id: " << self->id()
<< ", name: " << self->name() << "]: " << x->stringify()
<< std::endl;
return sec::unexpected_message;
}
result<message> silently_drop_unexpected(local_actor*,
const type_erased_tuple*) {
return sec::unexpected_message;
}
// local actors are created with a reference count of one that is adjusted // local actors are created with a reference count of one that is adjusted
// later on in spawn(); this prevents subtle bugs that lead to segfaults, // later on in spawn(); this prevents subtle bugs that lead to segfaults,
// e.g., when calling address() in the ctor of a derived class // e.g., when calling address() in the ctor of a derived class
...@@ -46,7 +74,8 @@ local_actor::local_actor(actor_config& cfg) ...@@ -46,7 +74,8 @@ local_actor::local_actor(actor_config& cfg)
context_(cfg.host), context_(cfg.host),
planned_exit_reason_(exit_reason::not_exited), planned_exit_reason_(exit_reason::not_exited),
timeout_id_(0), timeout_id_(0),
initial_behavior_fac_(std::move(cfg.init_fun)) { initial_behavior_fac_(std::move(cfg.init_fun)),
unexpected_handler_(print_and_drop_unexpected) {
if (cfg.groups != nullptr) if (cfg.groups != nullptr)
for (auto& grp : *cfg.groups) for (auto& grp : *cfg.groups)
join(grp); join(grp);
...@@ -55,14 +84,16 @@ local_actor::local_actor(actor_config& cfg) ...@@ -55,14 +84,16 @@ local_actor::local_actor(actor_config& cfg)
local_actor::local_actor() local_actor::local_actor()
: monitorable_actor(abstract_channel::is_abstract_actor_flag), : monitorable_actor(abstract_channel::is_abstract_actor_flag),
planned_exit_reason_(exit_reason::not_exited), planned_exit_reason_(exit_reason::not_exited),
timeout_id_(0) { timeout_id_(0),
unexpected_handler_(print_and_drop_unexpected) {
// nop // nop
} }
local_actor::local_actor(int init_flags) local_actor::local_actor(int init_flags)
: monitorable_actor(init_flags), : monitorable_actor(init_flags),
planned_exit_reason_(exit_reason::not_exited), planned_exit_reason_(exit_reason::not_exited),
timeout_id_(0) { timeout_id_(0),
unexpected_handler_(print_and_drop_unexpected) {
// nop // nop
} }
...@@ -209,7 +240,7 @@ msg_type filter_msg(local_actor* self, mailbox_element& node) { ...@@ -209,7 +240,7 @@ msg_type filter_msg(local_actor* self, mailbox_element& node) {
return msg_type::response; return msg_type::response;
// intercept system messages, e.g., signalizing migration // intercept system messages, e.g., signalizing migration
if (msg.size() > 1 && msg.match_element<sys_atom>(0) && node.sender) { if (msg.size() > 1 && msg.match_element<sys_atom>(0) && node.sender) {
bool mismatch = false; bool mismatch = true;
msg.apply({ msg.apply({
/* /*
[&](sys_atom, migrate_atom, const actor& mm) { [&](sys_atom, migrate_atom, const actor& mm) {
...@@ -278,6 +309,7 @@ msg_type filter_msg(local_actor* self, mailbox_element& node) { ...@@ -278,6 +309,7 @@ msg_type filter_msg(local_actor* self, mailbox_element& node) {
*/ */
[&](sys_atom, get_atom, std::string& what) { [&](sys_atom, get_atom, std::string& what) {
CAF_LOG_TRACE(CAF_ARG(what)); CAF_LOG_TRACE(CAF_ARG(what));
mismatch = false;
if (what == "info") { if (what == "info") {
CAF_LOG_DEBUG("reply to 'info' message"); CAF_LOG_DEBUG("reply to 'info' message");
node.sender->enqueue( node.sender->enqueue(
...@@ -293,9 +325,6 @@ msg_type filter_msg(local_actor* self, mailbox_element& node) { ...@@ -293,9 +325,6 @@ msg_type filter_msg(local_actor* self, mailbox_element& node) {
node.mid.response_id(), node.mid.response_id(),
{}, sec::invalid_sys_key), {}, sec::invalid_sys_key),
self->context()); self->context());
},
others >> [&] {
mismatch = true;
} }
}); });
return mismatch ? msg_type::ordinary : msg_type::sys_message; return mismatch ? msg_type::ordinary : msg_type::sys_message;
...@@ -372,7 +401,7 @@ public: ...@@ -372,7 +401,7 @@ public:
void delegate(T& x) { void delegate(T& x) {
auto rp = self_->make_response_promise(); auto rp = self_->make_response_promise();
if (! rp.pending()) { if (! rp.pending()) {
CAF_LOG_DEBUG("suppress response message due to invalid response promise"); CAF_LOG_DEBUG("suppress response message: invalid response promise");
return; return;
} }
invoke_result_visitor_helper f{std::move(rp)}; invoke_result_visitor_helper f{std::move(rp)};
...@@ -447,11 +476,15 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr, ...@@ -447,11 +476,15 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr,
if (ref_fun.timeout().valid()) { if (ref_fun.timeout().valid()) {
ref_fun.handle_timeout(); ref_fun.handle_timeout();
} }
} else if (! ref_fun(visitor, current_element_->msg)) { } else if (ref_fun(visitor, current_element_->msg) == match_case::no_match) {
//} else if (! post_process_invoke_res(this, false, // wrap unexpected message into an error object and try invoke again
// ref_fun(current_element_->msg))) { auto tmp = make_message(make_error(sec::unexpected_response,
CAF_LOG_WARNING("multiplexed response failure occured:" << CAF_ARG(id())); current_element_->msg));
quit(exit_reason::unhandled_request_error); if (ref_fun(visitor, tmp) == match_case::no_match) {
CAF_LOG_WARNING("multiplexed response failure occured:"
<< CAF_ARG(id()));
quit(exit_reason::unhandled_request_error);
}
} }
ptr.swap(current_element_); ptr.swap(current_element_);
mark_multiplexed_arrived(mid); mark_multiplexed_arrived(mid);
...@@ -467,11 +500,13 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr, ...@@ -467,11 +500,13 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr,
if (fun.timeout().valid()) { if (fun.timeout().valid()) {
fun.handle_timeout(); fun.handle_timeout();
} }
} else { } else if (fun(visitor, current_element_->msg) == match_case::no_match) {
if (! fun(visitor, current_element_->msg)) { // wrap unexpected message into an error object and try invoke again
//if (! post_process_invoke_res(this, false, auto tmp = make_message(make_error(sec::unexpected_response,
// fun(current_element_->msg))) { current_element_->msg));
CAF_LOG_WARNING("sync response failure occured:" << CAF_ARG(id())); if (fun(visitor, tmp) == match_case::no_match) {
CAF_LOG_WARNING("multiplexed response failure occured:"
<< CAF_ARG(id()));
quit(exit_reason::unhandled_request_error); quit(exit_reason::unhandled_request_error);
} }
} }
...@@ -485,26 +520,39 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr, ...@@ -485,26 +520,39 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr,
return im_dropped; return im_dropped;
} }
case msg_type::ordinary: case msg_type::ordinary:
if (! awaited_id.valid()) { if (awaited_id.valid()) {
auto had_timeout = has_timeout(); CAF_LOG_DEBUG("skipped asynchronous message:" << CAF_ARG(awaited_id));
if (had_timeout) { return im_skipped;
has_timeout(false); }
bool skipped = false;
auto had_timeout = has_timeout();
if (had_timeout)
has_timeout(false);
ptr.swap(current_element_);
switch (fun(visitor, current_element_->msg)) {
case match_case::skip:
skipped = true;
break;
default:
break;
case match_case::no_match: {
if (had_timeout)
has_timeout(true);
auto sres = unexpected_handler_(this,
current_element_->msg.cvals().get());
if (sres.flag != rt_skip_message)
visitor.visit(sres);
else
skipped = true;
} }
ptr.swap(current_element_); }
//auto is_req = current_element_->mid.is_request(); ptr.swap(current_element_);
auto res = fun(visitor, current_element_->msg); if (skipped) {
//auto res = post_process_invoke_res(this, is_req,
// fun(current_element_->msg));
ptr.swap(current_element_);
if (res)
return im_success;
// restore timeout if necessary
if (had_timeout) if (had_timeout)
has_timeout(true); has_timeout(true);
} else { return im_skipped;
CAF_LOG_DEBUG("skipped asynchronous message:" << CAF_ARG(awaited_id));
} }
return im_skipped; return im_success;
} }
// should be unreachable // should be unreachable
CAF_CRITICAL("invalid message type"); CAF_CRITICAL("invalid message type");
......
...@@ -25,7 +25,7 @@ match_case::~match_case() { ...@@ -25,7 +25,7 @@ match_case::~match_case() {
// nop // nop
} }
match_case::match_case(bool hw, uint32_t tt) : has_wildcard_(hw), token_(tt) { match_case::match_case(uint32_t tt) : token_(tt) {
// nop // nop
} }
......
...@@ -27,6 +27,20 @@ ...@@ -27,6 +27,20 @@
namespace caf { namespace caf {
namespace detail { namespace detail {
merged_tuple::cow_ptr merged_tuple::make(message x, message y) {
data_type data{x.vals(), y.vals()};
mapping_type mapping;
auto s = x.size();
for (size_t i = 0; i < s; ++i) {
if (x.match_element<index_mapping>(i))
mapping.emplace_back(1, x.get_as<index_mapping>(i).value - 1);
else
mapping.emplace_back(0, i);
}
return cow_ptr{make_counted<merged_tuple>(std::move(data),
std::move(mapping))};
}
merged_tuple::merged_tuple(data_type xs, mapping_type ys) merged_tuple::merged_tuple(data_type xs, mapping_type ys)
: data_(std::move(xs)), : data_(std::move(xs)),
type_token_(0xFFFFFFFF), type_token_(0xFFFFFFFF),
...@@ -41,19 +55,8 @@ merged_tuple::merged_tuple(data_type xs, mapping_type ys) ...@@ -41,19 +55,8 @@ merged_tuple::merged_tuple(data_type xs, mapping_type ys)
} }
} }
// creates a typed subtuple from `d` with mapping `v` merged_tuple::cow_ptr merged_tuple::copy() const {
merged_tuple::cow_ptr merged_tuple::make(message x, message y) { return cow_ptr{make_counted<merged_tuple>(data_, mapping_)};
data_type data{x.vals(), y.vals()};
mapping_type mapping;
auto s = x.size();
for (size_t i = 0; i < s; ++i) {
if (x.match_element<index_mapping>(i))
mapping.emplace_back(1, x.get_as<index_mapping>(i).value - 1);
else
mapping.emplace_back(0, i);
}
return cow_ptr{make_counted<merged_tuple>(std::move(data),
std::move(mapping))};
} }
void* merged_tuple::get_mutable(size_t pos) { void* merged_tuple::get_mutable(size_t pos) {
...@@ -72,16 +75,6 @@ size_t merged_tuple::size() const { ...@@ -72,16 +75,6 @@ size_t merged_tuple::size() const {
return mapping_.size(); return mapping_.size();
} }
merged_tuple::cow_ptr merged_tuple::copy() const {
return cow_ptr{make_counted<merged_tuple>(data_, mapping_)};
}
const void* merged_tuple::get(size_t pos) const {
CAF_ASSERT(pos < mapping_.size());
auto& p = mapping_[pos];
return data_[p.first]->get(p.second);
}
uint32_t merged_tuple::type_token() const { uint32_t merged_tuple::type_token() const {
return type_token_; return type_token_;
} }
...@@ -92,10 +85,10 @@ merged_tuple::rtti_pair merged_tuple::type(size_t pos) const { ...@@ -92,10 +85,10 @@ merged_tuple::rtti_pair merged_tuple::type(size_t pos) const {
return data_[p.first]->type(p.second); return data_[p.first]->type(p.second);
} }
void merged_tuple::save(size_t pos, serializer& sink) const { const void* merged_tuple::get(size_t pos) const {
CAF_ASSERT(pos < mapping_.size()); CAF_ASSERT(pos < mapping_.size());
auto& p = mapping_[pos]; auto& p = mapping_[pos];
data_[p.first]->save(p.second, sink); return data_[p.first]->get(p.second);
} }
std::string merged_tuple::stringify(size_t pos) const { std::string merged_tuple::stringify(size_t pos) const {
...@@ -104,6 +97,18 @@ std::string merged_tuple::stringify(size_t pos) const { ...@@ -104,6 +97,18 @@ std::string merged_tuple::stringify(size_t pos) const {
return data_[p.first]->stringify(p.second); return data_[p.first]->stringify(p.second);
} }
type_erased_value_ptr merged_tuple::copy(size_t pos) const {
CAF_ASSERT(pos < mapping_.size());
auto& p = mapping_[pos];
return data_[p.first]->copy(p.second);
}
void merged_tuple::save(size_t pos, serializer& sink) const {
CAF_ASSERT(pos < mapping_.size());
auto& p = mapping_[pos];
data_[p.first]->save(p.second, sink);
}
const merged_tuple::mapping_type& merged_tuple::mapping() const { const merged_tuple::mapping_type& merged_tuple::mapping() const {
return mapping_; return mapping_;
} }
......
...@@ -24,6 +24,7 @@ ...@@ -24,6 +24,7 @@
#include "caf/serializer.hpp" #include "caf/serializer.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/deserializer.hpp" #include "caf/deserializer.hpp"
#include "caf/message_builder.hpp"
#include "caf/message_handler.hpp" #include "caf/message_handler.hpp"
#include "caf/string_algorithms.hpp" #include "caf/string_algorithms.hpp"
...@@ -64,6 +65,20 @@ const void* message::at(size_t p) const { ...@@ -64,6 +65,20 @@ const void* message::at(size_t p) const {
return vals_->get(p); return vals_->get(p);
} }
message message::from(const type_erased_tuple* ptr) {
if (! ptr)
return message{};
auto dptr = dynamic_cast<const detail::message_data*>(ptr);
if (dptr) {
data_ptr dp{const_cast<detail::message_data*>(dptr), true};
return message{std::move(dp)};
}
message_builder mb;
for (size_t i = 0; i < ptr->size(); ++i)
mb.emplace(ptr->copy(i));
return mb.move_to_message();
}
bool message::match_element(size_t pos, uint16_t typenr, bool message::match_element(size_t pos, uint16_t typenr,
const std::type_info* rtti) const { const std::type_info* rtti) const {
return vals_->matches(pos, typenr, rtti); return vals_->matches(pos, typenr, rtti);
......
...@@ -19,7 +19,6 @@ ...@@ -19,7 +19,6 @@
#include "caf/monitorable_actor.hpp" #include "caf/monitorable_actor.hpp"
#include "caf/on.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/actor_cast.hpp" #include "caf/actor_cast.hpp"
...@@ -243,11 +242,10 @@ bool monitorable_actor::handle_system_message(mailbox_element& node, ...@@ -243,11 +242,10 @@ bool monitorable_actor::handle_system_message(mailbox_element& node,
node.mid.response_id(), {}, node.mid.response_id(), {},
ok_atom::value, std::move(what), ok_atom::value, std::move(what),
address(), name()); address(), name());
},
others >> [&] {
err = sec::unsupported_sys_message;
} }
}); });
if (! res && ! err)
err = sec::unsupported_sys_message;
if (err && node.mid.is_request()) if (err && node.mid.is_request())
res = mailbox_element::make_joint(ctrl(), node.mid.response_id(), res = mailbox_element::make_joint(ctrl(), node.mid.response_id(),
{}, std::move(err)); {}, std::move(err));
......
...@@ -21,41 +21,46 @@ ...@@ -21,41 +21,46 @@
namespace caf { namespace caf {
namespace {
const char* sec_strings[] = {
"<no-error>",
"unexpected_message",
"invalid_invoke_mutable",
"unexpected_response",
"request_receiver_down",
"request_timeout",
"no_actor_to_unpublish",
"no_actor_published_at_port",
"state_not_serializable",
"invalid_sys_key",
"unsupported_sys_message",
"disconnect_during_handshake",
"cannot_forward_to_invalid_actor",
"no_route_to_receiving_node",
"failed_to_assign_scribe_from_handle",
"cannot_close_invalid_port",
"cannot_connect_to_node",
"cannot_open_port",
"cannot_spawn_actor_from_arguments",
"no_such_riac_node"
};
} // namespace <anonymous>
const char* to_string(sec x) { const char* to_string(sec x) {
switch (x) { auto index = static_cast<size_t>(x);
case sec::unexpected_message: if (index > static_cast<size_t>(sec::no_such_riac_node))
return "unexpected_message"; return "<unknown>";
case sec::no_actor_to_unpublish: return sec_strings[index];
return "no_actor_to_unpublish";
case sec::no_actor_published_at_port:
return "no_actor_published_at_port";
case sec::state_not_serializable:
return "state_not_serializable";
case sec::invalid_sys_key:
return "invalid_sys_key";
case sec::disconnect_during_handshake:
return "disconnect_during_handshake";
case sec::cannot_forward_to_invalid_actor:
return "cannot_forward_to_invalid_actor";
case sec::no_route_to_receiving_node:
return "no_route_to_receiving_node";
case sec::failed_to_assign_scribe_from_handle:
return "failed_to_assign_scribe_from_handle";
case sec::cannot_close_invalid_port:
return "cannot_close_invalid_port";
case sec::cannot_connect_to_node:
return "cannot_connect_to_node";
case sec::cannot_open_port:
return "cannot_open_port";
case sec::cannot_spawn_actor_from_arguments:
return "cannot_spawn_actor_from_arguments";
default:
return "<unknown>";
}
} }
error make_error(sec x) { error make_error(sec x) {
return {static_cast<uint8_t>(x), atom("system")}; return {static_cast<uint8_t>(x), atom("system")};
} }
error make_error(sec x, message context) {
return {static_cast<uint8_t>(x), atom("system"), std::move(context)};
}
} // namespace caf } // namespace caf
...@@ -41,6 +41,51 @@ struct splitter_state { ...@@ -41,6 +41,51 @@ struct splitter_state {
behavior fan_out_fan_in(stateful_actor<splitter_state>* self, behavior fan_out_fan_in(stateful_actor<splitter_state>* self,
const std::vector<strong_actor_ptr>& workers) { const std::vector<strong_actor_ptr>& workers) {
auto f = [=](local_actor*, const type_erased_tuple* x) -> result<message> {
auto msg = message::from(x);
self->state.rp = self->make_response_promise();
self->state.pending = workers.size();
// request().await() has LIFO ordering
for (auto i = workers.rbegin(); i != workers.rend(); ++i)
// TODO: maybe infer some useful timeout or use config parameter?
self->request(actor_cast<actor>(*i), infinite, msg).await(
[=]() {
// nop
},
[=](error& err) mutable {
if (err == sec::unexpected_response) {
self->state.result += std::move(err.context());
if (--self->state.pending == 0)
self->state.rp.deliver(std::move(self->state.result));
} else {
self->state.rp.deliver(err);
self->quit();
}
}
);
return delegated<message>{};
};
self->set_unexpected_handler(f);
return [] {
// nop
};
/*
// TODO: maybe infer some useful timeout or use config parameter?
self->request(actor_cast<actor>(*i), infinite, msg)
.generic_await(
[=](const message& tmp) {
self->state.result += tmp;
if (--self->state.pending == 0)
self->state.rp.deliver(std::move(self->state.result));
},
[=](const error& err) {
self->state.rp.deliver(err);
self->quit();
}
);
};
set_unexpected_handler(f);
return { return {
others >> [=](const message& msg) { others >> [=](const message& msg) {
self->state.rp = self->make_response_promise(); self->state.rp = self->make_response_promise();
...@@ -63,6 +108,7 @@ behavior fan_out_fan_in(stateful_actor<splitter_state>* self, ...@@ -63,6 +108,7 @@ behavior fan_out_fan_in(stateful_actor<splitter_state>* self,
self->unbecome(); self->unbecome();
} }
}; };
*/
} }
} // namespace <anonymous> } // namespace <anonymous>
......
...@@ -52,8 +52,8 @@ public: ...@@ -52,8 +52,8 @@ public:
behavior make_behavior() override { behavior make_behavior() override {
return { return {
others >> [=](const message& msg) { [=](int x) {
return msg; return x;
} }
}; };
} }
......
...@@ -108,11 +108,6 @@ CAF_TEST(round_robin_actor_pool) { ...@@ -108,11 +108,6 @@ CAF_TEST(round_robin_actor_pool) {
} }
); );
anon_send_exit(workers.back(), exit_reason::user_shutdown); anon_send_exit(workers.back(), exit_reason::user_shutdown);
self->receive(
after(std::chrono::milliseconds(25)) >> [] {
// wait some time to give the pool time to remove the failed worker
}
);
self->receive( self->receive(
[&](const down_msg& dm) { [&](const down_msg& dm) {
CAF_CHECK(dm.source == workers.back()); CAF_CHECK(dm.source == workers.back());
......
...@@ -96,12 +96,8 @@ CAF_TEST(receive_atoms) { ...@@ -96,12 +96,8 @@ CAF_TEST(receive_atoms) {
} }
CAF_CHECK(matched_pattern[0] && matched_pattern[1] && matched_pattern[2]); CAF_CHECK(matched_pattern[0] && matched_pattern[1] && matched_pattern[2]);
self->receive( self->receive(
// "erase" message { 'b', 'a, 'c', 23.f } [](float) {
others >> [] { // erase float message
CAF_MESSAGE("drain mailbox");
},
after(std::chrono::seconds(0)) >> [] {
CAF_ERROR("mailbox empty");
} }
); );
atom_value x = atom("abc"); atom_value x = atom("abc");
...@@ -112,9 +108,6 @@ CAF_TEST(receive_atoms) { ...@@ -112,9 +108,6 @@ CAF_TEST(receive_atoms) {
self->receive( self->receive(
[](abc_atom) { [](abc_atom) {
CAF_MESSAGE("received 'abc'"); CAF_MESSAGE("received 'abc'");
},
others >> [] {
CAF_ERROR("unexpected message");
} }
); );
} }
......
...@@ -72,8 +72,8 @@ CAF_TEST(constructor_attach) { ...@@ -72,8 +72,8 @@ CAF_TEST(constructor_attach) {
quit(reason); quit(reason);
} }
}, },
others >> [=] { [=](die_atom x) {
forward_to(testee_); return delegate(testee_, x);
} }
}; };
} }
......
...@@ -36,7 +36,7 @@ public: ...@@ -36,7 +36,7 @@ public:
} }
behavior make_behavior() override { behavior make_behavior() override {
return { return {
others >> [] { [](const std::string&) {
throw std::runtime_error("whatever"); throw std::runtime_error("whatever");
} }
}; };
......
...@@ -211,10 +211,10 @@ public: ...@@ -211,10 +211,10 @@ public:
} }
behavior make_behavior() override { behavior make_behavior() override {
set_unexpected_handler(mirror_unexpected);
return { return {
others >> [=](const message& msg) -> message { [] {
quit(exit_reason::normal); // nop
return msg;
} }
}; };
} }
...@@ -231,10 +231,10 @@ public: ...@@ -231,10 +231,10 @@ public:
} }
behavior make_behavior() override { behavior make_behavior() override {
set_unexpected_handler(mirror_unexpected);
return { return {
others >> [=](const message& msg) { [] {
CAF_MESSAGE("simple_mirror: return current message"); // nop
return msg;
} }
}; };
} }
...@@ -255,14 +255,8 @@ behavior high_priority_testee(event_based_actor* self) { ...@@ -255,14 +255,8 @@ behavior high_priority_testee(event_based_actor* self) {
[=](b_atom) { [=](b_atom) {
CAF_MESSAGE("received \"b\" atom, about to quit"); CAF_MESSAGE("received \"b\" atom, about to quit");
self->quit(); self->quit();
},
others >> [&] {
CAF_ERROR("Unexpected message");
} }
); );
},
others >> [&] {
CAF_ERROR("Unexpected message");
} }
}; };
} }
...@@ -289,9 +283,6 @@ behavior slave(event_based_actor* self, actor master) { ...@@ -289,9 +283,6 @@ behavior slave(event_based_actor* self, actor master) {
[=](const exit_msg& msg) { [=](const exit_msg& msg) {
CAF_MESSAGE("slave: received exit message"); CAF_MESSAGE("slave: received exit message");
self->quit(msg.reason); self->quit(msg.reason);
},
others >> [&] {
CAF_ERROR("Unexpected message");
} }
}; };
} }
...@@ -358,7 +349,7 @@ CAF_TEST(detached_actors_and_schedulued_actors) { ...@@ -358,7 +349,7 @@ CAF_TEST(detached_actors_and_schedulued_actors) {
CAF_TEST(self_receive_with_zero_timeout) { CAF_TEST(self_receive_with_zero_timeout) {
scoped_actor self{system}; scoped_actor self{system};
self->receive( self->receive(
others >> [&] { [&] {
CAF_ERROR("Unexpected message"); CAF_ERROR("Unexpected message");
}, },
after(chrono::seconds(0)) >> [] { /* mailbox empty */ } after(chrono::seconds(0)) >> [] { /* mailbox empty */ }
...@@ -372,23 +363,15 @@ CAF_TEST(mirror) { ...@@ -372,23 +363,15 @@ CAF_TEST(mirror) {
self->receive ( self->receive (
[](const std::string& msg) { [](const std::string& msg) {
CAF_CHECK_EQUAL(msg, "hello mirror"); CAF_CHECK_EQUAL(msg, "hello mirror");
},
others >> [&] {
CAF_ERROR("Unexpected message");
} }
); );
self->send_exit(mirror, exit_reason::user_shutdown); self->send_exit(mirror, exit_reason::user_shutdown);
self->receive ( self->receive (
[&](const down_msg& dm) { [&](const down_msg& dm) {
if (dm.reason == exit_reason::user_shutdown) { if (dm.reason == exit_reason::user_shutdown)
CAF_MESSAGE("received `down_msg`"); CAF_MESSAGE("received `down_msg`");
} else
else {
CAF_ERROR("Unexpected message"); CAF_ERROR("Unexpected message");
}
},
others >> [&] {
CAF_ERROR("Unexpected message");
} }
); );
} }
...@@ -400,23 +383,15 @@ CAF_TEST(detached_mirror) { ...@@ -400,23 +383,15 @@ CAF_TEST(detached_mirror) {
self->receive ( self->receive (
[](const std::string& msg) { [](const std::string& msg) {
CAF_CHECK_EQUAL(msg, "hello mirror"); CAF_CHECK_EQUAL(msg, "hello mirror");
},
others >> [&] {
CAF_ERROR("Unexpected message");
} }
); );
self->send_exit(mirror, exit_reason::user_shutdown); self->send_exit(mirror, exit_reason::user_shutdown);
self->receive ( self->receive (
[&](const down_msg& dm) { [&](const down_msg& dm) {
if (dm.reason == exit_reason::user_shutdown) { if (dm.reason == exit_reason::user_shutdown)
CAF_MESSAGE("received `down_msg`"); CAF_MESSAGE("received `down_msg`");
} else
else {
CAF_ERROR("Unexpected message"); CAF_ERROR("Unexpected message");
}
},
others >> [&] {
CAF_ERROR("Unexpected message");
} }
); );
} }
...@@ -429,23 +404,15 @@ CAF_TEST(priority_aware_mirror) { ...@@ -429,23 +404,15 @@ CAF_TEST(priority_aware_mirror) {
self->receive ( self->receive (
[](const std::string& msg) { [](const std::string& msg) {
CAF_CHECK_EQUAL(msg, "hello mirror"); CAF_CHECK_EQUAL(msg, "hello mirror");
},
others >> [&] {
CAF_ERROR("Unexpected message");
} }
); );
self->send_exit(mirror, exit_reason::user_shutdown); self->send_exit(mirror, exit_reason::user_shutdown);
self->receive ( self->receive (
[&](const down_msg& dm) { [&](const down_msg& dm) {
if (dm.reason == exit_reason::user_shutdown) { if (dm.reason == exit_reason::user_shutdown)
CAF_MESSAGE("received `down_msg`"); CAF_MESSAGE("received `down_msg`");
} else
else {
CAF_ERROR("Unexpected message"); CAF_ERROR("Unexpected message");
}
},
others >> [&] {
CAF_ERROR("Unexpected message");
} }
); );
} }
...@@ -462,7 +429,7 @@ CAF_TEST(send_to_self) { ...@@ -462,7 +429,7 @@ CAF_TEST(send_to_self) {
} }
); );
self->send(self, message{}); self->send(self, message{});
self->receive(on() >> [] {}); self->receive([] {});
} }
CAF_TEST(echo_actor_messaging) { CAF_TEST(echo_actor_messaging) {
...@@ -472,9 +439,6 @@ CAF_TEST(echo_actor_messaging) { ...@@ -472,9 +439,6 @@ CAF_TEST(echo_actor_messaging) {
self->receive( self->receive(
[](const std::string& arg) { [](const std::string& arg) {
CAF_CHECK_EQUAL(arg, "hello echo"); CAF_CHECK_EQUAL(arg, "hello echo");
},
others >> [&] {
CAF_ERROR("Unexpected message");
} }
); );
} }
...@@ -507,61 +471,6 @@ CAF_TEST(spawn_event_testee2_test) { ...@@ -507,61 +471,6 @@ CAF_TEST(spawn_event_testee2_test) {
); );
} }
// exclude this tests; advanced match cases are currently not supported on MSVC
#ifndef CAF_WINDOWS
CAF_TEST(requests) {
scoped_actor self{system};
auto sync_testee = system.spawn([](blocking_actor* s) {
s->receive (
on("hi", arg_match) >> [&](actor from) {
s->request(from, chrono::minutes(1), "whassup?", s).receive(
[&](const string& str) {
CAF_CHECK(s->current_sender());
CAF_CHECK_EQUAL(str, "nothing");
s->send(from, "goodbye!");
}
);
},
others >> [&] {
CAF_ERROR("Unexpected message");
}
);
});
self->monitor(sync_testee);
self->send(sync_testee, "hi", self);
self->receive (
on("whassup?", arg_match) >> [&](actor other) -> std::string {
CAF_MESSAGE("received \"whassup?\" message");
// this is NOT a reply, it's just an asynchronous message
self->send(other, "a lot!");
return "nothing";
}
);
self->receive (
on("goodbye!") >> [] { CAF_MESSAGE("Received \"goodbye!\""); },
after(chrono::seconds(1)) >> [] { CAF_ERROR("Unexpected timeout"); }
);
self->receive (
[&](const down_msg& dm) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::normal);
CAF_CHECK(dm.source == sync_testee);
}
);
self->await_all_other_actors_done();
self->request(sync_testee, chrono::microseconds(1), "!?").receive(
[] {
CAF_ERROR("Unexpected empty message");
},
[&](error& err) {
if (err == sec::request_receiver_down)
CAF_MESSAGE("received `request_receiver_down`");
else
CAF_ERROR("received unexpected error: " << self->system().render(err));
}
);
}
#endif // CAF_WINDOWS
CAF_TEST(function_spawn) { CAF_TEST(function_spawn) {
scoped_actor self{system}; scoped_actor self{system};
auto f = [](const string& name) -> behavior { auto f = [](const string& name) -> behavior {
...@@ -625,8 +534,8 @@ CAF_TEST(constructor_attach) { ...@@ -625,8 +534,8 @@ CAF_TEST(constructor_attach) {
behavior make_behavior() { behavior make_behavior() {
return { return {
others >> [=] { [] {
CAF_ERROR("Unexpected message"); // nop
} }
}; };
} }
...@@ -646,6 +555,13 @@ CAF_TEST(constructor_attach) { ...@@ -646,6 +555,13 @@ CAF_TEST(constructor_attach) {
behavior make_behavior() { behavior make_behavior() {
trap_exit(true); trap_exit(true);
testee_ = spawn<testee, monitored>(this); testee_ = spawn<testee, monitored>(this);
auto f = [=](local_actor*, const type_erased_tuple*) -> result<message> {
CAF_MESSAGE("forward to testee");
//auto msg = message::from(x);
forward_to(testee_);
return delegated<message>{};
};
set_unexpected_handler(f);
return { return {
[=](const down_msg& msg) { [=](const down_msg& msg) {
CAF_CHECK_EQUAL(msg.reason, exit_reason::user_shutdown); CAF_CHECK_EQUAL(msg.reason, exit_reason::user_shutdown);
...@@ -658,10 +574,6 @@ CAF_TEST(constructor_attach) { ...@@ -658,10 +574,6 @@ CAF_TEST(constructor_attach) {
if (++downs_ == 2) { if (++downs_ == 2) {
quit(reason); quit(reason);
} }
},
others >> [=] {
CAF_MESSAGE("forward to testee");
forward_to(testee_);
} }
}; };
} }
...@@ -689,8 +601,8 @@ public: ...@@ -689,8 +601,8 @@ public:
} }
behavior make_behavior() override { behavior make_behavior() override {
return { return {
others >> [] { [](const std::string& str) {
throw std::runtime_error("whatever"); throw std::runtime_error(str);
} }
}; };
} }
...@@ -746,8 +658,8 @@ CAF_TEST(kill_the_immortal) { ...@@ -746,8 +658,8 @@ CAF_TEST(kill_the_immortal) {
auto wannabe_immortal = system.spawn([](event_based_actor* self) -> behavior { auto wannabe_immortal = system.spawn([](event_based_actor* self) -> behavior {
self->trap_exit(true); self->trap_exit(true);
return { return {
others >> [=] { [] {
CAF_ERROR("Unexpected message"); // nop
} }
}; };
}); });
...@@ -764,7 +676,13 @@ CAF_TEST(kill_the_immortal) { ...@@ -764,7 +676,13 @@ CAF_TEST(kill_the_immortal) {
CAF_TEST(exit_reason_in_scoped_actor) { CAF_TEST(exit_reason_in_scoped_actor) {
scoped_actor self{system}; scoped_actor self{system};
self->spawn<linked>([]() -> behavior { return others >> [] {}; }); self->spawn<linked>(
[]() -> behavior {
return [] {
// nop
};
}
);
self->planned_exit_reason(exit_reason::unhandled_exception); self->planned_exit_reason(exit_reason::unhandled_exception);
} }
...@@ -774,7 +692,7 @@ CAF_TEST(move_only_argument) { ...@@ -774,7 +692,7 @@ CAF_TEST(move_only_argument) {
auto f = [](event_based_actor* self, unique_int ptr) -> behavior { auto f = [](event_based_actor* self, unique_int ptr) -> behavior {
auto i = *ptr; auto i = *ptr;
return { return {
others >> [=] { [=](float) {
self->quit(); self->quit();
return i; return i;
} }
......
...@@ -19,15 +19,11 @@ ...@@ -19,15 +19,11 @@
#include "caf/config.hpp" #include "caf/config.hpp"
// exclude this suite; advanced match cases are currently not supported on MSVC
#ifndef CAF_WINDOWS
#define CAF_SUITE match #define CAF_SUITE match
#include "caf/test/unit_test.hpp" #include "caf/test/unit_test.hpp"
#include <functional> #include <functional>
#include "caf/on.hpp"
#include "caf/message_builder.hpp" #include "caf/message_builder.hpp"
#include "caf/message_handler.hpp" #include "caf/message_handler.hpp"
...@@ -105,11 +101,7 @@ function<void()> f(int idx) { ...@@ -105,11 +101,7 @@ function<void()> f(int idx) {
} }
CAF_TEST(atom_constants) { CAF_TEST(atom_constants) {
auto expr = on(hi_atom::value) >> f(0); message_handler expr{
CAF_CHECK_EQUAL(invoked(expr, hi_atom::value), 0);
CAF_CHECK_EQUAL(invoked(expr, ho_atom::value), -1);
CAF_CHECK_EQUAL(invoked(expr, hi_atom::value, hi_atom::value), -1);
message_handler expr2{
[](hi_atom) { [](hi_atom) {
s_invoked[0] = true; s_invoked[0] = true;
}, },
...@@ -117,86 +109,7 @@ CAF_TEST(atom_constants) { ...@@ -117,86 +109,7 @@ CAF_TEST(atom_constants) {
s_invoked[1] = true; s_invoked[1] = true;
} }
}; };
CAF_CHECK_EQUAL(invoked(expr2, ok_atom::value), -1); CAF_CHECK_EQUAL(invoked(expr, ok_atom::value), -1);
CAF_CHECK_EQUAL(invoked(expr2, hi_atom::value), 0); CAF_CHECK_EQUAL(invoked(expr, hi_atom::value), 0);
CAF_CHECK_EQUAL(invoked(expr2, ho_atom::value), 1); CAF_CHECK_EQUAL(invoked(expr, ho_atom::value), 1);
}
CAF_TEST(guards_called) {
bool guard_called = false;
auto guard = [&](int arg) -> optional<int> {
guard_called = true;
return arg;
};
auto expr = on(guard) >> f(0);
CAF_CHECK_EQUAL(invoked(expr, 42), 0);
CAF_CHECK_EQUAL(guard_called, true);
}
CAF_TEST(forwarding_optionals) {
auto expr = (
on(starts_with("--")) >> [](const string& str) {
CAF_CHECK_EQUAL(str, "help");
s_invoked[0] = true;
}
);
CAF_CHECK_EQUAL(invoked(expr, "--help"), 0);
CAF_CHECK_EQUAL(invoked(expr, "-help"), -1);
CAF_CHECK_EQUAL(invoked(expr, "--help", "--help"), -1);
CAF_CHECK_EQUAL(invoked(expr, 42), -1);
}
CAF_TEST(projections) {
auto expr = (
on(toint) >> [](int i) {
CAF_CHECK_EQUAL(i, 42);
s_invoked[0] = true;
}
);
CAF_CHECK_EQUAL(invoked(expr, "42"), 0);
CAF_CHECK_EQUAL(invoked(expr, "42f"), -1);
CAF_CHECK_EQUAL(invoked(expr, "42", "42"), -1);
CAF_CHECK_EQUAL(invoked(expr, 42), -1);
}
struct wrapped_int {
int value;
};
template <class T>
void serialize(T& in_out, wrapped_int& x, const unsigned int) {
in_out & x.value;
}
inline bool operator==(const wrapped_int& lhs, const wrapped_int& rhs) {
return lhs.value == rhs.value;
}
CAF_TEST(arg_match_pattern) {
auto expr = on(42, arg_match) >> [](int i) {
s_invoked[0] = true;
CAF_CHECK_EQUAL(i, 1);
};
CAF_CHECK_EQUAL(invoked(expr, 42, 1.f), -1);
CAF_CHECK_EQUAL(invoked(expr, 42), -1);
CAF_CHECK_EQUAL(invoked(expr, 1, 42), -1);
CAF_CHECK_EQUAL(invoked(expr, 42, 1), 0);
auto expr2 = on("-a", arg_match) >> [](const string& value) {
s_invoked[0] = true;
CAF_CHECK_EQUAL(value, "b");
};
CAF_CHECK_EQUAL(invoked(expr2, "b", "-a"), -1);
CAF_CHECK_EQUAL(invoked(expr2, "-a"), -1);
CAF_CHECK_EQUAL(invoked(expr2, "-a", "b"), 0);
auto expr3 = on(wrapped_int{42}, arg_match) >> [](wrapped_int i) {
s_invoked[0] = true;
CAF_CHECK_EQUAL(i.value, 1);
};
CAF_CHECK_EQUAL(invoked(expr3, wrapped_int{42}, 1.f), -1);
CAF_CHECK_EQUAL(invoked(expr3, 42), -1);
CAF_CHECK_EQUAL(invoked(expr3, wrapped_int{1}, wrapped_int{42}), -1);
CAF_CHECK_EQUAL(invoked(expr3, 42, 1), -1);
CAF_CHECK_EQUAL(invoked(expr3, wrapped_int{42}, wrapped_int{1}), 0);
} }
#endif // CAF_WINDOWS
...@@ -42,11 +42,17 @@ public: ...@@ -42,11 +42,17 @@ public:
} }
behavior make_behavior() override { behavior make_behavior() override {
auto f = [](local_actor* self, const type_erased_tuple* x) {
auto ptr = dynamic_cast<const detail::message_data*>(x);
CAF_REQUIRE(ptr);
CAF_CHECK_EQUAL(ptr->get_reference_count(), 2u);
self->quit();
return message::from(x);
};
set_unexpected_handler(f);
return { return {
others >> [=](message& msg) { [] {
CAF_CHECK_EQUAL(msg.cvals()->get_reference_count(), 2u); // nop
quit();
return std::move(msg);
} }
}; };
} }
...@@ -83,9 +89,6 @@ behavior tester::make_behavior() { ...@@ -83,9 +89,6 @@ behavior tester::make_behavior() {
&& dm.reason == exit_reason::normal && dm.reason == exit_reason::normal
&& current_message().cvals()->get_reference_count() == 1); && current_message().cvals()->get_reference_count() == 1);
quit(); quit();
},
others >> [&] {
CAF_ERROR("Unexpected message");
} }
}; };
} }
......
This diff is collapsed.
...@@ -41,11 +41,13 @@ using sub4_atom = atom_constant<atom("sub4")>; ...@@ -41,11 +41,13 @@ using sub4_atom = atom_constant<atom("sub4")>;
CAF_TEST(test_serial_reply) { CAF_TEST(test_serial_reply) {
actor_system system; actor_system system;
auto mirror_behavior = [=](event_based_actor* self) { auto mirror_behavior = [=](event_based_actor* self) -> behavior {
self->become(others >> [=](const message& msg) -> message { self->set_unexpected_handler(mirror_unexpected);
CAF_MESSAGE("return msg: " << to_string(msg)); return {
return msg; [] {
}); // nop
}
};
}; };
auto master = system.spawn([=](event_based_actor* self) { auto master = system.spawn([=](event_based_actor* self) {
CAF_MESSAGE("ID of master: " << self->id()); CAF_MESSAGE("ID of master: " << self->id());
......
...@@ -289,7 +289,6 @@ CAF_TEST(custom_struct) { ...@@ -289,7 +289,6 @@ CAF_TEST(custom_struct) {
} }
CAF_TEST(atoms) { CAF_TEST(atoms) {
atom_value x;
auto foo = atom("foo"); auto foo = atom("foo");
CAF_CHECK(foo == roundtrip(foo)); CAF_CHECK(foo == roundtrip(foo));
CAF_CHECK(foo == msg_roundtrip(foo)); CAF_CHECK(foo == msg_roundtrip(foo));
......
...@@ -30,8 +30,8 @@ namespace { ...@@ -30,8 +30,8 @@ namespace {
behavior testee() { behavior testee() {
return { return {
others >> [] { [] {
// ignore // nop
} }
}; };
} }
......
...@@ -30,18 +30,17 @@ CAF_TEST(simple_reply_response) { ...@@ -30,18 +30,17 @@ CAF_TEST(simple_reply_response) {
actor_system system; actor_system system;
auto s = system.spawn([](event_based_actor* self) -> behavior { auto s = system.spawn([](event_based_actor* self) -> behavior {
return ( return (
others >> [=](const message& msg) -> message { [=](ok_atom x) -> atom_value {
CAF_CHECK(to_string(msg) == "('ok')");
self->quit(); self->quit();
return msg; return x;
} }
); );
}); });
scoped_actor self{system}; scoped_actor self{system};
self->send(s, ok_atom::value); self->send(s, ok_atom::value);
self->receive( self->receive(
others >> [&](const message& msg) { [](ok_atom) {
CAF_CHECK(to_string(msg) == "('ok')"); // nop
} }
); );
} }
...@@ -184,28 +184,55 @@ public: ...@@ -184,28 +184,55 @@ public:
} }
behavior_type wait4string() { behavior_type wait4string() {
return {on<get_state_msg>() >> [] { return "wait4string"; }, return {
on<string>() >> [=] { become(wait4int()); }, [](const get_state_msg&) {
(on<float>() || on<int>()) >> skip_message}; return "wait4string";
},
[=](const string&) {
become(wait4int());
},
[](float) {
return skip_message();
},
[](int) {
return skip_message();
}
};
} }
behavior_type wait4int() { behavior_type wait4int() {
return { return {
on<get_state_msg>() >> [] { return "wait4int"; }, [](const get_state_msg&) {
on<int>() >> [=]()->int {become(wait4float()); return "wait4int";
},
[=](int) -> int {
become(wait4float());
return 42; return 42;
}, },
(on<float>() || on<string>()) >> skip_message [](float) {
return skip_message();
},
[](const string&) {
return skip_message();
}
}; };
} }
behavior_type wait4float() { behavior_type wait4float() {
return { return {
on<get_state_msg>() >> [] { [](const get_state_msg&) {
return "wait4float"; return "wait4float";
}, },
on<float>() >> [=] { become(wait4string()); }, [=](float) {
(on<string>() || on<int>()) >> skip_message}; become(wait4string());
},
[](const string&) {
return skip_message();
},
[](int) {
return skip_message();
}
};
} }
behavior_type make_behavior() override { behavior_type make_behavior() override {
...@@ -344,9 +371,11 @@ CAF_TEST(typed_spawns) { ...@@ -344,9 +371,11 @@ CAF_TEST(typed_spawns) {
CAF_MESSAGE("finished test series with `typed_server2`"); CAF_MESSAGE("finished test series with `typed_server2`");
scoped_actor self{system}; scoped_actor self{system};
test_typed_spawn(self->spawn<typed_server3>("hi there", self)); test_typed_spawn(self->spawn<typed_server3>("hi there", self));
self->receive(on("hi there") >> [] { self->receive(
CAF_MESSAGE("received \"hi there\""); [](const string& str) {
}); CAF_REQUIRE(str == "hi there");
}
);
} }
CAF_TEST(event_testee_series) { CAF_TEST(event_testee_series) {
......
...@@ -24,6 +24,7 @@ ...@@ -24,6 +24,7 @@
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/send.hpp" #include "caf/send.hpp"
#include "caf/after.hpp"
#include "caf/exception.hpp" #include "caf/exception.hpp"
#include "caf/make_counted.hpp" #include "caf/make_counted.hpp"
#include "caf/event_based_actor.hpp" #include "caf/event_based_actor.hpp"
...@@ -277,10 +278,6 @@ void basp_broker_state::learned_new_node(const node_id& nid) { ...@@ -277,10 +278,6 @@ void basp_broker_state::learned_new_node(const node_id& nid) {
[=](const down_msg& dm) { [=](const down_msg& dm) {
CAF_LOG_TRACE(CAF_ARG(dm)); CAF_LOG_TRACE(CAF_ARG(dm));
this_actor->quit(dm.reason); this_actor->quit(dm.reason);
},
others >> [=](const message& msg) {
CAF_LOG_ERROR("unexpected:" << CAF_ARG(msg));
static_cast<void>(msg); // suppress unused warning
} }
); );
}, },
...@@ -658,12 +655,8 @@ behavior basp_broker::make_behavior() { ...@@ -658,12 +655,8 @@ behavior basp_broker::make_behavior() {
state.instance.handle_heartbeat(context()); state.instance.handle_heartbeat(context());
delayed_send(this, std::chrono::milliseconds{interval}, delayed_send(this, std::chrono::milliseconds{interval},
tick_atom::value, interval); tick_atom::value, interval);
}, }
// catch-all error handler };
others >> [=](const message& msg) {
CAF_LOG_ERROR("unexpected message: " << CAF_ARG(msg));
static_cast<void>(msg); // suppress unused warning
}};
} }
resumable::resume_result basp_broker::resume(execution_unit* ctx, size_t mt) { resumable::resume_result basp_broker::resume(execution_unit* ctx, size_t mt) {
......
...@@ -27,6 +27,7 @@ ...@@ -27,6 +27,7 @@
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/send.hpp" #include "caf/send.hpp"
#include "caf/actor.hpp" #include "caf/actor.hpp"
#include "caf/after.hpp"
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/node_id.hpp" #include "caf/node_id.hpp"
......
...@@ -37,6 +37,14 @@ ...@@ -37,6 +37,14 @@
#include "caf/io/network/interfaces.hpp" #include "caf/io/network/interfaces.hpp"
#include "caf/io/network/test_multiplexer.hpp" #include "caf/io/network/test_multiplexer.hpp"
namespace {
struct anything { };
anything any_vals;
} // namespace <anonymous>
namespace std { namespace std {
ostream& operator<<(ostream& out, const caf::io::basp::message_type& x) { ostream& operator<<(ostream& out, const caf::io::basp::message_type& x) {
...@@ -44,11 +52,11 @@ ostream& operator<<(ostream& out, const caf::io::basp::message_type& x) { ...@@ -44,11 +52,11 @@ ostream& operator<<(ostream& out, const caf::io::basp::message_type& x) {
} }
template <class T> template <class T>
ostream& operator<<(ostream& out, const caf::variant<caf::anything, T>& x) { ostream& operator<<(ostream& out, const caf::variant<anything, T>& x) {
using std::to_string; using std::to_string;
using caf::to_string; using caf::to_string;
using caf::io::basp::to_string; using caf::io::basp::to_string;
if (get<caf::anything>(&x) != nullptr) if (get<anything>(&x) != nullptr)
return out << "*"; return out << "*";
return out << to_string(get<T>(x)); return out << to_string(get<T>(x));
} }
...@@ -355,8 +363,7 @@ public: ...@@ -355,8 +363,7 @@ public:
variant<anything, actor_id> source_actor, variant<anything, actor_id> source_actor,
variant<anything, actor_id> dest_actor, variant<anything, actor_id> dest_actor,
const Ts&... xs) { const Ts&... xs) {
CAF_MESSAGE("expect " << num << ". sent message to be a " CAF_MESSAGE("expect " << num);
<< operation);
buffer buf; buffer buf;
this_->to_payload(buf, xs...); this_->to_payload(buf, xs...);
buffer& ob = this_->mpx()->output_buffer(hdl); buffer& ob = this_->mpx()->output_buffer(hdl);
...@@ -637,10 +644,10 @@ CAF_TEST(remote_actor_and_send) { ...@@ -637,10 +644,10 @@ CAF_TEST(remote_actor_and_send) {
CAF_TEST(actor_serialize_and_deserialize) { CAF_TEST(actor_serialize_and_deserialize) {
auto testee_impl = [](event_based_actor* testee_self) -> behavior { auto testee_impl = [](event_based_actor* testee_self) -> behavior {
testee_self->set_unexpected_handler(mirror_unexpected_once);
return { return {
others >> [=] { [] {
testee_self->quit(); // nop
return testee_self->current_message();
} }
}; };
}; };
......
...@@ -49,20 +49,15 @@ void ping(event_based_actor* self, size_t num_pings) { ...@@ -49,20 +49,15 @@ void ping(event_based_actor* self, size_t num_pings) {
CAF_MESSAGE("received `kickoff_atom`"); CAF_MESSAGE("received `kickoff_atom`");
self->send(pong, ping_atom::value, 1); self->send(pong, ping_atom::value, 1);
self->become( self->become(
[=](pong_atom, int value)->std::tuple<atom_value, int> { [=](pong_atom, int value)->std::tuple<atom_value, int> {
if (++*count >= num_pings) { if (++*count >= num_pings) {
CAF_MESSAGE("received " << num_pings CAF_MESSAGE("received " << num_pings
<< " pings, call self->quit"); << " pings, call self->quit");
self->quit(); self->quit();
}
return std::make_tuple(ping_atom::value, value + 1);
} }
return std::make_tuple(ping_atom::value, value + 1); );
},
others >> [=] {
CAF_ERROR("Unexpected message");
});
},
others >> [=] {
CAF_ERROR("Unexpected message");
} }
); );
} }
...@@ -81,16 +76,10 @@ void pong(event_based_actor* self) { ...@@ -81,16 +76,10 @@ void pong(event_based_actor* self) {
[=](const down_msg& dm) { [=](const down_msg& dm) {
CAF_MESSAGE("received down_msg{" << to_string(dm.reason) << "}"); CAF_MESSAGE("received down_msg{" << to_string(dm.reason) << "}");
self->quit(dm.reason); self->quit(dm.reason);
},
others >> [=] {
CAF_ERROR("Unexpected message");
} }
); );
// reply to 'ping' // reply to 'ping'
return std::make_tuple(pong_atom::value, value); return std::make_tuple(pong_atom::value, value);
},
others >> [=] {
CAF_ERROR("Unexpected message");
} }
); );
} }
...@@ -138,9 +127,6 @@ void peer_fun(broker* self, connection_handle hdl, const actor& buddy) { ...@@ -138,9 +127,6 @@ void peer_fun(broker* self, connection_handle hdl, const actor& buddy) {
CAF_MESSAGE("received: " << to_string(dm)); CAF_MESSAGE("received: " << to_string(dm));
if (dm.source == buddy) if (dm.source == buddy)
self->quit(dm.reason); self->quit(dm.reason);
},
others >> [=](const message& msg) {
CAF_MESSAGE("unexpected: " << to_string(msg));
} }
); );
} }
...@@ -155,9 +141,6 @@ behavior peer_acceptor_fun(broker* self, const actor& buddy) { ...@@ -155,9 +141,6 @@ behavior peer_acceptor_fun(broker* self, const actor& buddy) {
}, },
[=](publish_atom) -> uint16_t { [=](publish_atom) -> uint16_t {
return self->add_tcp_doorman(0, "127.0.0.1").second; return self->add_tcp_doorman(0, "127.0.0.1").second;
},
others >> [&] {
CAF_ERROR("Unexpected message");
} }
}; };
} }
......
...@@ -156,9 +156,6 @@ behavior http_worker(http_broker* self, connection_handle hdl) { ...@@ -156,9 +156,6 @@ behavior http_worker(http_broker* self, connection_handle hdl) {
}, },
[=](const connection_closed_msg&) { [=](const connection_closed_msg&) {
self->quit(); self->quit();
},
others >> [=](const message& msg) {
aout(self) << "unexpected message: " << msg << endl;
} }
}; };
} }
...@@ -168,10 +165,7 @@ behavior server(broker* self) { ...@@ -168,10 +165,7 @@ behavior server(broker* self) {
return { return {
[=](const new_connection_msg& ncm) { [=](const new_connection_msg& ncm) {
aout(self) << "fork on new connection" << endl; aout(self) << "fork on new connection" << endl;
auto worker = self->fork(http_worker, ncm.handle); self->fork(http_worker, ncm.handle);
},
others >> [=](const message& msg) {
aout(self) << "Unexpected message: " << msg << endl;
} }
}; };
} }
......
...@@ -49,10 +49,10 @@ struct fixture { ...@@ -49,10 +49,10 @@ struct fixture {
}; };
behavior make_reflector_behavior(event_based_actor* self) { behavior make_reflector_behavior(event_based_actor* self) {
self->set_unexpected_handler(mirror_unexpected_once);
return { return {
others >> [=](const message& msg) { [] {
self->quit(); // nop
return msg;
} }
}; };
} }
...@@ -85,6 +85,7 @@ struct await_reflector_reply_behavior { ...@@ -85,6 +85,7 @@ struct await_reflector_reply_behavior {
// `grp` may be either local or remote // `grp` may be either local or remote
void make_client_behavior(event_based_actor* self, void make_client_behavior(event_based_actor* self,
actor server, group grp) { actor server, group grp) {
self->set_unexpected_handler(skip_unexpected);
self->spawn_in_group(grp, make_reflector_behavior); self->spawn_in_group(grp, make_reflector_behavior);
self->spawn_in_group(grp, make_reflector_behavior); self->spawn_in_group(grp, make_reflector_behavior);
self->request(server, infinite, spawn_atom::value, grp).then( self->request(server, infinite, spawn_atom::value, grp).then(
......
...@@ -36,10 +36,11 @@ using namespace caf; ...@@ -36,10 +36,11 @@ using namespace caf;
namespace { namespace {
behavior mirror() { behavior mirror(event_based_actor* self) {
self->set_unexpected_handler(mirror_unexpected);
return { return {
others >> [=](message& msg) { [] {
return std::move(msg); // nop
} }
}; };
} }
...@@ -47,8 +48,8 @@ behavior mirror() { ...@@ -47,8 +48,8 @@ behavior mirror() {
behavior client(event_based_actor* self, actor serv) { behavior client(event_based_actor* self, actor serv) {
self->send(serv, ok_atom::value); self->send(serv, ok_atom::value);
return { return {
others >> [=] { [] {
CAF_ERROR("Unexpected message"); // nop
} }
}; };
} }
......
...@@ -55,20 +55,15 @@ behavior ping(event_based_actor* self, size_t num_pings) { ...@@ -55,20 +55,15 @@ behavior ping(event_based_actor* self, size_t num_pings) {
CAF_MESSAGE("received `kickoff_atom`"); CAF_MESSAGE("received `kickoff_atom`");
self->send(pong, ping_atom::value, 1); self->send(pong, ping_atom::value, 1);
self->become( self->become(
[=](pong_atom, int value) -> std::tuple<ping_atom, int> { [=](pong_atom, int value) -> std::tuple<ping_atom, int> {
if (++*count >= num_pings) { if (++*count >= num_pings) {
CAF_MESSAGE("received " << num_pings CAF_MESSAGE("received " << num_pings
<< " pings, call self->quit"); << " pings, call self->quit");
self->quit(); self->quit();
}
return std::make_tuple(ping_atom::value, value + 1);
} }
return std::make_tuple(ping_atom::value, value + 1); );
},
others >> [=] {
CAF_ERROR("Unexpected message");
});
},
others >> [=] {
CAF_ERROR("Unexpected message");
} }
}; };
} }
...@@ -87,16 +82,10 @@ behavior pong(event_based_actor* self) { ...@@ -87,16 +82,10 @@ behavior pong(event_based_actor* self) {
[=](const down_msg& dm) { [=](const down_msg& dm) {
CAF_MESSAGE("received down_msg{" << to_string(dm.reason) << "}"); CAF_MESSAGE("received down_msg{" << to_string(dm.reason) << "}");
self->quit(dm.reason); self->quit(dm.reason);
},
others >> [=] {
CAF_ERROR("Unexpected message");
} }
); );
// reply to 'ping' // reply to 'ping'
return std::make_tuple(pong_atom::value, value); return std::make_tuple(pong_atom::value, value);
},
others >> [=] {
CAF_ERROR("Unexpected message");
} }
}; };
} }
......
...@@ -47,8 +47,8 @@ public: ...@@ -47,8 +47,8 @@ public:
behavior make_behavior() override { behavior make_behavior() override {
return { return {
others >> [&] { [] {
CAF_ERROR("Unexpected message"); // nop
} }
}; };
} }
......
Subproject commit 0dd8d434261fd30dfa4ec4b5600c6fd46de8e6f1 Subproject commit 14c46d87a54a6b28955555d7bed13a52b71a8ab3
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