Commit 23b305ef authored by Joseph Noir's avatar Joseph Noir

Merge branch 'unstable' of github.com:Neverlord/libcppa into unstable

parents f99226cf 637d7d9a
......@@ -2,8 +2,8 @@ cmake_minimum_required(VERSION 2.8)
project(cppa CXX)
set(LIBCPPA_VERSION_MAJOR 0)
set(LIBCPPA_VERSION_MINOR 7)
set(LIBCPPA_VERSION_PATCH 2)
set(LIBCPPA_VERSION_MINOR 8)
set(LIBCPPA_VERSION_PATCH 0)
# prohibit in-source builds
if ("${CMAKE_SOURCE_DIR}" STREQUAL "${CMAKE_BINARY_DIR}")
......
Version 0.8
-----------
__2013_10_01__
- Added support for typed actors
- Added `brokers`: actors that encapsulate networking
Version 0.7.2
-------------
......
......@@ -31,7 +31,7 @@ PROJECT_NAME = libcppa
# This could be handy for archiving the generated documentation or
# if some version control system is used.
PROJECT_NUMBER = "Version 0.7.2"
PROJECT_NUMBER = "Version 0.8"
# The OUTPUT_DIRECTORY tag is used to specify the (relative or absolute)
# base path where the generated documentation will be put.
......@@ -565,7 +565,7 @@ WARN_LOGFILE =
# directories like "/usr/src/myproject". Separate the files or directories
# with spaces.
INPUT = @CMAKE_HOME_DIRECTORY@/cppa/ @CMAKE_HOME_DIRECTORY@/cppa/util @CMAKE_HOME_DIRECTORY@/cppa/intrusive @CMAKE_HOME_DIRECTORY@/cppa/network @CMAKE_HOME_DIRECTORY@/cppa/opencl
INPUT = @CMAKE_HOME_DIRECTORY@/cppa/ @CMAKE_HOME_DIRECTORY@/cppa/util @CMAKE_HOME_DIRECTORY@/cppa/intrusive @CMAKE_HOME_DIRECTORY@/cppa/io @CMAKE_HOME_DIRECTORY@/cppa/opencl
# This tag can be used to specify the character encoding of the source files
# that doxygen parses. Internally doxygen uses the UTF-8 encoding, which is
......
......@@ -30,7 +30,8 @@ Usage: $0 [OPTION]... [VAR=VALUE]...
--with-gcc=FILE path to g++ executable
--dual-build build both with gcc and clang
--no-examples build libcppa without examples
--no-qt-examples build libcppa with all but Qt examples
--no-qt-examples build libcppa without Qt examples
--no-protobuf-examples build libcppa without protobuf examples
--no-unit-tests build libcppa without unit tests
--build-static build libcppa as static and shared library
--build-static-only build libcppa as static library only
......@@ -225,6 +226,9 @@ while [ $# -ne 0 ]; do
--no-qt-examples)
append_cache_entry CPPA_NO_QT_EXAMPLES STRING yes
;;
--no-protobuf-examples)
append_cache_entry CPPA_NO_PROTOBUF_EXAMPLES STRING yes
;;
--no-unit-tests)
append_cache_entry CPPA_NO_UNIT_TESTS STRING yes
;;
......
......@@ -87,7 +87,7 @@ class actor_companion_mixin : public Base {
/**
* @brief Defines the message handler.
* @note While the message handler is invoked, @p self will point
* to the companion object to enable send() and reply().
* to the companion object to enable function calls to send().
*/
template<typename... Ts>
void set_message_handler(Ts&&... args) {
......@@ -97,7 +97,7 @@ class actor_companion_mixin : public Base {
/**
* @brief Invokes the message handler with @p msg.
* @note While the message handler is invoked, @p self will point
* to the companion object to enable send() and reply().
* to the companion object to enable function calls to send().
*/
void handle_message(const message_pointer& msg) {
if (!msg) return;
......
......@@ -87,6 +87,13 @@ class any_tuple {
template<typename... Ts>
any_tuple(cow_tuple<Ts...>&& arg) : m_vals(std::move(arg.m_vals)) { }
/**
* @brief Creates a tuple from a set of values.
*/
template<typename T, typename... Ts>
any_tuple(T v0, Ts&&... vs)
: m_vals(make_cow_tuple(std::move(v0), std::forward<Ts>(vs)...).vals()) { }
/**
* @brief Move constructor.
*/
......
......@@ -35,8 +35,16 @@
// if boost.context is not available on your platform
//#define CPPA_DISABLE_CONTEXT_SWITCHING
#if defined(__GNUC__)
#if defined(__clang__)
# define CPPA_CLANG
# define CPPA_DEPRECATED __attribute__((__deprecated__))
#elif defined(__GNUC__)
# define CPPA_GCC
# define CPPA_DEPRECATED __attribute__((__deprecated__))
#elif defined(_MSC_VER)
# define CPPA_DEPRECATED __declspec(deprecated)
#else
# define CPPA_DEPRECATED
#endif
#if defined(__APPLE__)
......
......@@ -276,7 +276,7 @@
* on(atom("compute"), arg_match) >> [](int i0, int i1, int i2)
* {
* // send our result back to the sender of this messages
* reply(atom("result"), i0 + i1 + i2);
* return make_cow_tuple(atom("result"), i0 + i1 + i2);
* }
* );
* @endcode
......@@ -296,10 +296,10 @@
* void math_actor() {
* receive_loop (
* on(atom("plus"), arg_match) >> [](int a, int b) {
* reply(atom("result"), a + b);
* return make_cow_tuple(atom("result"), a + b);
* },
* on(atom("minus"), arg_match) >> [](int a, int b) {
* reply(atom("result"), a - b);
* return make_cow_tuple(atom("result"), a - b);
* }
* );
* }
......@@ -340,7 +340,7 @@
* std::vector<int> vec {1, 2, 3, 4};
* auto i = vec.begin();
* receive_for(i, vec.end()) (
* on(atom("get")) >> [&]() { reply(atom("result"), *i); }
* on(atom("get")) >> [&]() -> any_tuple { return {atom("result"), *i}; }
* );
* @endcode
*
......@@ -654,6 +654,11 @@ inline const actor_ostream& operator<<(const actor_ostream& o, const any_tuple&
return o.write(cppa::to_string(arg));
}
// disambiguate between conversion to string and to any_tuple
inline const actor_ostream& operator<<(const actor_ostream& o, const char* arg) {
return o << std::string{arg};
}
template<typename T>
inline typename std::enable_if<
!std::is_convertible<T, std::string>::value
......
......@@ -41,20 +41,28 @@
#include "cppa/util/duration.hpp"
#include "cppa/util/type_traits.hpp"
namespace cppa {
typedef optional<any_tuple> bhvr_invoke_result;
} // namespace cppa
namespace cppa { namespace detail {
struct optional_any_tuple_visitor {
typedef optional<any_tuple> result_type;
inline result_type operator()() const { return any_tuple{}; }
inline result_type operator()(const none_t&) const { return none; }
inline bhvr_invoke_result operator()() const { return any_tuple{}; }
inline bhvr_invoke_result operator()(const none_t&) const { return none; }
template<typename T>
inline result_type operator()(T& value) const {
inline bhvr_invoke_result operator()(T& value) const {
return make_any_tuple(std::move(value));
}
template<typename... Ts>
inline result_type operator()(cow_tuple<Ts...>& value) const {
inline bhvr_invoke_result operator()(cow_tuple<Ts...>& value) const {
return any_tuple{std::move(value)};
}
inline bhvr_invoke_result operator()(any_tuple& value) const {
return std::move(value);
}
};
template<typename... Ts>
......@@ -72,9 +80,9 @@ class behavior_impl : public ref_counted {
inline behavior_impl(util::duration tout) : m_timeout(tout) { }
virtual optional<any_tuple> invoke(any_tuple&) = 0;
virtual optional<any_tuple> invoke(const any_tuple&) = 0;
inline optional<any_tuple> invoke(any_tuple&& arg) {
virtual bhvr_invoke_result invoke(any_tuple&) = 0;
virtual bhvr_invoke_result invoke(const any_tuple&) = 0;
inline bhvr_invoke_result invoke(any_tuple&& arg) {
any_tuple tmp(std::move(arg));
return invoke(tmp);
}
......@@ -93,12 +101,12 @@ class behavior_impl : public ref_counted {
struct combinator : behavior_impl {
pointer first;
pointer second;
optional<any_tuple> invoke(any_tuple& arg) {
bhvr_invoke_result invoke(any_tuple& arg) {
auto res = first->invoke(arg);
if (!res) return second->invoke(arg);
return res;
}
optional<any_tuple> invoke(const any_tuple& arg) {
bhvr_invoke_result invoke(const any_tuple& arg) {
auto res = first->invoke(arg);
if (!res) return second->invoke(arg);
return res;
......@@ -139,8 +147,6 @@ class default_behavior_impl : public behavior_impl {
public:
typedef optional<any_tuple> invoke_result;
template<typename Expr>
default_behavior_impl(Expr&& expr, const timeout_definition<F>& d)
: super(d.timeout), m_expr(std::forward<Expr>(expr)), m_fun(d.handler) { }
......@@ -149,11 +155,11 @@ class default_behavior_impl : public behavior_impl {
default_behavior_impl(Expr&& expr, util::duration tout, F f)
: super(tout), m_expr(std::forward<Expr>(expr)), m_fun(f) { }
invoke_result invoke(any_tuple& tup) {
bhvr_invoke_result invoke(any_tuple& tup) {
return eval_res(m_expr(tup));
}
invoke_result invoke(const any_tuple& tup) {
bhvr_invoke_result invoke(const any_tuple& tup) {
return eval_res(m_expr(tup));
}
......@@ -170,7 +176,7 @@ class default_behavior_impl : public behavior_impl {
private:
template<typename... Ts>
typename std::enable_if<has_match_hint<Ts...>::value, invoke_result>::type
typename std::enable_if<has_match_hint<Ts...>::value, bhvr_invoke_result>::type
eval_res(optional_variant<Ts...>&& res) {
if (res) {
if (res.template is<match_hint>()) {
......@@ -185,7 +191,7 @@ class default_behavior_impl : public behavior_impl {
}
template<typename... Ts>
typename std::enable_if<!has_match_hint<Ts...>::value, invoke_result>::type
typename std::enable_if<!has_match_hint<Ts...>::value, bhvr_invoke_result>::type
eval_res(optional_variant<Ts...>&& res) {
return apply_visitor(optional_any_tuple_visitor{}, res);
}
......
......@@ -54,7 +54,7 @@ class event_based_actor_impl : public event_based_actor {
void init() { apply(m_init); }
void on_exit() {
void on_exit() override {
typedef typename util::get_callable_trait<CleanupFun>::arg_types arg_types;
std::integral_constant<size_t, util::tl_size<arg_types>::value> token;
on_exit_impl(m_on_exit, token);
......
......@@ -54,17 +54,17 @@ class object_array : public abstract_tuple {
void push_back(const object& what);
virtual void* mutable_at(size_t pos) override;
void* mutable_at(size_t pos) override;
virtual size_t size() const override;
size_t size() const override;
virtual object_array* copy() const override;
object_array* copy() const override;
virtual const void* at(size_t pos) const override;
const void* at(size_t pos) const override;
virtual const uniform_type_info* type_at(size_t pos) const override;
const uniform_type_info* type_at(size_t pos) const override;
virtual const std::string* tuple_type_names() const override;
const std::string* tuple_type_names() const override;
private:
......
......@@ -406,17 +406,24 @@ class receive_policy {
case ordinary_message: {
if (!awaited_response.valid()) {
auto previous_node = hm_begin(client, node, policy);
if (fun(node->msg)) {
// make sure synchronous request
// always receive a response
auto id = node->mid;
auto res = fun(node->msg);
if (res) {
auto mid = node->mid;
auto sender = node->sender;
if (id.valid() && !id.is_answered() && sender) {
CPPA_LOGMF(CPPA_WARNING, client,
"actor did not reply to a "
"synchronous request message");
sender->enqueue({client, sender, id.response_id()},
make_any_tuple(atom("VOID")));
message_header hdr{client, sender, mid.response_id()};
if (res->empty()) {
// make sure synchronous requests
// always receive a response
if (mid.valid() && !mid.is_answered() && sender) {
CPPA_LOGMF(CPPA_WARNING, client,
"actor did not reply to a "
"synchronous request message");
sender->enqueue(std::move(hdr),
make_any_tuple(atom("VOID")));
}
} else {
// respond by using the result of 'fun'
sender->enqueue(std::move(hdr), std::move(*res));
}
hm_cleanup(client, previous_node, policy);
return hm_msg_handled;
......
......@@ -61,9 +61,9 @@ class thread_pool_scheduler : public scheduler {
void enqueue(scheduled_actor* what);
virtual local_actor_ptr exec(spawn_options opts, scheduled_actor_ptr ptr) override;
local_actor_ptr exec(spawn_options opts, scheduled_actor_ptr ptr) override;
virtual local_actor_ptr exec(spawn_options opts, init_callback init_cb, void_function f) override;
local_actor_ptr exec(spawn_options opts, init_callback init_cb, void_function f) override;
private:
......
......@@ -64,7 +64,7 @@ struct type_to_ptype_impl_helper : wrapped_ptype<pt_null> { };
template<typename T>
struct type_to_ptype_impl_helper<true, T> {
static constexpr auto ptype = type_to_ptype_int<
static constexpr primitive_type ptype = type_to_ptype_int<
std::numeric_limits<T>::is_integer,
std::numeric_limits<T>::is_signed,
sizeof(T)
......
......@@ -47,6 +47,7 @@ namespace cppa {
/**
* @brief Base class for all event-based actor implementations.
* @extends scheduled_actor
*/
class event_based_actor : public extend<scheduled_actor>::with<stackless> {
......@@ -63,8 +64,6 @@ class event_based_actor : public extend<scheduled_actor>::with<stackless> {
*/
virtual void init() = 0;
virtual void quit(std::uint32_t reason = exit_reason::normal);
scheduled_actor_type impl_type();
static intrusive_ptr<event_based_actor> from(std::function<void()> fun);
......
......@@ -99,20 +99,18 @@ class broker : public extend<local_actor>::with<threadless, stackless> {
void write(const connection_handle& hdl, util::buffer&& buf);
static broker_ptr from(std::function<void (broker*, connection_handle)> fun,
template<typename F, typename... Ts>
static broker_ptr from(F fun,
input_stream_ptr in,
output_stream_ptr out);
template<typename F, typename T0, typename... Ts>
static broker_ptr from(F fun, input_stream_ptr in, output_stream_ptr out,
T0&& arg0, Ts&&... args) {
return from(std::bind(std::move(fun),
std::placeholders::_1,
std::placeholders::_2,
detail::fwd<T0>(arg0),
detail::fwd<Ts>(args)...),
std::move(in),
std::move(out));
output_stream_ptr out,
Ts&&... args) {
auto hdl = connection_handle::from_int(in->read_handle());
return from_impl(std::bind(std::move(fun),
std::placeholders::_1,
hdl,
detail::fwd<Ts>(args)...),
std::move(in),
std::move(out));
}
static broker_ptr from(std::function<void (broker*)> fun, acceptor_uptr in);
......@@ -126,20 +124,15 @@ class broker : public extend<local_actor>::with<threadless, stackless> {
std::move(in));
}
actor_ptr fork(std::function<void (broker*, connection_handle)> fun,
connection_handle hdl);
template<typename F, typename T0, typename... Ts>
template<typename F, typename... Ts>
actor_ptr fork(F fun,
connection_handle hdl,
T0&& arg0,
Ts&&... args) {
return this->fork(std::bind(std::move(fun),
std::placeholders::_1,
std::placeholders::_2,
detail::fwd<T0>(arg0),
detail::fwd<Ts>(args)...),
hdl);
return this->fork_impl(std::bind(std::move(fun),
std::placeholders::_1,
hdl,
detail::fwd<Ts>(args)...),
hdl);
}
template<typename F>
......@@ -167,6 +160,13 @@ class broker : public extend<local_actor>::with<threadless, stackless> {
private:
actor_ptr fork_impl(std::function<void (broker*)> fun,
connection_handle hdl);
static broker_ptr from_impl(std::function<void (broker*)> fun,
input_stream_ptr in,
output_stream_ptr out);
void invoke_message(const message_header& hdr, any_tuple msg);
void erase_io(int id);
......
......@@ -53,7 +53,7 @@ class buffered_writing : public Base {
: super{std::forward<Ts>(args)...}, m_middleman{mm}, m_out{out}
, m_has_unwritten_data{false} { }
virtual continue_writing_result continue_writing() override {
continue_writing_result continue_writing() override {
CPPA_LOG_TRACE("");
CPPA_LOG_DEBUG_IF(!m_has_unwritten_data, "nothing to write (done)");
while (m_has_unwritten_data) {
......
......@@ -62,6 +62,24 @@ class message_future;
class local_scheduler;
class sync_handle_helper;
#ifdef CPPA_DOCUMENTATION
/**
* @brief Policy tag that causes {@link event_based_actor::become} to
* discard the current behavior.
* @relates local_actor
*/
constexpr auto discard_behavior;
/**
* @brief Policy tag that causes {@link event_based_actor::become} to
* keep the current behavior available.
* @relates local_actor
*/
constexpr auto keep_behavior;
#else
template<bool DiscardBehavior>
struct behavior_policy { static constexpr bool discard_old = DiscardBehavior; };
......@@ -74,20 +92,12 @@ struct is_behavior_policy<behavior_policy<DiscardBehavior>> : std::true_type { }
typedef behavior_policy<false> keep_behavior_t;
typedef behavior_policy<true > discard_behavior_t;
/**
* @brief Policy tag that causes {@link event_based_actor::become} to
* discard the current behavior.
* @relates local_actor
*/
constexpr discard_behavior_t discard_behavior = discard_behavior_t{};
/**
* @brief Policy tag that causes {@link event_based_actor::become} to
* keep the current behavior available.
* @relates local_actor
*/
constexpr keep_behavior_t keep_behavior = keep_behavior_t{};
#endif
/**
* @brief Base class for local running Actors.
* @extends actor
......@@ -119,16 +129,30 @@ class local_actor : public extend<actor>::with<memory_cached> {
void leave(const group_ptr& what);
/**
* @brief Finishes execution of this actor.
* @brief Finishes execution of this actor after any currently running
* message handler is done.
*
* This member function clear the behavior stack of the running actor
* and invokes {@link on_exit()}. The actors does not finish execution
* if the implementation of {@link on_exit()} sets a new behavior.
* When setting a new behavior in {@link on_exit()}, one has to make sure
* to not produce an infinite recursion.
*
* If {@link on_exit()} did not set a new behavior, the actor sends an
* exit message to all of its linked actors, sets its state to @c exited
* and finishes execution.
*
* Causes this actor to send an exit message to all of its
* linked actors, sets its state to @c exited and finishes execution.
* @param reason Exit reason that will be send to
* linked actors and monitors.
* linked actors and monitors. Can be queried using
* {@link planned_exit_reason()}, e.g., from inside
* {@link on_exit()}.
* @note Throws {@link actor_exited} to unwind the stack
* when called in context-switching or thread-based actors.
* @warning This member function throws imeediately in thread-based actors
* that do not use the behavior stack, i.e., actors that use
* blocking API calls such as {@link receive()}.
*/
virtual void quit(std::uint32_t reason = exit_reason::normal) = 0;
virtual void quit(std::uint32_t reason = exit_reason::normal);
/**
* @brief Checks whether this actor traps exit messages.
......@@ -160,8 +184,6 @@ class local_actor : public extend<actor>::with<memory_cached> {
/**
* @brief Returns the sender of the last dequeued message.
* @warning Only set during callback invocation.
* @note Implicitly used by the function {@link cppa::reply}.
* @see cppa::reply()
*/
inline actor_ptr& last_sender();
......@@ -314,6 +336,10 @@ class local_actor : public extend<actor>::with<memory_cached> {
void debug_name(std::string str);
inline std::uint32_t planned_exit_reason() const;
inline void planned_exit_reason(std::uint32_t value);
protected:
inline void remove_handler(message_id id);
......@@ -354,12 +380,16 @@ class local_actor : public extend<actor>::with<memory_cached> {
// allows actors to keep previous behaviors and enables unbecome()
detail::behavior_stack m_bhvr_stack;
// set by quit
std::uint32_t m_planned_exit_reason;
/** @endcond */
private:
std::function<void()> m_sync_failure_handler;
std::function<void()> m_sync_timeout_handler;
};
/**
......@@ -373,7 +403,9 @@ typedef intrusive_ptr<local_actor> local_actor_ptr;
* inline and template member function implementations *
******************************************************************************/
void local_actor::dequeue(behavior&& bhvr) {
/** @cond PRIVATE */
inline void local_actor::dequeue(behavior&& bhvr) {
behavior tmp{std::move(bhvr)};
dequeue(tmp);
}
......@@ -456,6 +488,16 @@ inline void local_actor::remove_handler(message_id id) {
m_bhvr_stack.erase(id);
}
inline std::uint32_t local_actor::planned_exit_reason() const {
return m_planned_exit_reason;
}
inline void local_actor::planned_exit_reason(std::uint32_t value) {
m_planned_exit_reason = value;
}
/** @endcond */
} // namespace cppa
#endif // CPPA_CONTEXT_HPP
......@@ -62,7 +62,7 @@ class mailbox_based : public Base {
template<typename... Ts>
mailbox_based(Ts&&... args) : Base(std::forward<Ts>(args)...) { }
virtual void cleanup(std::uint32_t reason) override {
void cleanup(std::uint32_t reason) override {
detail::sync_request_bouncer f{reason};
m_mailbox.close(f);
Base::cleanup(reason);
......
......@@ -62,7 +62,7 @@ class optional {
/**
* @post <tt>valid() == false</tt>
*/
optional(const none_t&) : m_valid(false) { }
optional(const none_t& = none) : m_valid(false) { }
/**
* @brief Creates an @p option from @p value.
......@@ -141,6 +141,22 @@ class optional {
return m_value;
}
/**
* @brief Returns the value.
*/
inline const T* operator->() const {
CPPA_REQUIRE(valid());
return &m_value;
}
/**
* @brief Returns the value.
*/
inline T* operator->() {
CPPA_REQUIRE(valid());
return &m_value;
}
/**
* @brief Returns the value.
*/
......
......@@ -76,12 +76,15 @@ struct optional_variant_move_helper {
T& lhs;
optional_variant_move_helper(T& lhs_ref) : lhs(lhs_ref) { }
template<typename U>
inline void operator()(const U& rhs) const {
inline void operator()(U& rhs) const {
lhs = std::move(rhs);
}
inline void operator()() const {
lhs = unit;
}
inline void operator()(const none_t&) const {
lhs = none;
}
};
template<typename... Ts>
......@@ -382,6 +385,8 @@ struct optional_variant_cmp_helper {
operator()(const U&) const { return false; }
// variant is void
bool operator()() const { return false; }
// variant is undefined
bool operator()(const none_t&) const { return false; }
};
} // namespace detail
......@@ -435,6 +440,17 @@ std::ostream& operator<<(std::ostream& lhs, const optional_variant<T0, Ts...>& r
return apply_visitor(detail::optional_variant_ostream_helper{lhs}, rhs);
}
template<typename T, typename... Ts>
optional_variant<T, typename util::rm_const_and_ref<Ts>::type...>
make_optional_variant(T value, Ts&&... args) {
return {std::move(value), std::forward<Ts>(args)...};
}
template<typename... Ts>
inline optional_variant<Ts...> make_optional_variant(optional_variant<Ts...> value) {
return std::move(value);
}
} // namespace cppa
#endif // OPTIONAL_VARIANT_HPP
......@@ -44,7 +44,7 @@ class prioritizing : public Base {
public:
virtual mailbox_element* try_pop() override {
mailbox_element* try_pop() override {
auto result = m_high_priority_mailbox.try_pop();
return (result) ? result : this->m_mailbox.try_pop();
}
......@@ -56,18 +56,18 @@ class prioritizing : public Base {
typedef prioritizing combined_type;
virtual void cleanup(std::uint32_t reason) override {
void cleanup(std::uint32_t reason) override {
detail::sync_request_bouncer f{reason};
m_high_priority_mailbox.close(f);
Base::cleanup(reason);
}
virtual bool mailbox_empty() override {
bool mailbox_empty() override {
return m_high_priority_mailbox.empty()
&& this->m_mailbox.empty();
}
virtual void enqueue(const message_header& hdr, any_tuple msg) override {
void enqueue(const message_header& hdr, any_tuple msg) override {
typename Base::mailbox_type* mbox = nullptr;
if (hdr.priority == message_priority::high) {
mbox = &m_high_priority_mailbox;
......
......@@ -67,7 +67,7 @@ void receive_loop(Ts&&... args);
* @code
* int i = 0;
* receive_for(i, 10) (
* on(atom("get")) >> [&]() { reply("result", i); }
* on(atom("get")) >> [&]() -> any_tuple { return {"result", i}; }
* );
* @endcode
* @param begin First value in range.
......
......@@ -61,7 +61,7 @@ class sb_actor : public Base {
* @brief Overrides {@link event_based_actor::init()} and sets
* the initial actor behavior to <tt>Derived::init_state</tt>.
*/
virtual void init() override {
void init() override {
become(util::dptr<Derived>(this)->init_state);
}
......
......@@ -81,6 +81,25 @@ inline void send_tuple(destination_header hdr, any_tuple what) {
else fhdr.deliver(std::move(what));
}
/**
* @brief Sends @p what to the receiver specified in @p hdr.
*/
template<typename... Signatures, typename... Ts>
void send(const typed_actor_ptr<Signatures...>& whom, Ts&&... what) {
static constexpr int input_pos = util::tl_find_if<
util::type_list<Signatures...>,
detail::input_is<util::type_list<
typename detail::implicit_conversions<
typename util::rm_const_and_ref<
Ts
>::type
>::type...
>>::template eval
>::value;
static_assert(input_pos >= 0, "typed actor does not support given input");
send(whom.unbox(), std::forward<Ts>(what)...);
}
/**
* @brief Sends <tt>{what...}</tt> to the receiver specified in @p hdr.
* @pre <tt>sizeof...(Ts) > 0</tt>
......@@ -287,7 +306,7 @@ message_future timed_sync_send(actor_ptr whom,
* @brief Sends a message to the sender of the last received message.
* @param what Message content as a tuple.
*/
inline void reply_tuple(any_tuple what) {
CPPA_DEPRECATED inline void reply_tuple(any_tuple what) {
self->reply_message(std::move(what));
}
......@@ -296,7 +315,7 @@ inline void reply_tuple(any_tuple what) {
* @param what Message elements.
*/
template<typename... Ts>
inline void reply(Ts&&... what) {
CPPA_DEPRECATED inline void reply(Ts&&... what) {
self->reply_message(make_any_tuple(std::forward<Ts>(what)...));
}
......
......@@ -55,19 +55,20 @@ enum class spawn_options : int {
};
#endif
#ifndef CPPA_DOCUMENTATION
namespace {
#endif
/**
* @brief Concatenates two {@link spawn_options}.
* @relates spawn_options
*/
constexpr spawn_options operator+(const spawn_options& lhs,
const spawn_options& rhs) {
return static_cast<spawn_options>(static_cast<int>(lhs) | static_cast<int>(rhs));
return static_cast<spawn_options>( static_cast<int>(lhs)
| static_cast<int>(rhs));
}
#ifndef CPPA_DOCUMENTATION
namespace {
#endif
/**
* @brief Denotes default settings.
*/
......
......@@ -64,43 +64,58 @@ class stacked : public Base {
static constexpr auto receive_flag = detail::rp_nestable;
virtual void dequeue(behavior& bhvr) override {
void dequeue(behavior& bhvr) override {
++m_nested_count;
m_recv_policy.receive(util::dptr<Subtype>(this), bhvr);
--m_nested_count;
check_quit();
}
virtual void dequeue_response(behavior& bhvr, message_id request_id) override {
void dequeue_response(behavior& bhvr, message_id request_id) override {
++m_nested_count;
m_recv_policy.receive(util::dptr<Subtype>(this), bhvr, request_id);
--m_nested_count;
check_quit();
}
virtual void run() {
auto dthis = util::dptr<Subtype>(this);
if (!dthis->m_bhvr_stack.empty()) dthis->exec_behavior_stack();
exec_behavior_stack();
if (m_behavior) m_behavior();
auto dthis = util::dptr<Subtype>(this);
auto rsn = dthis->planned_exit_reason();
dthis->cleanup(rsn == exit_reason::not_exited ? exit_reason::normal : rsn);
}
inline void set_behavior(std::function<void()> fun) {
m_behavior = std::move(fun);
}
virtual void quit(std::uint32_t reason) override {
this->cleanup(reason);
throw actor_exited(reason);
}
virtual void exec_behavior_stack() override {
this->m_bhvr_stack.exec(m_recv_policy, util::dptr<Subtype>(this));
void exec_behavior_stack() override {
m_in_bhvr_loop = true;
auto dthis = util::dptr<Subtype>(this);
if (!stack_empty() && !m_nested_count) {
while (!stack_empty()) {
++m_nested_count;
dthis->m_bhvr_stack.exec(m_recv_policy, dthis);
--m_nested_count;
check_quit();
}
}
m_in_bhvr_loop = false;
}
protected:
template<typename... Ts>
stacked(Ts&&... args) : Base(std::forward<Ts>(args)...) { }
stacked(Ts&&... args) : Base(std::forward<Ts>(args)...)
, m_in_bhvr_loop(false)
, m_nested_count(0) { }
virtual void do_become(behavior&& bhvr, bool discard_old) override {
void do_become(behavior&& bhvr, bool discard_old) override {
become_impl(std::move(bhvr), discard_old, message_id());
}
virtual void become_waiting_for(behavior bhvr, message_id mid) override {
void become_waiting_for(behavior bhvr, message_id mid) override {
become_impl(std::move(bhvr), false, mid);
}
......@@ -114,6 +129,10 @@ class stacked : public Base {
private:
inline bool stack_empty() {
return util::dptr<Subtype>(this)->m_bhvr_stack.empty();
}
void become_impl(behavior&& bhvr, bool discard_old, message_id mid) {
auto dthis = util::dptr<Subtype>(this);
if (bhvr.timeout().valid()) {
......@@ -126,6 +145,27 @@ class stacked : public Base {
dthis->m_bhvr_stack.push_back(std::move(bhvr), mid);
}
void check_quit() {
// do nothing if other message handlers are still running
if (m_nested_count > 0) return;
auto dthis = util::dptr<Subtype>(this);
auto rsn = dthis->planned_exit_reason();
if (rsn != exit_reason::not_exited) {
dthis->on_exit();
if (stack_empty()) {
dthis->cleanup(rsn);
throw actor_exited(rsn);
}
else {
dthis->planned_exit_reason(exit_reason::not_exited);
if (!m_in_bhvr_loop) exec_behavior_stack();
}
}
}
bool m_in_bhvr_loop;
size_t m_nested_count;
};
} // namespace cppa
......
......@@ -76,7 +76,7 @@ class thread_mapped_actor : public extend<local_actor, thread_mapped_actor>::
inline void initialized(bool value) { m_initialized = value; }
virtual bool initialized() const override;
bool initialized() const override;
private:
......
......@@ -89,15 +89,23 @@ class threaded : public Base {
void run_detached() {
auto dthis = util::dptr<Subtype>(this);
dthis->init();
dthis->m_bhvr_stack.exec(dthis->m_recv_policy, dthis);
dthis->on_exit();
if (dthis->planned_exit_reason() != exit_reason::not_exited) {
// init() did indeed call quit() for some reason
dthis->on_exit();
}
while (!dthis->m_bhvr_stack.empty()) {
dthis->m_bhvr_stack.exec(dthis->m_recv_policy, dthis);
dthis->on_exit();
}
auto rsn = dthis->planned_exit_reason();
dthis->cleanup(rsn == exit_reason::not_exited ? exit_reason::normal : rsn);
}
inline void initialized(bool value) {
m_initialized = value;
}
virtual bool initialized() const override {
bool initialized() const override {
return m_initialized;
}
......@@ -125,10 +133,15 @@ class threaded : public Base {
}
}
virtual void enqueue(const message_header& hdr, any_tuple msg) override {
void enqueue(const message_header& hdr, any_tuple msg) override {
enqueue_impl(this->m_mailbox, hdr, std::move(msg));
}
bool chained_enqueue(const message_header& hdr, any_tuple msg) override {
enqueue(hdr, std::move(msg));
return false;
}
timeout_type init_timeout(const util::duration& rel_time) {
auto result = std::chrono::high_resolution_clock::now();
result += rel_time;
......
......@@ -41,6 +41,10 @@ namespace cppa {
struct typed_actor_result_visitor {
typed_actor_result_visitor() : m_hdl(self->make_response_handle()) { }
typed_actor_result_visitor(response_handle hdl) : m_hdl(hdl) { }
inline void operator()(const none_t&) const {
CPPA_LOG_ERROR("a typed actor received a "
"non-matching message: "
......@@ -51,29 +55,27 @@ struct typed_actor_result_visitor {
template<typename T>
inline void operator()(T& value) const {
reply(std::move(value));
reply_to(m_hdl, std::move(value));
}
template<typename... Ts>
inline void operator()(cow_tuple<Ts...>& value) const {
reply_tuple(std::move(value));
reply_tuple_to(m_hdl, std::move(value));
}
template<typename R>
inline void operator()(typed_continue_helper<R>& ch) const {
auto hdl = self->make_response_handle();
auto hdl = m_hdl;
ch.continue_with([=](R value) {
reply_to(hdl, std::move(value));
typed_actor_result_visitor tarv{hdl};
auto ov = make_optional_variant(std::move(value));
apply_visitor(tarv, ov);
});
}
template<typename... Rs>
inline void operator()(typed_continue_helper<cow_tuple<Rs...>>& ch) const {
auto hdl = self->make_response_handle();
ch.continue_with([=](cow_tuple<Rs...> value) {
reply_tuple_to(hdl, std::move(value));
});
}
private:
response_handle m_hdl;
};
......@@ -95,7 +97,7 @@ class typed_actor : public event_based_actor {
});
}
virtual void do_become(behavior&&, bool) override {
void do_become(behavior&&, bool) override {
CPPA_LOG_ERROR("typed actors are not allowed to call become()");
quit(exit_reason::unallowed_function_call);
}
......
......@@ -49,6 +49,9 @@ spawn_typed(const match_expr<Ts...>&);
template<typename... Ts>
void send_exit(const typed_actor_ptr<Ts...>&, std::uint32_t);
template<typename... Signatures, typename... Ts>
void send(const typed_actor_ptr<Signatures...>&, Ts&&...);
template<typename... Ts, typename... Us>
typed_message_future<
typename detail::deduce_output_type<
......@@ -77,6 +80,10 @@ class typed_actor_ptr {
template<typename... Ts>
friend void send_exit(const typed_actor_ptr<Ts...>&, std::uint32_t);
template<typename... Sigs, typename... Ts>
friend void send(const typed_actor_ptr<Sigs...>&, Ts&&...);
template<typename... Ts, typename... Us>
friend typed_message_future<
typename detail::deduce_output_type<
......@@ -114,7 +121,9 @@ class typed_actor_ptr {
private:
/** @cond PRIVATE */
const actor_ptr& unbox() const { return m_ptr; }
/** @endcond */
typed_actor_ptr(actor_ptr ptr) : m_ptr(std::move(ptr)) { }
......
......@@ -33,6 +33,7 @@
#include <string>
#include <vector>
#include <sstream>
#include <algorithm>
#include <type_traits>
......@@ -47,12 +48,14 @@ std::vector<std::string> split(const std::string& str,
template<typename Iterator>
typename std::enable_if<is_forward_iterator<Iterator>::value,std::string>::type
join(Iterator begin, Iterator end, const std::string& glue = "") {
std::string result;
std::for_each(begin, end, [&](const std::string& str) {
if (!result.empty()) result += glue;
result += str;
});
return result;
bool first = true;
std::ostringstream oss;
for ( ; begin != end; ++begin) {
if (first) first = false;
else oss << glue;
oss << *begin;
}
return oss.str();
}
template<typename Container>
......
......@@ -4,9 +4,9 @@ project(cppa_examples CXX)
add_custom_target(all_examples)
macro(add name folder)
add_executable(${name} ${folder}/${name}.cpp ${ARGN})
target_link_libraries(${name} ${CMAKE_DL_LIBS} ${CPPA_LIBRARY} ${PTHREAD_LIBRARIES})
add_dependencies(${name} all_examples)
add_executable(${name} ${folder}/${name}.cpp ${ARGN})
target_link_libraries(${name} ${CMAKE_DL_LIBS} ${CPPA_LIBRARY} ${PTHREAD_LIBRARIES})
add_dependencies(${name} all_examples)
endmacro()
add(announce_1 type_system)
......@@ -22,26 +22,26 @@ add(distributed_calculator remote_actors)
add(group_server remote_actors)
add(group_chat remote_actors)
find_package(Protobuf)
if (PROTOBUF_FOUND)
if (NOT "${CPPA_NO_PROTOBUF_EXAMPLES}" STREQUAL "yes")
find_package(Protobuf)
if (PROTOBUF_FOUND)
PROTOBUF_GENERATE_CPP(ProtoSources ProtoHeaders "${CMAKE_CURRENT_SOURCE_DIR}/remote_actors/pingpong.proto")
include_directories(${PROTOBUF_INCLUDE_DIR})
add_executable(protobuf_broker remote_actors/protobuf_broker.cpp ${ProtoSources})
target_link_libraries(protobuf_broker ${CMAKE_DL_LIBS} ${CPPA_LIBRARY} ${PTHREAD_LIBRARIES} ${PROTOBUF_LIBRARIES})
add_dependencies(protobuf_broker all_examples)
endif (PROTOBUF_FOUND)
endif (PROTOBUF_FOUND)
endif ()
find_package(CURL)
if (CURL_FOUND)
add_executable(curl_fuse curl/curl_fuse.cpp)
target_link_libraries(curl_fuse ${CMAKE_DL_LIBS} ${CPPA_LIBRARY} ${PTHREAD_LIBRARIES} ${CURL_LIBRARY})
add_dependencies(curl_fuse all_examples)
add_executable(curl_fuse curl/curl_fuse.cpp)
target_link_libraries(curl_fuse ${CMAKE_DL_LIBS} ${CPPA_LIBRARY} ${PTHREAD_LIBRARIES} ${CURL_LIBRARY})
add_dependencies(curl_fuse all_examples)
endif (CURL_FOUND)
if (NOT "${CPPA_NO_QT_EXAMPLES}" STREQUAL "yes")
find_package(Qt4)
if (QT4_FOUND)
include(${QT_USE_FILE})
QT4_ADD_RESOURCES(GROUP_CHAT_RCS )
......@@ -61,7 +61,7 @@ if (NOT "${CPPA_NO_QT_EXAMPLES}" STREQUAL "yes")
${QT_LIBRARIES})
add_dependencies(qt_group_chat all_examples)
endif (QT4_FOUND)
endif()
endif ()
if (ENABLE_OPENCL)
macro(add_opencl name folder)
......@@ -69,7 +69,6 @@ if (ENABLE_OPENCL)
target_link_libraries(${name} ${CMAKE_DL_LIBS} ${CPPA_LIBRARY} ${PTHREAD_LIBRARIES} ${OPENCL_LIBRARIES})
add_dependencies(${name} all_examples)
endmacro()
add_opencl(simple_matrix opencl)
add_opencl(proper_matrix opencl)
endif (ENABLE_OPENCL)
......@@ -129,7 +129,7 @@ class base_actor : public event_based_actor {
return ::print(m_color.c_str(), m_name.c_str());
}
virtual void on_exit() override {
void on_exit() override {
print() << "on_exit" << color::reset_endl;
}
......@@ -152,7 +152,7 @@ class client_job : public base_actor {
protected:
virtual void init() override {
void init() override {
print() << "init" << color::reset_endl;
send(m_parent,
atom("read"),
......@@ -186,7 +186,7 @@ class client : public base_actor {
protected:
virtual void init() override {
void init() override {
using std::chrono::milliseconds;
link_to(m_parent);
print() << "init" << color::reset_endl;
......@@ -224,21 +224,20 @@ class curl_worker : public base_actor {
protected:
virtual void init() override {
void init() override {
print() << "init" << color::reset_endl;
m_curl = curl_easy_init();
curl_easy_setopt(m_curl, CURLOPT_WRITEFUNCTION, &curl_worker::cb);
curl_easy_setopt(m_curl, CURLOPT_NOSIGNAL, 1);
become (
on(atom("read"), arg_match) >> [=](const std::string& file_name,
uint64_t offset,
uint64_t range) {
on(atom("read"), arg_match)
>> [=](const std::string& fname, uint64_t offset, uint64_t range)
-> cow_tuple<atom_value, util::buffer> {
print() << "read" << color::reset_endl;
bool request_done = false;
while (!request_done) {
for (;;) {
m_buf.clear();
// set URL
curl_easy_setopt(m_curl, CURLOPT_URL, file_name.c_str());
curl_easy_setopt(m_curl, CURLOPT_URL, fname.c_str());
// set range
std::ostringstream oss;
oss << offset << "-" << range;
......@@ -270,11 +269,9 @@ class curl_worker : public base_actor {
<< " bytes with 'HTTP RETURN CODE': "
<< hc
<< color::reset_endl;
reply(atom("reply"), m_buf);
// tell parent that this worker is done
send(m_parent, atom("finished"));
request_done = true;
break;
return make_cow_tuple(atom("reply"), m_buf);
case 404: // file does not exist
print() << "http error: download failed with "
"'HTTP RETURN CODE': 404 (file does "
......@@ -282,14 +279,14 @@ class curl_worker : public base_actor {
<< color::reset_endl;
}
}
// avoid 100% cpu utilization if remote side is not accessible
usleep(100000); // 100000 == 100ms
}
// avoid 100% cpu utilization if remote side is not accessible
if (!request_done) usleep(100000); // 100000 == 100ms
}
);
}
virtual void on_exit() override {
void on_exit() override {
curl_easy_cleanup(m_curl);
print() << "on_exit" << color::reset_endl;
}
......@@ -317,7 +314,7 @@ class curl_master : public base_actor {
protected:
virtual void init() override {
void init() override {
print() << "init" << color::reset_endl;
// spawn workers
for(size_t i = 0; i < num_curl_workers; ++i) {
......
......@@ -9,13 +9,13 @@ void mirror() {
// wait for messages
become (
// invoke this lambda expression if we receive a string
on_arg_match >> [](const string& what) {
on_arg_match >> [](const string& what) -> string {
// prints "Hello World!" via aout (thread-safe cout wrapper)
aout << what << endl;
// replies "!dlroW olleH"
reply(string(what.rbegin(), what.rend()));
// terminates this actor ('become' otherwise loops forever)
self->quit();
// reply "!dlroW olleH"
return string(what.rbegin(), what.rend());
}
);
}
......
......@@ -21,10 +21,10 @@ void blocking_math_fun() {
// - on<atom("plus"), int, int>()
// - on(atom("plus"), val<int>, val<int>)
on(atom("plus"), arg_match) >> [](int a, int b) {
reply(atom("result"), a + b);
return make_cow_tuple(atom("result"), a + b);
},
on(atom("minus"), arg_match) >> [](int a, int b) {
reply(atom("result"), a - b);
return make_cow_tuple(atom("result"), a - b);
},
on(atom("quit")) >> [&]() {
// note: this actor uses the blocking API, hence self->quit()
......@@ -38,10 +38,10 @@ void calculator() {
// execute this behavior until actor terminates
become (
on(atom("plus"), arg_match) >> [](int a, int b) {
reply(atom("result"), a + b);
return make_cow_tuple(atom("result"), a + b);
},
on(atom("minus"), arg_match) >> [](int a, int b) {
reply(atom("result"), a - b);
return make_cow_tuple(atom("result"), a - b);
},
on(atom("quit")) >> [] {
// terminate actor with normal exit reason
......
......@@ -28,11 +28,11 @@ using namespace cppa::placeholders;
// our "service"
void calculator() {
become (
on(atom("plus"), arg_match) >> [](int a, int b) {
reply(atom("result"), a + b);
on(atom("plus"), arg_match) >> [](int a, int b) -> any_tuple {
return {atom("result"), a + b};
},
on(atom("minus"), arg_match) >> [](int a, int b) {
reply(atom("result"), a - b);
on(atom("minus"), arg_match) >> [](int a, int b) -> any_tuple {
return {atom("result"), a - b};
},
on(atom("quit")) >> [=]() {
self->quit();
......
......@@ -61,10 +61,10 @@ void ping(size_t num_pings) {
on(atom("kickoff"), arg_match) >> [=](const actor_ptr& pong) {
send(pong, atom("ping"), 1);
become (
on(atom("pong"), arg_match) >> [=](int value) {
on(atom("pong"), arg_match) >> [=](int value) -> any_tuple {
aout << "pong: " << value << endl;
if (++*count >= num_pings) self->quit();
else reply(atom("ping"), value + 1);
return {atom("ping"), value + 1};
}
);
}
......
......@@ -17,9 +17,7 @@ class foo {
public:
foo() : m_a(0), m_b(0) { }
foo(int a0, int b0) : m_a(a0), m_b(b0) { }
foo(int a0 = 0, int b0 = 0) : m_a(a0), m_b(b0) { }
foo(const foo&) = default;
......@@ -64,4 +62,3 @@ int main(int, char**) {
shutdown();
return 0;
}
......@@ -202,6 +202,4 @@ void actor::cleanup(std::uint32_t reason) {
}
}
} // namespace cppa
......@@ -47,17 +47,17 @@ class continuation_decorator : public detail::behavior_impl {
}
template<typename T>
inline optional<any_tuple> invoke_impl(T& tup) {
inline bhvr_invoke_result invoke_impl(T& tup) {
auto res = m_decorated->invoke(tup);
if (res) return m_fun(*res);
return none;
}
optional<any_tuple> invoke(any_tuple& tup) {
bhvr_invoke_result invoke(any_tuple& tup) {
return invoke_impl(tup);
}
optional<any_tuple> invoke(const any_tuple& tup) {
bhvr_invoke_result invoke(const any_tuple& tup) {
return invoke_impl(tup);
}
......
......@@ -63,21 +63,10 @@ class default_broker : public broker {
typedef std::function<void (broker*)> function_type;
struct fork_flag { };
template<typename... Ts>
default_broker(function_type&& fun, Ts&&... args)
: broker{std::forward<Ts>(args)...}, m_fun{move(fun)} { }
template<typename... Ts>
default_broker(fork_flag,
std::function<void (broker*, connection_handle)>&& fun,
connection_handle hdl,
Ts&&... args)
: broker{std::forward<Ts>(args)...}
, m_fun{std::bind(move(fun), std::placeholders::_1, hdl)} { }
void init() override {
enqueue(nullptr, make_any_tuple(atom("INITMSG")));
become(
......@@ -265,7 +254,6 @@ class broker::doorman : public broker::servant {
}
continue_reading_result continue_reading() override {
CPPA_REQUIRE(parent() != nullptr);
CPPA_LOG_TRACE("");
for (;;) {
optional<stream_ptr_pair> opt{none};
......@@ -437,15 +425,10 @@ local_actor_ptr init_and_launch(broker_ptr ptr) {
return move(ptr);
}
broker_ptr broker::from(std::function<void (broker*, connection_handle)> fun,
input_stream_ptr in,
output_stream_ptr out) {
auto hdl = connection_handle::from_int(in->read_handle());
return make_counted<default_broker>(std::bind(move(fun),
std::placeholders::_1,
hdl),
move(in),
move(out));
broker_ptr broker::from_impl(std::function<void (broker*)> fun,
input_stream_ptr in,
output_stream_ptr out) {
return make_counted<default_broker>(move(fun), move(in), move(out));
}
broker_ptr broker::from(std::function<void (broker*)> fun, acceptor_uptr in) {
......@@ -475,15 +458,12 @@ accept_handle broker::add_doorman(acceptor_uptr ptr) {
return id;
}
actor_ptr broker::fork(std::function<void (broker*, connection_handle)> fun,
connection_handle hdl) {
actor_ptr broker::fork_impl(std::function<void (broker*)> fun,
connection_handle hdl) {
auto i = m_io.find(hdl);
if (i == m_io.end()) throw std::invalid_argument("invalid handle");
scribe* sptr = i->second.get(); // non-owning pointer
auto result = make_counted<default_broker>(default_broker::fork_flag{},
move(fun),
hdl,
move(i->second));
auto result = make_counted<default_broker>(move(fun), move(i->second));
init_and_launch(result);
sptr->set_parent(result); // set new parent
m_io.erase(i);
......
......@@ -33,7 +33,7 @@
#include "cppa/config.hpp"
#include "cppa/detail/demangle.hpp"
#ifdef CPPA_GCC
#if defined(CPPA_GCC) || defined(CPPA_CLANG)
#include <cxxabi.h>
#endif
......@@ -79,7 +79,7 @@ std::string demangle(const char* decorated) {
}
}
free(undecorated);
# ifdef __clang__
# ifdef CPPA_CLANG
// replace "std::__1::" with "std::" (fixes strange clang names)
std::string needle = "std::__1::";
std::string fixed_string = "std::";
......
......@@ -59,13 +59,17 @@ class default_scheduled_actor : public event_based_actor {
m_initialized = true;
m_fun();
if (m_bhvr_stack.empty()) {
if (exit_reason() == exit_reason::not_exited) quit(exit_reason::normal);
set_state(actor_state::done);
m_bhvr_stack.clear();
m_bhvr_stack.cleanup();
on_exit();
next.swap(m_chained_actor);
set_state(actor_state::done);
if (exit_reason() == exit_reason::not_exited) {
quit(exit_reason::normal);
}
else {
set_state(actor_state::done);
m_bhvr_stack.clear();
m_bhvr_stack.cleanup();
on_exit();
next.swap(m_chained_actor);
set_state(actor_state::done);
}
return resume_result::actor_done;
}
}
......@@ -94,15 +98,26 @@ resume_result event_based_actor::resume(util::fiber*, actor_ptr& next_job) {
CPPA_REQUIRE( state() == actor_state::ready
|| state() == actor_state::pending);
scoped_self_setter sss{this};
auto done_cb = [&]() {
auto done_cb = [&]() -> bool {
CPPA_LOG_TRACE("");
if (exit_reason() == exit_reason::not_exited) quit(exit_reason::normal);
if (exit_reason() == exit_reason::not_exited) {
if (planned_exit_reason() == exit_reason::not_exited) {
planned_exit_reason(exit_reason::normal);
}
on_exit();
if (!m_bhvr_stack.empty()) {
planned_exit_reason(exit_reason::not_exited);
return false; // on_exit did set a new behavior
}
cleanup(planned_exit_reason());
}
set_state(actor_state::done);
m_bhvr_stack.clear();
m_bhvr_stack.cleanup();
on_exit();
CPPA_REQUIRE(next_job == nullptr);
next_job.swap(m_chained_actor);
return true;
};
CPPA_REQUIRE(next_job == nullptr);
try {
......@@ -150,9 +165,8 @@ resume_result event_based_actor::resume(util::fiber*, actor_ptr& next_job) {
"set actor with ID "
<< m_chained_actor->id()
<< " as successor");
if (m_bhvr_stack.empty()) {
if (m_bhvr_stack.empty() && done_cb()) {
CPPA_LOGMF(CPPA_DEBUG, self, "behavior stack empty");
done_cb();
return resume_result::actor_done;
}
m_bhvr_stack.cleanup();
......@@ -185,27 +199,6 @@ resume_result event_based_actor::resume(util::fiber*, actor_ptr& next_job) {
return resume_result::actor_done;
}
void event_based_actor::quit(std::uint32_t reason) {
CPPA_LOG_TRACE("reason = " << reason
<< ", class " << detail::demangle(typeid(*this)));
cleanup(reason);
m_bhvr_stack.clear();
if (reason == exit_reason::unallowed_function_call) {
CPPA_LOG_WARNING("actor tried to use a blocking function");
// when using receive(), the non-blocking nature of event-based
// actors breaks any assumption the user has about his code,
// in particular, receive_loop() is a deadlock when not throwing
// an exception here
aout << "*** warning: event-based actor killed because it tried to "
"use receive()\n";
throw actor_exited(reason);
}
}
scheduled_actor_type event_based_actor::impl_type() {
return event_based_impl;
}
......
......@@ -91,7 +91,7 @@ struct group_nameserver : event_based_actor {
void init() {
become (
on(atom("GET_GROUP"), arg_match) >> [](const std::string& name) {
reply(atom("GROUP"), group::get("local", name));
return make_cow_tuple(atom("GROUP"), group::get("local", name));
},
on(atom("SHUTDOWN")) >> [=] {
quit();
......
......@@ -76,7 +76,8 @@ constexpr const char* s_default_debug_name = "actor";
local_actor::local_actor(bool sflag)
: m_chaining(sflag), m_trap_exit(false)
, m_is_scheduled(sflag), m_dummy_node(), m_current_node(&m_dummy_node) {
, m_is_scheduled(sflag), m_dummy_node(), m_current_node(&m_dummy_node)
, m_planned_exit_reason(exit_reason::not_exited) {
# ifdef CPPA_DEBUG_MODE
new (&m_debug_name) std::string (std::to_string(m_id) + "@local");
# endif // CPPA_DEBUG_MODE
......@@ -189,4 +190,24 @@ void local_actor::dequeue_response(behavior&, message_id) {
quit(exit_reason::unallowed_function_call);
}
void local_actor::quit(std::uint32_t reason) {
CPPA_LOG_TRACE("reason = " << reason
<< ", class " << detail::demangle(typeid(*this)));
if (reason == exit_reason::unallowed_function_call) {
// this is the only reason that causes an exception
cleanup(reason);
m_bhvr_stack.clear();
CPPA_LOG_WARNING("actor tried to use a blocking function");
// when using receive(), the non-blocking nature of event-based
// actors breaks any assumption the user has about his code,
// in particular, receive_loop() is a deadlock when not throwing
// an exception here
aout << "*** warning: event-based actor killed because it tried to "
"use receive()\n";
throw actor_exited(reason);
}
m_bhvr_stack.clear();
planned_exit_reason(reason);
}
} // namespace cppa
......@@ -597,7 +597,7 @@ class default_meta_tuple : public uniform_type_info {
}
any_tuple as_any_tuple(void* ptr) const override {
return any_tuple{cast(ptr)};
return any_tuple{static_cast<any_tuple::raw_ptr>(cast(ptr))};
}
const char* name() const override {
......
......@@ -17,7 +17,7 @@ size_t s_pongs = 0;
behavior ping_behavior(size_t num_pings) {
return (
on(atom("pong"), arg_match) >> [num_pings](int value) {
on(atom("pong"), arg_match) >> [num_pings](int value) -> any_tuple {
CPPA_LOGF_ERROR_IF(!self->last_sender(), "last_sender() == nullptr");
CPPA_LOGF_INFO("received {'pong', " << value << "}");
//cout << to_string(self->last_dequeued()) << endl;
......@@ -27,7 +27,7 @@ behavior ping_behavior(size_t num_pings) {
send_exit(self->last_sender(), exit_reason::user_defined);
self->quit();
}
else reply(atom("ping"), value);
return {atom("ping"), value};
},
others() >> [] {
CPPA_LOGF_ERROR("unexpected message; "
......@@ -39,9 +39,9 @@ behavior ping_behavior(size_t num_pings) {
behavior pong_behavior() {
return (
on(atom("ping"), arg_match) >> [](int value) {
on(atom("ping"), arg_match) >> [](int value) -> any_tuple {
CPPA_LOGF_INFO("received {'ping', " << value << "}");
reply(atom("pong"), value + 1);
return {atom("pong"), value + 1};
},
others() >> [] {
CPPA_LOGF_ERROR("unexpected message; "
......
......@@ -29,7 +29,7 @@
#include <memory>
#include<iostream>
#include <iostream>
#include "test.hpp"
#include "cppa/cppa.hpp"
......@@ -39,44 +39,43 @@ using namespace cppa;
namespace { constexpr size_t message_size = sizeof(atom_value) + sizeof(int); }
void report_unexpected() {
std::cerr << "unexpected message: " << to_string(self->last_dequeued())
<< std::endl;
}
void ping(size_t num_pings) {
auto count = std::make_shared<size_t>(0);
become (
on(atom("kickoff"), arg_match) >> [=](const actor_ptr& pong) {
send(pong, atom("ping"), 1);
become (
on(atom("pong"), arg_match) >> [=](int value) {
on(atom("pong"), arg_match)
>> [=](int value) -> cow_tuple<atom_value, int> {
if (++*count >= num_pings) self->quit();
else reply(atom("ping"), value + 1);
return make_cow_tuple(atom("ping"), value + 1);
},
others() >> report_unexpected
others() >> CPPA_UNEXPECTED_MSG_CB()
);
},
others() >> report_unexpected
others() >> CPPA_UNEXPECTED_MSG_CB()
);
}
void pong() {
become (
on(atom("ping"), arg_match) >> [](int value) {
on(atom("ping"), arg_match)
>> [](int value) -> cow_tuple<atom_value, int> {
self->monitor(self->last_sender());
reply(atom("pong"), value);
// set next behavior
become (
on(atom("ping"), arg_match) >> [](int value) {
reply(atom("pong"), value);
return make_cow_tuple(atom("pong"), value);
},
on(atom("DOWN"), arg_match) >> [=](uint32_t reason) {
self->quit(reason);
},
others() >> report_unexpected
others() >> CPPA_UNEXPECTED_MSG_CB()
);
// reply to 'ping'
return {atom("pong"), value};
},
others() >> report_unexpected
others() >> CPPA_UNEXPECTED_MSG_CB()
);
}
......@@ -86,14 +85,9 @@ void peer(io::broker* thisptr, io::connection_handle hdl, const actor_ptr& buddy
cout << "num_connections() != 1" << endl;
throw std::logic_error("num_connections() != 1");
}
//thisptr->for_each_connection([=](io::connection_handle hdl) {
// thisptr->receive_policy(hdl, io::broker::exactly, message_size);
//});
auto write = [=](atom_value type, int value) {
//thisptr->for_each_connection([=](io::connection_handle hdl) {
thisptr->write(hdl, sizeof(type), &type);
thisptr->write(hdl, sizeof(value), &value);
//});
thisptr->write(hdl, sizeof(type), &type);
thisptr->write(hdl, sizeof(value), &value);
};
become (
on(atom("IO_closed"), arg_match) >> [=](io::connection_handle) {
......@@ -115,7 +109,7 @@ void peer(io::broker* thisptr, io::connection_handle hdl, const actor_ptr& buddy
on(atom("DOWN"), arg_match) >> [=](uint32_t reason) {
if (thisptr->last_sender() == buddy) self->quit(reason);
},
others() >> report_unexpected
others() >> CPPA_UNEXPECTED_MSG_CB()
);
}
......@@ -127,7 +121,7 @@ void peer_acceptor(io::broker* thisptr, const actor_ptr& buddy) {
thisptr->fork(peer, hdl, buddy);
self->quit();
},
others() >> report_unexpected
others() >> CPPA_UNEXPECTED_MSG_CB()
);
}
......
......@@ -367,7 +367,7 @@ template<typename... Ts>
any_tuple make_dynamically_typed(Ts&&... args) {
auto oarr = new detail::object_array;
make_dynamically_typed_impl(*oarr, std::forward<Ts>(args)...);
return any_tuple{oarr};
return any_tuple{static_cast<any_tuple::raw_ptr>(oarr)};
}
void test_wildcards() {
......
......@@ -28,13 +28,11 @@ struct int_visitor {
using dlimits = std::numeric_limits<double>;
struct double_visitor {
double operator()(none_t) const { return dlimits::signaling_NaN(); }
double operator()(void) const { return dlimits::quiet_NaN(); }
double operator()(int i) const { return i; }
double operator()(float f) const { return f; }
double operator()(double d) const { return d; }
double operator()() const { return dlimits::quiet_NaN(); }
template<typename T>
double operator()(T value) const { return value; }
};
int main() {
......
......@@ -25,8 +25,8 @@ void reflector() {
become (
others() >> [=] {
CPPA_LOGF_INFO("reflect and quit");
reply_tuple(self->last_dequeued());
self->quit();
return self->last_dequeued();
}
);
}
......@@ -111,17 +111,17 @@ void spawn5_server(actor_ptr client, bool inverted) {
void spawn5_client() {
CPPA_SET_DEBUG_NAME("spawn5_client");
become (
on(atom("GetGroup")) >> [] {
on(atom("GetGroup")) >> []() -> group_ptr {
CPPA_LOGF_INFO("received {'GetGroup'}");
reply(group::get("local", "foobar"));
return group::get("local", "foobar");
},
on(atom("Spawn5"), arg_match) >> [=](const group_ptr& grp) {
on(atom("Spawn5"), arg_match) >> [=](const group_ptr& grp) -> any_tuple {
CPPA_LOGF_INFO("received {'Spawn5'}");
actor_vector vec;
for (int i = 0; i < 5; ++i) {
vec.push_back(spawn_in_group(grp, reflector));
}
reply(atom("ok"), std::move(vec));
return {atom("ok"), std::move(vec)};
},
on(atom("Spawn5Done")) >> [] {
CPPA_LOGF_INFO("received {'Spawn5Done'}");
......@@ -211,14 +211,15 @@ class client : public event_based_actor {
void test_group_comm_inverted() {
CPPA_PRINT("test group communication via network (inverted setup)");
become (
on(atom("GClient")) >> [=] {
on(atom("GClient")) >> [=]() -> any_tuple {
auto cptr = self->last_sender();
auto s5c = spawn<monitored>(spawn5_client);
reply(atom("GClient"), s5c);
// set next behavior
await_down(s5c, [=] {
CPPA_CHECKPOINT();
self->quit();
});
return {atom("GClient"), s5c};
}
);
}
......@@ -241,18 +242,18 @@ class server : public event_based_actor {
void await_spawn_ping() {
CPPA_PRINT("await {'SpawnPing'}");
become (
on(atom("SpawnPing")) >> [=] {
on(atom("SpawnPing")) >> [=]() -> any_tuple {
CPPA_PRINT("received {'SpawnPing'}");
auto client = self->last_sender();
CPPA_LOGF_ERROR_IF(!client, "last_sender() == nullptr");
CPPA_LOGF_INFO("spawn event-based ping actor");
auto pptr = spawn<monitored>(event_based_ping, num_pings);
reply(atom("PingPtr"), pptr);
CPPA_LOGF_INFO("wait until spawned ping actor is done");
await_down(pptr, [=] {
CPPA_CHECK_EQUAL(pongs(), num_pings);
await_sync_msg();
});
return {atom("PingPtr"), pptr};
}
);
}
......@@ -260,11 +261,11 @@ class server : public event_based_actor {
void await_sync_msg() {
CPPA_PRINT("await {'SyncMsg'}");
become (
on(atom("SyncMsg"), arg_match) >> [=](float f) {
on(atom("SyncMsg"), arg_match) >> [=](float f) -> atom_value {
CPPA_PRINT("received {'SyncMsg', " << f << "}");
CPPA_CHECK_EQUAL(f, 4.2f);
reply(atom("SyncReply"));
await_foobars();
return atom("SyncReply");
}
);
}
......@@ -273,13 +274,13 @@ class server : public event_based_actor {
CPPA_PRINT("await foobars");
auto foobars = make_shared<int>(0);
become (
on(atom("foo"), atom("bar"), arg_match) >> [=](int i) {
on(atom("foo"), atom("bar"), arg_match) >> [=](int i) -> any_tuple {
++*foobars;
reply_tuple(self->last_dequeued());
if (i == 99) {
CPPA_CHECK_EQUAL(*foobars, 100);
test_group_comm();
}
return self->last_dequeued();
}
);
}
......@@ -287,13 +288,13 @@ class server : public event_based_actor {
void test_group_comm() {
CPPA_PRINT("test group communication via network");
become (
on(atom("GClient")) >> [=] {
on(atom("GClient")) >> [=]() -> any_tuple {
auto cptr = self->last_sender();
auto s5c = spawn<monitored>(spawn5_client);
reply(atom("GClient"), s5c);
await_down(s5c, [=] {
test_group_comm_inverted(cptr);
});
return {atom("GClient"), s5c};
}
);
}
......
......@@ -159,7 +159,7 @@ int main() {
oarr->push_back(object::from(static_cast<uint32_t>(42)));
oarr->push_back(object::from("foo" ));
any_tuple atuple1(oarr);
any_tuple atuple1{static_cast<any_tuple::raw_ptr>(oarr)};
try {
auto opt = tuple_cast<uint32_t, string>(atuple1);
CPPA_CHECK(opt.valid());
......
......@@ -34,29 +34,29 @@ class event_testee : public sb_actor<event_testee> {
event_testee() {
wait4string = (
on<string>() >> [=]() {
on<string>() >> [=] {
become(wait4int);
},
on<atom("get_state")>() >> [=]() {
reply("wait4string");
on<atom("get_state")>() >> [=] {
return "wait4string";
}
);
wait4float = (
on<float>() >> [=]() {
on<float>() >> [=] {
become(wait4string);
},
on<atom("get_state")>() >> [=]() {
reply("wait4float");
on<atom("get_state")>() >> [=] {
return "wait4float";
}
);
wait4int = (
on<int>() >> [=]() {
on<int>() >> [=] {
become(wait4float);
},
on<atom("get_state")>() >> [=]() {
reply("wait4int");
on<atom("get_state")>() >> [=] {
return "wait4int";
}
);
}
......@@ -86,10 +86,10 @@ struct chopstick : public sb_actor<chopstick> {
behavior taken_by(actor_ptr whom) {
return (
on<atom("take")>() >> [=]() {
reply(atom("busy"));
on<atom("take")>() >> [=] {
return atom("busy");
},
on(atom("put"), whom) >> [=]() {
on(atom("put"), whom) >> [=] {
become(available);
},
on(atom("break")) >> [=]() {
......@@ -104,11 +104,11 @@ struct chopstick : public sb_actor<chopstick> {
chopstick() {
available = (
on(atom("take"), arg_match) >> [=](actor_ptr whom) {
on(atom("take"), arg_match) >> [=](actor_ptr whom) -> atom_value {
become(taken_by(whom));
reply(atom("taken"));
return atom("taken");
},
on(atom("break")) >> [=]() {
on(atom("break")) >> [=] {
quit();
}
);
......@@ -125,7 +125,7 @@ class testee_actor {
string_received = true;
},
on<atom("get_state")>() >> [&] {
reply("wait4string");
return "wait4string";
}
)
.until(gref(string_received));
......@@ -138,7 +138,7 @@ class testee_actor {
float_received = true;
},
on<atom("get_state")>() >> [&] {
reply("wait4float");
return "wait4float";
}
)
.until(gref(float_received));
......@@ -153,7 +153,7 @@ class testee_actor {
wait4float();
},
on<atom("get_state")>() >> [&] {
reply("wait4int");
return "wait4int";
}
);
}
......@@ -227,10 +227,10 @@ class fixed_stack : public sb_actor<fixed_stack> {
full = (
on(atom("push"), arg_match) >> [=](int) { },
on(atom("pop")) >> [=]() {
reply(atom("ok"), data.back());
on(atom("pop")) >> [=]() -> any_tuple {
data.pop_back();
become(filled);
return {atom("ok"), data.back()};
}
);
......@@ -240,11 +240,12 @@ class fixed_stack : public sb_actor<fixed_stack> {
if (data.size() == max_size)
become(full);
},
on(atom("pop")) >> [=]() {
reply(atom("ok"), data.back());
on(atom("pop")) >> [=]() -> any_tuple {
auto result = data.back();
data.pop_back();
if (data.empty())
become(empty);
return {atom("ok"), result};
}
);
......@@ -253,8 +254,8 @@ class fixed_stack : public sb_actor<fixed_stack> {
data.push_back(what);
become(filled);
},
on(atom("pop")) >> [=]() {
reply(atom("failure"));
on(atom("pop")) >> [=] {
return atom("failure");
}
);
......@@ -264,9 +265,9 @@ class fixed_stack : public sb_actor<fixed_stack> {
void echo_actor() {
become (
others() >> [] {
reply_tuple(self->last_dequeued());
others() >> []() -> any_tuple {
self->quit(exit_reason::normal);
return self->last_dequeued();
}
);
}
......@@ -276,7 +277,7 @@ struct simple_mirror : sb_actor<simple_mirror> {
behavior init_state = (
others() >> [] {
reply_tuple(self->last_dequeued());
return self->last_dequeued();
}
);
......@@ -311,7 +312,30 @@ struct high_priority_testee_class : event_based_actor {
}
};
struct my_request { int a; int b; };
bool operator==(const my_request& lhs, const my_request& rhs) {
return lhs.a == rhs.a && lhs.b == rhs.b;
}
typed_actor_ptr<replies_to<my_request>::with<bool>>
spawn_typed_server() {
return spawn_typed(
on_arg_match >> [](const my_request& req) {
return req.a == req.b;
}
);
}
void test_typed_actors() {
announce<my_request>(&my_request::a, &my_request::b);
auto sptr = spawn_typed_server();
sync_send(sptr, my_request{2, 2}).await(
[](bool value) {
CPPA_CHECK_EQUAL(value, true);
}
);
send_exit(sptr, exit_reason::user_defined);
auto ptr0 = spawn_typed(
on_arg_match >> [](double d) {
return d * d;
......@@ -340,6 +364,14 @@ void test_typed_actors() {
);
}
);
// check async messages
send(ptr0_float, 4.0f);
receive(
on_arg_match >> [](float f) {
CPPA_CHECK_EQUAL(f, 4.0f / 2.0f);
}
);
// check sync messages
sync_send(ptr0_float, 4.0f).await(
[](float f) {
CPPA_CHECK_EQUAL(f, 4.0f / 2.0f);
......@@ -382,14 +414,51 @@ void test_typed_actors() {
CPPA_CHECKPOINT();
}
struct master : event_based_actor {
void init() override {
become(
on(atom("done")) >> []() {
self->quit(exit_reason::user_defined);
}
);
}
};
struct slave : event_based_actor {
slave(actor_ptr master) : master{master} { }
void init() override {
link_to(master);
become (
others() >> CPPA_UNEXPECTED_MSG_CB()
);
}
actor_ptr master;
};
int main() {
CPPA_TEST(test_spawn);
cout << "sizeof(event_based_actor) = " << sizeof(event_based_actor) << endl;
cout << "sizeof(broker) = " << sizeof(io::broker) << endl;
aout << "sizeof(event_based_actor) = " << sizeof(event_based_actor) << endl;
aout << "sizeof(broker) = " << sizeof(io::broker) << endl;
test_typed_actors();
// check whether detached actors and scheduled actors interact w/o errors
auto m = spawn<master, detached>();
spawn<slave>(m);
spawn<slave>(m);
send(m, atom("done"));
await_all_others_done();
CPPA_CHECKPOINT();
CPPA_PRINT("test send()");
send(self, 1, 2, 3, true);
receive(on(1, 2, 3, true) >> [] { });
......@@ -496,8 +565,8 @@ int main() {
auto factory = factory::event_based([&](int* i, float*, string*) {
become (
on(atom("get_int")) >> [i]() {
reply(*i);
on(atom("get_int")) >> [i] {
return *i;
},
on(atom("set_int"), arg_match) >> [i](int new_value) {
*i = new_value;
......@@ -548,7 +617,7 @@ int main() {
}
);
vector<int> expected{9, 8, 7, 6, 5, 4, 3, 2, 1, 0};
CPPA_CHECK(values == expected);
CPPA_CHECK_EQUAL(util::join(values, ","), util::join(values, ","));
}
// terminate st
send_exit(st, exit_reason::user_defined);
......@@ -558,7 +627,7 @@ int main() {
auto sync_testee1 = spawn<blocking_api>([] {
receive (
on(atom("get")) >> [] {
reply(42, 2);
return make_cow_tuple(42, 2);
}
);
});
......@@ -603,8 +672,9 @@ int main() {
on_arg_match >> [&](const string& str) {
CPPA_CHECK(self->last_sender() != nullptr);
CPPA_CHECK_EQUAL(str, "nothing");
reply("goodbye!");
self->quit();
// TODO: should return the value instead
send(self->last_sender(), "goodbye!");
},
after(chrono::minutes(1)) >> [] {
cerr << "PANIC!!!!" << endl;
......@@ -621,11 +691,11 @@ int main() {
self->monitor(sync_testee);
send(sync_testee, "hi");
receive (
on("whassup?") >> [&] {
CPPA_CHECK(true);
on("whassup?") >> [&]() -> std::string {
CPPA_CHECKPOINT();
// this is NOT a reply, it's just an asynchronous message
send(self->last_sender(), "a lot!");
reply("nothing");
return "nothing";
}
);
receive (
......@@ -728,8 +798,8 @@ int main() {
auto f = factory::event_based([](string* name) {
become (
on(atom("get_name")) >> [name]() {
reply(atom("name"), *name);
on(atom("get_name")) >> [name] {
return make_cow_tuple(atom("name"), *name);
}
);
});
......
......@@ -7,15 +7,15 @@ using namespace cppa::placeholders;
struct sync_mirror : sb_actor<sync_mirror> {
behavior init_state = (
others() >> [] { reply_tuple(self->last_dequeued()); }
others() >> [] { return self->last_dequeued(); }
);
};
// replies to 'f' with 0.0f and to 'i' with 0
struct float_or_int : sb_actor<float_or_int> {
behavior init_state = (
on(atom("f")) >> [] { reply(0.0f); },
on(atom("i")) >> [] { reply(0); }
on(atom("f")) >> [] { return 0.0f; },
on(atom("i")) >> [] { return 0; }
);
};
......@@ -76,9 +76,9 @@ struct B : popular_actor {
struct C : sb_actor<C> {
behavior init_state = (
on(atom("gogo")) >> [=] {
reply(atom("gogogo"));
on(atom("gogo")) >> [=]() -> atom_value {
self->quit();
return atom("gogogo");
}
);
};
......@@ -109,6 +109,11 @@ struct D : popular_actor {
reply_to(handle, last_dequeued());
quit();
});
/*/
return sync_send_tuple(buddy(), last_dequeued()).then([=]() -> any_tuple {
quit();
return last_dequeued();
});//*/
}
);
}
......@@ -273,7 +278,7 @@ int main() {
spawn<monitored + blocking_api>([] { // client
auto s = spawn<server, linked>(); // server
auto w = spawn<linked>([] { // worker
become(on(atom("request")) >> []{ reply(atom("response")); });
become(on(atom("request")) >> []{ return atom("response"); });
});
// first 'idle', then 'request'
send_as(w, s, atom("idle"));
......
......@@ -90,10 +90,10 @@ optional<int> str2int(const std::string& str) {
struct dummy_receiver : event_based_actor {
void init() {
become(
on_arg_match >> [=](expensive_copy_struct& ecs) {
on_arg_match >> [=](expensive_copy_struct& ecs) -> expensive_copy_struct {
ecs.value = 42;
reply(std::move(ecs));
quit();
return std::move(ecs);
}
);
}
......
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